<dependencies>
    <!-- springboot-web組件 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.0.3</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

@RequestMapping("/sendMsg")
    public String sendMsg() {
        OrderEntity orderEntity = new OrderEntity("123456","騰訊視頻會(huì)員");
        SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);
        System.out.println("返回發(fā)送消息狀態(tài)：" + kaicoTopic);
        return "success";
    }

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag1", consumerGroup = "kaico_consumer", messageModel = MessageModel.CLUSTERING)
public class OrdeConsumer2 implements RocketMQListener<OrderEntity> {
    @Override
    public void onMessage(OrderEntity o) {
        System.out.println("kaico_consumer2消費(fèi)者接收對(duì)象:" + o.toString());
    }
}

 SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);

 //異步發(fā)送
        rocketMQTemplate.asyncSend("kaicoTopic" + ":" + "tag1", orderEntity, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("異步發(fā)送消息成功");
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("異步發(fā)送消息失敗");
            }
        });

//單向(oneway)發(fā)送
        rocketMQTemplate.sendOneWay("kaicoTopic"+":"+"tag1", orderEntity);

@Service
public class NoSpringBootOrderConsumer {
    private DefaultMQPushConsumer defaultMQPushConsumer;
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    protected String consumerGroup;
    protected String topic;
    protected String topicTag;
    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }
    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }
    public void setTopic(String topic) {
        this.topic = topic;
    }
    public void setTopicTag(String topicTag) {
        this.topicTag = topicTag;
    }
    public static String encoding = System.getProperty("file.encoding");
    /*
     * @Author ex_fengkai
     * @Description //TODO 初始化數(shù)據(jù)（消費(fèi)者組名稱、topic、topic的tag、nameServer的信息）
     * @Date 2020/11/9 14:36
     * @Param []
     * @return void
     **/
    private void initParam() {
        this.consumerGroup = "kaico_consumer3";
        this.topic = "kaicoTopic";
        this.topicTag = "tag1";
        this.setNamesrvAddr(namesrvAddr);
    }
    @PostConstruct
    private void init() throws InterruptedException, MQClientException {
        initParam();
        // ConsumerGroupName需要由應(yīng)用來(lái)保證唯一，用于把多個(gè)Consumer組織到一起，提高并發(fā)處理能力
        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr); //設(shè)置nameServer服務(wù)器
        defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        defaultMQPushConsumer.setVipChannelEnabled(false);
        // 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開(kāi)始消費(fèi)
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 訂閱指定Topic下的topicTag
        System.out.println("consumerGroup:" + consumerGroup + " topic:" + topic + " ,topicTag:" + topicTag);
        defaultMQPushConsumer.subscribe(topic, topicTag);
        // 設(shè)置為集群消費(fèi)
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        // 通過(guò)匿名消息監(jiān)聽(tīng)處理消息消費(fèi)
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            // 默認(rèn)msgs里只有一條消息，可以通過(guò)設(shè)置consumeMessageBatchMaxSize參數(shù)來(lái)批量接收消息
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                if (msg.getTopic().equals(topic) && msg.getTags() != null && msg.getTags().equals(topicTag)) {
                    // 執(zhí)行topic下對(duì)應(yīng)tag的消費(fèi)邏輯
                    try {
                        onMessage(new String(msg.getBody(),"utf-8"));

                    } catch (UnsupportedEncodingException e) {
                        System.out.println("系統(tǒng)不支持消息編碼格式:" + encoding);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                    } catch (Exception e) {
                        System.out.println("消息處理異常");
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    System.out.println("consumerGroup:" + consumerGroup + " MsgId:" + msg.getMsgId() + " was done!");
                }
                // 如果沒(méi)有return success ，consumer會(huì)重新消費(fèi)該消息，直到return success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Consumer對(duì)象在使用之前必須要調(diào)用start初始化，初始化一次即可
        defaultMQPushConsumer.start();
        System.out.println("consumerGroup:" + consumerGroup + " namesrvAddr:" + namesrvAddr + "  start success!");
    }
    @PreDestroy
    public void destroy() {
        defaultMQPushConsumer.shutdown();
    }
    private void onMessage(String s) {
        System.out.println(consumerGroup + "用spring的方式的消費(fèi)者消費(fèi)：" + s);
    }
}

//發(fā)送順序消息
    @RequestMapping("/sendMsg1")
    public String sendMsg1() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Long orderId = System.currentTimeMillis();
        String insertSql = getSqlMsg("insert", orderId);
        String updateSql = getSqlMsg("update", orderId);
        String deleteSql = getSqlMsg("delete", orderId);
        Message insertMsg = new Message("kaicoTopic", "tag6", insertSql.getBytes());
        Message updateMsg = new Message("kaicoTopic", "tag6", updateSql.getBytes());
        Message deleteMsg = new Message("kaicoTopic", "tag6", deleteSql.getBytes());
        DefaultMQProducer producer = rocketMQTemplate.getProducer();
        rocketMQTemplate.getProducer().send(insertMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return  mqs.get(0);
                    }
                }, orderId);
        rocketMQTemplate.getProducer().send(updateMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return mqs.get(0);
                    }
                }, orderId);
        rocketMQTemplate.getProducer().send(deleteMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return  mqs.get(0);
                    }
                }, orderId);
        return orderId + "";
    }

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag6", consumerGroup = "kaico_consumer1",
        messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.ORDERLY, consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt msg) {
        System.out.println(Thread.currentThread().getName() + "-kaico_consumer1消費(fèi)者接收對(duì)象:隊(duì)列" + msg.getQueueId()
                + "=消息：" +  new String(msg.getBody()));
    }
}

 public String saveOrder() {
     // 提前生成我們的訂單id
     String orderId = System.currentTimeMillis() + "";
     /**
      * 1.提前生成我們的半消息
      * 2.半消息發(fā)送成功之后，在執(zhí)行我們的本地事務(wù)
      */
     OrderEntity orderEntity = createOrder(orderId);
     String msg = JSONObject.toJSONString(orderEntity);
     MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(msg);
     stringMessageBuilder.setHeader("msg", msg);
     Message message = stringMessageBuilder.build();
     // 該消息不允許被消費(fèi)者消費(fèi)，生產(chǎn)者的事務(wù)邏輯代碼在生產(chǎn)者事務(wù)監(jiān)聽(tīng)類中executeLocalTransaction方法中執(zhí)行。
     rocketMQTemplate.sendMessageInTransaction("kaicoProducer",
             "orderTopic", message, null);
     return orderId;

}

@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "kaicoProducer") //這個(gè)mayiktProducer生產(chǎn)者的事務(wù)管理
public class SyncProducerListener implements RocketMQLocalTransactionListener {
    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private TransationalUtils transationalUtils;
    /**
     * 執(zhí)行我們訂單的事務(wù)
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        //拿到消息
        Object object = headers.get("msg");
        if (object == null) {
            return null;
        }
        String orderMsg = (String) object;
        OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
        TransactionStatus begin = null;
        try {
            begin = transationalUtils.begin();
            int result = orderMapper.addOrder(orderEntity);
            transationalUtils.commit(begin);
            if (result <= 0) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            // 告訴我們的Broke可以消費(fèi)者該消息
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            if (begin != null) {
                transationalUtils.rollback(begin);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        //add.Order
        return null;
    }
    /**
     * 提供給我們的Broker定時(shí)檢查
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        Object object = headers.get("msg");
        if (object == null) {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        String orderMsg = (String) object;
        OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
        String orderId = orderEntity.getOrderId();
        // 直接查詢我們的數(shù)據(jù)庫(kù)
        OrderEntity orderDbEntity = orderMapper.findOrderId(orderId);
        if (orderDbEntity == null) {
            //不確認(rèn)，繼續(xù)重試
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        //提交事務(wù)
        return RocketMQLocalTransactionState.COMMIT;
    }
}

@Service
@RocketMQMessageListener(topic = "orderTopic", consumerGroup = "kaicoTopic")
public class OrdeConsumer implements RocketMQListener<String> {
    @Autowired
    private DispatchMapper dispatchMapper;
    @Override
    public void onMessage(String msg) {
        OrderEntity orderEntity = JSONObject.parseObject(msg, OrderEntity.class);
        String orderId = orderEntity.getOrderId();
        // 模擬userid為=123456
        DispatchEntity dispatchEntity = new DispatchEntity(orderId, 123456L);
        dispatchMapper.insertDistribute(dispatchEntity);
    }
}