start mqnameserv.cmd

start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

DROP TABLE IF EXISTS `spring_cloud_config`;

CREATE TABLE `spring_cloud_config` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `akey` varchar(30) DEFAULT NULL,
  `avalue` varchar(128) DEFAULT NULL,
  `application` varchar(30) DEFAULT NULL,
  `aprofile` varchar(30) DEFAULT NULL,
  `label` varchar(30) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;



INSERT INTO `spring_cloud_config` (`id`, `akey`, `avalue`, `application`, `aprofile`, `label`)
VALUES
    (2,'name_server','127.0.0.1:9876','product-service','dev','dev'),
    (3,'name_server','127.0.0.1:9876','order-service','dev','dev'),
    (4,'order_topic','order_topic','order-service','dev','dev'),
    (5,'order_topic','order_topic','product-service','dev','dev');

package com.hqs.chargeorderservice.mqservice;

import com.alibaba.fastjson.JSONObject;
import com.hqs.chargeorderservice.config.Jms;
import com.hqs.chargeorderservice.service.ProduceOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
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.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;


/**
 *
 * @Description: 分布式事務(wù)RocketMQ 生產(chǎn)者
 */
@Slf4j
@Component
public class TransactionProducer {

    /**
     * 需要自定義事務(wù)監(jiān)聽(tīng)器 用于 事務(wù)的二次確認(rèn) 和 事務(wù)回查
     */
    private TransactionListener transactionListener ;

    /**
     * 這里的生產(chǎn)者和之前的不一樣
     */
    private TransactionMQProducer producer = null;

    /**
     * 官方建議自定義線程 給線程取自定義名稱(chēng) 發(fā)現(xiàn)問(wèn)題更好排查
     */
    private ExecutorService executorService = new ThreadPoolExecutor(2, 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;
        }

    });

    public TransactionProducer(@Autowired Jms jms, @Autowired ProduceOrderService produceOrderService) {
        transactionListener = new TransactionListenerImpl(produceOrderService);
        // 初始化 事務(wù)生產(chǎn)者
        producer = new TransactionMQProducer(jms.getOrderTopic());
        // 添加服務(wù)器地址
        producer.setNamesrvAddr(jms.getNameServer());
        // 添加事務(wù)監(jiān)聽(tīng)器
        producer.setTransactionListener(transactionListener);
        // 添加自定義線程池
        producer.setExecutorService(executorService);

        start();
    }

    public TransactionMQProducer getProducer() {
        return this.producer;
    }

    /**
     * 對(duì)象在使用之前必須要調(diào)用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一般在應(yīng)用上下文，使用上下文監(jiān)聽(tīng)器，進(jìn)行關(guān)閉
     */
    public void shutdown() {
        this.producer.shutdown();
    }
}

/**
 * @Description: 自定義事務(wù)監(jiān)聽(tīng)器
 */
@Slf4j
class TransactionListenerImpl implements TransactionListener {

    @Autowired
    private ProduceOrderService produceOrderService ;

    public TransactionListenerImpl( ProduceOrderService produceOrderService) {
        this.produceOrderService = produceOrderService;
    }

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.info("=========本地事務(wù)開(kāi)始執(zhí)行=============");
        String message = new String(msg.getBody());
        JSONObject jsonObject = JSONObject.parseObject(message);
        Integer productId = jsonObject.getInteger("productId");
        Integer total = jsonObject.getInteger("total");
        Integer tradeNo = jsonObject.getInteger("tradeNo");
        int userId = Integer.parseInt(arg.toString());
        //模擬執(zhí)行本地事務(wù)begin=======
        /**
         * 本地事務(wù)執(zhí)行會(huì)有三種可能
         * 1、commit 成功
         * 2、Rollback 失敗
         * 3、網(wǎng)絡(luò)等原因服務(wù)宕機(jī)收不到返回結(jié)果
         */
        log.info("本地事務(wù)執(zhí)行參數(shù),用戶(hù)id={},商品ID={},銷(xiāo)售庫(kù)存={}",userId,productId,total);
        int result = produceOrderService.save(userId, productId, total, tradeNo);
        //模擬執(zhí)行本地事務(wù)end========
        //TODO 實(shí)際開(kāi)發(fā)下面不需要我們手動(dòng)返回，而是根據(jù)本地事務(wù)執(zhí)行結(jié)果自動(dòng)返回
        //1、二次確認(rèn)消息，然后消費(fèi)者可以消費(fèi)
        if (result == 0) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        //2、回滾消息，Broker端會(huì)刪除半消息
        if (result == 1) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        //3、Broker端會(huì)進(jìn)行回查消息
        if (result == 2) {
            return LocalTransactionState.UNKNOW;
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    /**
     * 只有上面接口返回 LocalTransactionState.UNKNOW 才會(huì)調(diào)用查接口被調(diào)用
     *
     * @param msg 消息
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        log.info("==========回查接口=========");
        String key = msg.getKeys();
        //TODO 1、必須根據(jù)key先去檢查本地事務(wù)消息是否完成。
        /**
         * 因?yàn)橛蟹N情況就是：上面本地事務(wù)執(zhí)行成功了，但是return LocalTransactionState.COMMIT_MESSAG的時(shí)候
         * 服務(wù)掛了，那么最終 Brock還未收到消息的二次確定，還是個(gè)半消息 ，所以當(dāng)重新啟動(dòng)的時(shí)候還是回調(diào)這個(gè)回調(diào)接口。
         * 如果不先查詢(xún)上面本地事務(wù)的執(zhí)行情況 直接在執(zhí)行本地事務(wù)，那么就相當(dāng)于成功執(zhí)行了兩次本地事務(wù)了。
         */
        // TODO 2、這里返回要么commit 要么rollback。沒(méi)有必要在返回 UNKNOW
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}