springboot-rocketmq-starter starter源碼github地址：https://github.com/Lickey1991/rocketmq-spring-boot-starter
筆者這里fork https://github.com/maihaoche/rocketmq-spring-boot-starter項目，做了一下小改動，一個項目支持多個producer group，如果不需要此修改，可以check此項目。
springboot-demo springboot接入demo源碼 github地址：https://github.com/Lickey1991/rocketmq-springbooot-demo
rocketmq-demo-springMVC springMVC接入demo源碼 github地址：https://github.com/Lickey1991/rocketmq-springmvc-demo
RocketMq版本 4.5
1.spring Boot 項目接入
1.1 maven 添加依賴
<dependency>
<groupId>com.lickey</groupId>
<artifactId>spring-boot-starter-rocketmq</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
1.2 添加rocketMq配置
rocketMq NameSvrAddr spring.rocketmq.name-server-address: 10.17.0.171:9876;10.17.0.172:9876
# 可選, 如果無需發(fā)送消息則忽略該配置
spring.rocketmq.producer-group: ${rocketmq.groupid.demo}
# 發(fā)送超時配置毫秒數(shù), 可選, 默認3000
spring.rocketmq.send-msg-timeout: 5000
# 追溯消息具體消費情況的開關，默認打開
#trace-enabled: false
2. Sping MVC 項目接入
2.1 maven 添加依賴
<dependencies>
<dependency>
<groupId>com.lickey</groupId>
<artifactId>spring-boot-starter-rocketmq</artifactId>
<version>1.0.0-SNAPSHOT</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
2.2 添加rocketMq配置
添加rocketmq.properties 文件，配置內(nèi)容與springboot一致
2.3 springMvc 配置@Value讀取properties
- application-context.xml配置
<!--加載 RocketMq starter autoConfiguration-->
<context:component-scan base-package="com.lickey.starter.rocketmq.config"/>
<!-- 加載配置屬性文件 -->
<context:property-placeholder ignore-unresolvable="true" location="classpath:rocketmq.properties" />
2.spring-mvc.xml或者dispatcher-servlet.xml 配置
<!-- 加載配置屬性文件 -->
<context:property-placeholder ignore-unresolvable="true" location="classpath:rocketmq.properties" />
因為controller配置在spring-mvc.xml/dispatcher-servlet.xml 中，所以想要在controller中使用@value("${key}")讀取properties 需要在spring-mvc.xml/dispatcher-servlet.xml 也定義properties配置
RocketMq配置中 name-sercer-address可以在 rocketMq的控制臺中查看，daily控制臺中查看。 多個地址，中間使用分號【;】進行分隔
3. 創(chuàng)建producer
3.1 普通producer
- 創(chuàng)建`producer
參考demo中RocketMqProducer，如果需要在消息發(fā)送完成后，做統(tǒng)一處理邏輯例如記錄日志等，需要重寫doAfterSyncSend方法
使用@MQProducer注解創(chuàng)建普通producer，producerGroup為生產(chǎn)者組，如果此處不指定生產(chǎn)這組，則取配置文件中spring.rocketmq.producer-group值為默認生產(chǎn)者組，二者必須有個值不能為空，否則創(chuàng)建失敗，需要繼承 AbstractMQProducer 類
2.發(fā)送同步消息
參考TestServiceImpl#sendMqMessqge，調(diào)用producer的syncSend (Message message)
使用MessageBuilder構建消息體Message，通過靜態(tài)of方法，指定message的topic與tag，或者指定messageBody信息。同時提供topic，tag，massageBody對相應的set方法。
setKey設置rocketMq消息中的key，消息發(fā)送后在rocketMq控制臺查看detail信息如下
3.發(fā)送異步消息
參考demoTestServiceImpl#asyncSend，調(diào)用producer的asyncSend (Message message, SendCallback sendCallback)方法
sendCallBack為發(fā)送后的回調(diào)，實現(xiàn)方式參考DemoSendCallback，實現(xiàn)SendCallBack接口，重寫onSuccess和onException方法
4.發(fā)送順序消息
調(diào)用producer的syncSendOrderly(Message message, String hashKey)方法，該方式使用rocketMq順序消息的hashKey計算消息順序方法
3.2創(chuàng)建事物producer
1.創(chuàng)建producer
參考demo中的RocketMqTransactionProducer,使用@MQTransactionProducer注解創(chuàng)建事物producer，producerGroup為生產(chǎn)者組，必須有值不能為空且不能與其他producer組相同，否則創(chuàng)建失敗，需要繼承 AbstractMQTransactionProducer類，實現(xiàn)executeLocalTransaction、checkLocalTransaction方法(checkLocalTransation方法作用本地事物執(zhí)行時間過長 或者集群收到producer傳過來的狀態(tài)是unknow，集群通過查詢事物消息TOPIC，回調(diào)check)，該功能在rocketMq 3.6版本中去除?；卣{(diào)的實現(xiàn)通過ClientRemotingProcessor#processRequest。 所以checkLocalTransaction 無需寫實際方法體。想要測試，在executeLocalTransaction中返回UNKNOW，processRequest處斷點觀察?；卣{(diào)觸發(fā)，有興趣可以看下相關文章
2.發(fā)送事物消息
參考demoTestServiceImpl#transactionSend，調(diào)用producer的sendMessageInTransaction(Message msg, Object arg)方法
4.創(chuàng)建消費者
參考demo中DemoTopicConsumer，使用@MQConsumer注解創(chuàng)建事物，consumerGroup為消費者組，topic為監(jiān)聽的topic，重寫process方法，方法參數(shù)message為反序列化后的messageBody實例，extMap key包括MessageExtConst中 MessageExt、Message.property兩部分
MessageExtConst中extMap包含的key
/** 來自 MessageExt */
public static final String PROPERTY_EXT_QUEUE_ID = "QUEUE_ID";
public static final String PROPERTY_EXT_STORE_SIZE = "STORE_SIZE";
public static final String PROPERTY_EXT_QUEUE_OFFSET = "QUEUE_OFFSET";
public static final String PROPERTY_EXT_SYS_FLAG = "SYS_FLAG";
public static final String PROPERTY_EXT_BORN_TIMESTAMP = "BORN_TIMESTAMP";
public static final String PROPERTY_EXT_BORN_HOST = "BORN_HOST";
public static final String PROPERTY_EXT_STORE_TIMESTAMP = "STORE_TIMESTAMP";
public static final String PROPERTY_EXT_STORE_HOST = "STORE_HOST";
public static final String PROPERTY_EXT_MSG_ID = "MSG_ID";
public static final String PROPERTY_EXT_COMMIT_LOG_OFFSET = "COMMIT_LOG_OFFSET";
public static final String PROPERTY_EXT_RECONSUME_TIMES = "RECONSUME_TIMES";
public static final String PROPERTY_EXT_PREPARED_TRANSACTION_OFFSET = "PREPARED_TRANSACTION_OFFSET";
public static final String PROPERTY_EXT_BODY_CRC = "BODY_CRC";
/** 以下屬性來自 Message.property */
public static final String PROPERTY_KEYS = "KEYS";
public static final String PROPERTY_TAGS = "TAGS";
public static final String PROPERTY_WAIT_STORE_MSG_OK = "WAIT";
public static final String PROPERTY_DELAY_TIME_LEVEL = "DELAY";
public static final String PROPERTY_RETRY_TOPIC = "RETRY_TOPIC";
public static final String PROPERTY_REAL_TOPIC = "REAL_TOPIC";
public static final String PROPERTY_REAL_QUEUE_ID = "REAL_QID";
public static final String PROPERTY_TRANSACTION_PREPARED = "TRAN_MSG";
public static final String PROPERTY_PRODUCER_GROUP = "PGROUP";
public static final String PROPERTY_MIN_OFFSET = "MIN_OFFSET";
public static final String PROPERTY_MAX_OFFSET = "MAX_OFFSET";
public static final String PROPERTY_BUYER_ID = "BUYER_ID";
public static final String PROPERTY_ORIGIN_MESSAGE_ID = "ORIGIN_MESSAGE_ID";
public static final String PROPERTY_TRANSFER_FLAG = "TRANSFER_FLAG";
public static final String PROPERTY_CORRECTION_FLAG = "CORRECTION_FLAG";
public static final String PROPERTY_MQ2_FLAG = "MQ2_FLAG";
public static final String PROPERTY_RECONSUME_TIME = "RECONSUME_TIME";
public static final String PROPERTY_MSG_REGION = "MSG_REGION";
public static final String PROPERTY_TRACE_SWITCH = "TRACE_ON";
public static final String PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX = "UNIQ_KEY";
public static final String PROPERTY_MAX_RECONSUME_TIMES = "MAX_RECONSUME_TIMES";
public static final String PROPERTY_CONSUME_START_TIMESTAMP = "CONSUME_START_TIME";
以上是接入starter Demo
