RocketMQ與MYSQL事務訊息整合
1、基礎理論知識篇“兩階段提交”如果你瞭解可以跳過這段，當然如果你想深入瞭解你可以購買相關書籍或去搜索相關資料閱讀
兩階段提交分為 正常提交和異常提交或異常回滾
上面是正常提交的示意圖，協調者發起預提交請求，參與者回覆成功之後協調者再次發起commit請求，統一提交事物。事物結束。
如果這兩階段提交過程當中有任何一個請求出現異常就會回滾，如下流程：
異常請求包括預提交 返回預提交的應答，commit請求 等任何一個失敗都會導致整個事物回滾。
二階段提交的問題
“二階段提交”還有一個很嚴重的問題就是如果commit過程當中失敗了 就導致了全部事物失敗，代價很大，簡單粗暴的處理方式
還有一個問題是如果 commit過程中網路出現問題 commit沒有被整個事物的參與者之一或者多個收到，這個時候就會出現資料不一致現象。
可能大家會提到 協調者是誰，參與者又是誰那？
這裡簡單說下自己的理解
如果在你的應用程式中你是通過 begin等相關操作語句開始的，比如 你使用了spring的@Transactional註解等，
那協調者就是你的“應用程式”，參與者就是 mysql或其他支援事物的資料庫系統。
如果你就直接向mysql傳送了一條sql語句mysql是自行提交的，那協調者和參與者都是mysql資料庫自己
2、這裡說下mysql對所謂的“重複資料”提供的相關sql或關鍵字。
unique 唯一主鍵約束
在sql事物中和應用程式中都可以捕獲這個錯誤碼或異常，可以作為冪等判斷的一個依據。
upset 操作，發現唯一主鍵衝突然後更新相關資料，mongodb有直接使用的sql方法語句
示例：insert into tablename(column1) values(123) on duplicate key update column1 =column1 +123
ignore 忽略操作對於多餘的操作直接忽略
insert ignore into tablename(column1) values(123)
基礎篇說完很多內容如果想深入瞭解可以自己找資料處理。下面是華麗分割線
3、在我們原有的認知裡有一個方案就差那麼一點點就可以大面積使用的。
我們之前可能想過怎樣既能傳送mq又能寫資料庫，下面這個方案會分接近我們的願望。
我們遵從如下步驟進行程式碼處理：
1、開啟資料庫事物執行相關sql
2、傳送MQ訊息
3、提交資料庫事物
（注意：以上每一步都是在上一步執行成功之後在執行下一步的）
根據步驟我畫出了下面的流程圖
其實這個流程是有一個漏洞的，如果我把上面的流程圖改造為下面的二階段提交的示意圖就會很明顯的看出來
不知道大家有麼有發現問題，是不是 各種提交和回滾操作都是針對的資料庫，而不是MQ。commit資料庫事物出現異常就會造成資料不一致現象。
其實也不用在想有沒有其他的流程方案能解決分散式雙寫問題，只要存在多寫問題就存在資料不一致問題的現象，
所以就出現了3pc Paxos 等協議來解決分散式事物/一致性的問題。
下面我們開始介紹怎麼使用mysql和RocketMQ來實現事物問題
華麗分割線
4、RocketMQ事物訊息的過程
1、傳送MQ的事物訊息
2、事物訊息傳送成功後會同步觸發對應執行本地介面來執行鍼對mysql資料庫的操作
3、如果有未commit的訊息，RocketMQ 的 broker會定時間隔時間來回查資料庫事物是否已經提交完成
5、結合RocketMQ的事物訊息與Mysql資料庫事物的實現思想
如果上面的二階段提交你已經理解了，你會發現我這裡設計的流程（上面圖的流程）有點不太一樣的地方
什麼地方那？
MQ事物訊息回滾的時候是因為mysql資料庫事物沒有提交成功而導致的，也就是說如果mysql資料庫事務成功了MQ的事務訊息是一定要成功的
否則就會出現事物不一致的現象。
假如傳送MQ的prepare訊息成功了，執行mysql事物的操作也成功了，但是偏偏返回給MQ的commit訊息丟失了，那這個時候資料庫訊息並不會回滾。
所以就有了回查本地事物訊息是否成功的操作，來對MQ的訊息做個補償動作實現資料一致性
理解了二階段提交以及RocketMQ的事物實現之後你就可以自己設計事物相關操作的執行順序了
（這裡的流程設計以及包括我的程式碼實現是以我的理解做出的最佳實踐）
6、RocketMQ與Mysql事物結合注意事項
1、如果應用程式承擔協調者的工作就儘量晚開啟事物和儘量早的提交資料庫事物，事物中的sql對資料競爭多的sql儘量靠後
因為執行資料庫事物會有各種鎖操作，減少鎖的生命週期，資料庫是稀缺資源，大家能省則省
2、資料庫事物最好設定超時時間，超時之後自動解除，最好不超過1分鐘
3、MQ預設1分鐘之後回查一次已傳送message但未commit的訊息，最多回查15次，之後會執行回滾操作
4、應用程式一定要做好冪等處理（可以參考上面mysql相關語句實現冪等介面）
5、網路不要太差，否則會造成大量的重試，重試就會影響訊息的及時性
6、適用場景
單次請求數量小
每次請求會有資料產生，而不是查詢產生的資料（比如 insert操作叫生產資料，select操作不要生產資料）
下游可以接受一定的延遲（這裡有兩個因素，有應用程式本身和Broker，這裡指broker）
下游服務或系統以接收到的訊息為依據做響應的操作
MQ訊息作為主要資訊進行傳遞
下面說下具體程式碼實現
華麗分割線
7、實戰程式碼解析
首先附上原始碼地址 https://github.com/zygfengyuwuzu/springboot-rocketmq-example
下面將針對關鍵程式碼進行講解
首先介紹一下程式碼目錄
瞭解了上面的程式碼目錄下面說下程式碼的執行流程
首先看事物訊息生產者的例項物件建立
package rocketmq_example.mqandmysqltraction.producer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 生產者和消費者測試的時候記得注掉一中的一個以免觀察不出效果
*
*/
@Component
public class TransactionProducer {
static Logger logger = LoggerFactory.getLogger(TransactionProducer.class);
public DefaultMQProducer producer = null;
@Autowired
TransactionListener transactionListenerImp;
@PostConstruct
private void init() throws MQClientException {
logger.info("MQ事物生產者初始化開始--------------------------------------------------");
TransactionMQProducer transactionProducer = new TransactionMQProducer("mytestgroup");
// Producer 組名， 多個 Producer 如果屬於一 個應用，傳送同樣的訊息，則應該將它們 歸為同一組
//transactionProducer.setProducerGroup("mytestgroup");
// Name Server 地址列表
transactionProducer.setNamesrvAddr("10.10.6.71:9876;10.10.6.72:9876");
// 超時時間 這裡一定要大於資料庫事物執行的超時時間
transactionProducer.setSendMsgTimeout(90000);
//這個執行緒池作用就是 mqbroker端回撥資訊的本地處理執行緒池
ExecutorService executorService = new ThreadPoolExecutor(1, 5, 100, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
transactionProducer.setExecutorService(executorService);
transactionProducer.setTransactionListener(transactionListenerImp);
producer = transactionProducer;
producer.start();
logger.info("MQ事物生產者初始化結束--------------------------------------------------");
}
public SendResult send(Message me) throws Exception {
return producer.send(me);
}
/**
* 傳送普通訊息
* @param Topic
* @param Tags
* @param body
* @return
* @throws Exception
*/
public SendResult send(String Topic, String Tags, String body) throws Exception {
Message me = new Message();
// 標示
me.setTopic(Topic);
// 標籤
me.setTags(Tags);
// 內容
me.setBody(body.getBytes(RemotingHelper.DEFAULT_CHARSET));
return producer.send(me);
}
/**
* 傳送普通訊息
* @param Topic
* @param Tags
* @param key
* @param body
* @return
* @throws Exception
*/
public SendResult send(String Topic, String Tags, String key, String body) throws Exception {
try {
Message me = new Message(Topic, Tags, key, 0, body.getBytes(RemotingHelper.DEFAULT_CHARSET), true);
return producer.send(me);
} catch (Exception e) {
logger.error("傳送MQ資訊異常Topic{},Tags{},key{},body{}", Topic, Tags, key, body);
throw e;
}
}
@PreDestroy
public void Destroy() {
producer.shutdown();
}
}
上面的程式碼我們接收到請求傳輸過來的資料之後，首先做了MQ訊息物件的建立，建立成功之後直接傳送MQ事物訊息
事物訊息傳送成功之後會呼叫上面設定的介面實現類的TransactionListenerImpl.executeLocalTransaction()這個方法。
介面實現的方法程式碼如下：
package rocketmq_example.mqandmysqltraction.producer;
import java.util.List;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import rocketmq_example.mqandmysqltraction.MyTableModel;
import rocketmq_example.mqandmysqltraction.MytableService;
/**
* 把資料庫事物巢狀在mq事物當中不能顯示丟擲異常
*
*
*
*
* @author zyg
*
*/
@Component
public class TransactionListenerImpl implements TransactionListener {
static Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);
@Autowired
MytableService mytableService;
/**
* 一定要設定執行sql時間，儘量不要超時
*
*/
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
logger.info("開始執行本地資料庫事物 transactionid:{}", msg.getTransactionId());
LocalTransactionState lts = LocalTransactionState.UNKNOW;
@SuppressWarnings("unchecked")
List<MyTableModel> mytablelist = (List<MyTableModel>) arg;
try {
long start=System.currentTimeMillis();
//資料庫事物執行時間不要超過mq回查時間 預設15分鐘
mytableService.execMytableinsert2(mytablelist, msg.getTransactionId());
logger.info("執行資料庫事物耗時:{}",System.currentTimeMillis()-start);
lts = LocalTransactionState.COMMIT_MESSAGE;
} catch (Exception e) {
logger.error("資料庫事務異常", e);
lts = LocalTransactionState.ROLLBACK_MESSAGE;
}
logger.info("結束執行本地資料庫事物 transactionid:{} 返回:{}", msg.getTransactionId(),lts);
return lts;
}
/**
* 去資料庫查詢看看是否存在已經成功傳送預提交資料而沒有commit成功的mq資訊
* 每分鐘1次預設15次
*
* 這裡可以做個計數 讓MQ重試5次/5分鐘就回滾減輕MQ回查的壓力
*
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
if (mytableService.existMyTableModelByMsgid(msg.getTransactionId())) {
logger.info("查詢到已提交事物 transactionid:{}",msg.getTransactionId());
return LocalTransactionState.COMMIT_MESSAGE;
} else {
logger.info("未查到已提交事物 transactionid:{}",msg.getTransactionId());
return LocalTransactionState.UNKNOW;
}
}
}
上面程式碼有兩個方法，這裡說下兩個方法的作用和執行時間
executeLocalTransaction這個方法是傳送完 事物訊息 之後同步被呼叫到的方法，用來執行本地事物操作
executeLocalTransaction方法有兩個引數，第一個是傳送成功之後的message訊息，在這個方法中包含事物ID其實就是msgid
第二個引數是object型別的是從dataapi傳過來，
我的程式碼中沒做任何處理直接傳遞過來了然後直接轉化傳遞給了service層進行事物處理
這個executeLocalTransaction方法裡面為什麼要直接返回commit或rollback，
目的是儘量快的告訴MQ我的資料庫事務執行成功了，
儘快將half訊息轉為正常訊息，已備消費者消費到做業務處理。
這裡完全可以直接返回unknow，等待broker回查來實現commit操作的。但是這樣做對回查訊息broker造成一定的壓力。
上面程式碼的第二個方法是提供給broker回撥執行的，進行檢查本地事務是否成功執行的操作，發起方是broker
這裡面我們接收到broker的回查請求之後直接去資料庫查詢是否存在broker提供的事務ID的資料
如果存在返回commit標識，如果不存在返回unknow標識以等待下一次再來回查
到此我們的一個事務操作就算完成了
另外大家可以直接檢視service層的實現程式碼，就不一一解釋了
package rocketmq_example.mqandmysqltraction;
import java.util.ArrayList;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.fasterxml.jackson.databind.ObjectMapper;
@Service
public class MytableService {
static Logger logger = LoggerFactory.getLogger(MytableService.class);
@Autowired
IMytableMapper mytable;
@Autowired
ObjectMapper objMapper;
/**
* 這裡可以顯示提交事物 返回boolean 一條一條插入只是為了展現事物的特性 獲取所有異常 處理你的業務邏輯等等
*
* @param mytablemodels
* @return
*/
@Transactional(rollbackFor = Exception.class, timeout = 60000)
public List<Integer> execMytableinsert2(List<MyTableModel> mytablemodels, String msgid) {
// logger.info("開始執行資料庫事物");
List<Integer> result = new ArrayList<Integer>();
for (MyTableModel myTableModel : mytablemodels) {
// 插入資料庫
myTableModel.setMsgid(msgid);
mytable.insertmytable(myTableModel);
result.add(myTableModel.getId());
}
// logger.info("結束執行資料庫事物");
return result;
}
public boolean existMyTableModelById(Integer id) {
MyTableModel myTableModel = mytable.selectMyTableModelById(id);
if (myTableModel != null && null != myTableModel.getId()) {
return true;
}
return false;
}
/**
* 查詢是否存在已經發送過的msgid訊息
*
* @param msgid
* @return
*/
public boolean existMyTableModelByMsgid(String msgid) {
int count = mytable.selectMyTableModelByMsgid(msgid);
if (count > 0) {
return true;
}
return false;
}
public void insetmsg(MyTableModel mytablemodel) {
try {
mytable.insertmsgrecord(mytablemodel);
} catch (org.springframework.dao.DuplicateKeyException e) {
logger.error("主鍵衝突異常被捕獲",e);
}
}
}
非常感謝你能看到這裡！！！看到這裡相信你已經對本篇部落格的內容有所瞭解了！如果有什麼問題或者想不通的地方歡迎評論區進行討論。
如果有不正確的地方懇請