<!-- SpringBoot 整合 RocketMQ 官方依賴 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

# RocketMQ 配置
rocketmq:
  # NameServer 集群地址
  name-server: 127.0.0.1:9876
  # 生產(chǎn)者配置
  producer:
    # 生產(chǎn)者組名
    group: demo-producer-group
    # 消息發(fā)送超時(shí)時(shí)間
    send-message-timeout: 3000
    # 最大重試次數(shù)
    retry-times-when-send-failed: 2
  # 消費(fèi)者默認(rèn)配置
  consumer:
    group: demo-consumer-group

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class MqProducer {
    @Autowired
    private RocketMQTemplate rocketMqTemplate;
    // 定義全局Topic
    private static final String TOPIC = "demo-normal-topic";
    /**
     * 同步發(fā)送消息（可靠，適合核心業(yè)務(wù)）
     */
    public void sendSyncMsg(String msg) {
        SendResult sendResult = rocketMqTemplate.syncSend(TOPIC, msg);
        System.out.println("同步發(fā)送結(jié)果：" + sendResult.getSendStatus());
    }
    /**
     * 異步發(fā)送消息（高吞吐，適合非核心業(yè)務(wù)）
     */
    public void sendAsyncMsg(String msg) {
        rocketMqTemplate.asyncSend(TOPIC, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("異步發(fā)送成功");
            }
            @Override
            public void onException(Throwable e) {
                System.err.println("異步發(fā)送失?。? + e.getMessage());
            }
        });
    }
    /**
     * 單向發(fā)送（極致高性能，無(wú)需響應(yīng)）
     */
    public void sendOneWayMsg(String msg) {
        rocketMqTemplate.sendOneWay(TOPIC, msg);
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(
        topic = "demo-normal-topic",
        consumerGroup = "demo-consumer-group"
)
public class NormalMsgConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        // 執(zhí)行業(yè)務(wù)邏輯
        System.out.println("收到普通消息：" + message);
        // 異常自動(dòng)進(jìn)入重試隊(duì)列，無(wú)需手動(dòng)處理
        // int a = 1 / 0;
    }
}

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(
        topic = "demo-broadcast-topic",
        consumerGroup = "demo-broadcast-group",
        consumeMode = ConsumeMode.BROADCASTING // 開啟廣播消費(fèi)
)
public class BroadcastConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("廣播消費(fèi)消息：" + message);
    }
}

public void sendOrderMsg(String bizKey, String msg) {
    // bizKey相同 → 固定發(fā)送同一隊(duì)列 → 保證順序
    rocketMqTemplate.syncSendOrderly("demo-order-topic", msg, bizKey);
}

import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(
        topic = "demo-order-topic",
        consumerGroup = "demo-order-group",
        messageModel = MessageModel.CLUSTERING
)
public class OrderMsgConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("順序消費(fèi)：" + message);
    }
}

/**
 * 發(fā)送延遲消息
 * 延遲等級(jí)：1~18級(jí) → 1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h
 */
public void sendDelayMsg(String msg) {
    // 3級(jí)延遲 = 10秒后消費(fèi)
    rocketMqTemplate.syncSend("demo-delay-topic", msg, 3);
}

import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;
public void sendBatchMsg() {
    List<Message> messageList = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        Message message = new Message("demo-batch-topic", ("批量消息" + i).getBytes());
        messageList.add(message);
    }
    // 批量發(fā)送
    rocketMqTemplate.syncSend(messageList);
}

// 發(fā)送訂單消息，tag = ORDER
public void sendTagMsg() {
    rocketMqTemplate.syncSend("demo-tag-topic:ORDER", "訂單創(chuàng)建消息");
}

@RocketMQMessageListener(
        topic = "demo-tag-topic",
        consumerGroup = "demo-tag-group",
        selectorExpression = "ORDER||PAY" // 只消費(fèi)ORDER、PAY標(biāo)簽消息
)
@Component
public class TagFilterConsumer implements RocketMQListener<String>{
    @Override
    public void onMessage(String s) {
        System.out.println("過濾消費(fèi)消息：" + s);
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@Component
@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener {
    @Autowired
    private RocketMQTemplate rocketMqTemplate;
    private static final String TRANS_TOPIC = "demo-trans-topic";
    // 發(fā)送事務(wù)消息入口
    public void sendTransMsg(String content) {
        Message<String> message = MessageBuilder.withPayload(content).build();
        rocketMqTemplate.sendMessageInTransaction(TRANS_TOPIC, message, content);
    }
    // 執(zhí)行本地事務(wù)
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        try {
            // 模擬本地?cái)?shù)據(jù)庫(kù)事務(wù)
            System.out.println("執(zhí)行本地事務(wù)：" + arg);
            // 成功則提交消息
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            // 異?；貪L消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
    // 事務(wù)回查
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 查詢數(shù)據(jù)庫(kù)事務(wù)狀態(tài)，此處模擬成功
        return RocketMQLocalTransactionState.COMMIT;
    }
}