springboot乾貨——（二十）整合rocketmq
阿新 • • 發佈：2019-02-17
本篇主要講springboot如何整合rocketmq，具體的rocketmq相關知識可以檢視我的rocketmq相關專欄：RocketMQ菜鳥進化系列
1.專案結構
2.pom.xml
這邊主要是加入了mq的jar包
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.2.6</version>
</dependency>
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.gwd</groupId> <artifactId>springboot-rocketmq</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>springboot-rocketmq</name> <description>Demo project for springboot-rocketmq</description> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.0.0.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>com.alibaba.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>3.2.6</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
3.application.properties
# 消費者的組名
apache.rocketmq.consumer.PushConsumer=PushConsumer
# 生產者的組名
apache.rocketmq.producer.producerGroup=Producer
# NameServer地址
apache.rocketmq.namesrvAddr=XXX.14.250.114:9876
4.訊息生產者RocketMQProvider
以順序傳送訊息為例
package com.gwd.rocketmq; import java.util.List; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.util.StopWatch; import com.alibaba.rocketmq.client.producer.DefaultMQProducer; import com.alibaba.rocketmq.client.producer.MessageQueueSelector; import com.alibaba.rocketmq.client.producer.SendResult; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.common.message.MessageQueue; /** * @FileName RocketMQServer.java * @Description:TODO * @author JackHisen(gu.weidong) * @version V1.0 * @createtime 2018年3月22日 下午5:47:49 * 修改歷史： * 時間 作者 版本 描述 *==================================================== * */ //@Component @Service public class RocketMQProvider { /** * 生產者的組名 */ @Value("${apache.rocketmq.producer.producerGroup}") private String producerGroup; /** * NameServer 地址 */ @Value("${apache.rocketmq.namesrvAddr}") private String namesrvAddr; // @PostConstruct //@PostContruct是spring框架的註解，在方法上加該註解會在專案啟動的時候執行該方法，也可以理解為在spring容器初始化的時候執行該方法。 public void defaultMQProducer() { //生產者的組名 DefaultMQProducer producer = new DefaultMQProducer(producerGroup); //指定NameServer地址，多個地址以 ; 隔開 producer.setNamesrvAddr(namesrvAddr); try { /** * Producer物件在使用之前必須要呼叫start初始化，初始化一次即可 * 注意：切記不可以在每次傳送訊息時，都呼叫start方法 */ producer.start(); //建立一個訊息例項，包含 topic、tag 和 訊息體 //如下：topic 為 "TopicTest"，tag 為 "push" Message message = new Message("TopicTest", "push", "傳送訊息----zhisheng-----".getBytes()); StopWatch stop = new StopWatch(); stop.start(); for (int i = 0; i < 10; i++) { SendResult result = producer.send(message,new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { Integer id = (Integer) arg; int index = id % mqs.size(); return mqs.get(index); } },1); System.out.println("傳送響應：MsgId:" + result.getMsgId() + "，傳送狀態:" + result.getSendStatus()); } stop.stop(); System.out.println("----------------傳送十條訊息耗時：" + stop.getTotalTimeMillis()); } catch (Exception e) { e.printStackTrace(); } finally { producer.shutdown(); } } }
一般訊息是通過輪詢所有佇列來發送的（負載均衡策略），順序訊息可以根據業務，比如說訂單號相同的訊息傳送到同一個佇列，這邊在程式碼中指定了1，1處這個值相同的獲取到的佇列是同一個佇列。
producer.send(message,new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
},1);
5.訊息消費者RocketMQConsumer
package com.gwd.rocketmq;
import javax.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
/**
* @FileName RocketMQClient.java
* @Description:TODO
* @author JackHisen(gu.weidong)
* @version V1.0
* @createtime 2018年3月22日 下午6:58:24
* 修改歷史：
* 時間 作者 版本 描述
*====================================================
*
*/
@Service
public class RocketMQConsumer {
/**
* 消費者的組名
*/
@Value("${apache.rocketmq.consumer.PushConsumer}")
private String consumerGroup;
/**
* NameServer 地址
*/
@Value("${apache.rocketmq.namesrvAddr}")
private String namesrvAddr;
// @PostConstruct //@PostContruct是spring框架的註解，在方法上加該註解會在專案啟動的時候執行該方法，也可以理解為在spring容器初始化的時候執行該方法。
public void defaultMQPushConsumer() {
//消費者的組名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//指定NameServer地址，多個地址以 ; 隔開
consumer.setNamesrvAddr(namesrvAddr);
try {
//訂閱PushTopic下Tag為push的訊息
consumer.subscribe("TopicTest", "push");
//設定Consumer第一次啟動是從佇列頭部開始消費還是佇列尾部開始消費
//如果非第一次啟動，那麼按照上次消費的位置繼續消費
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {//->為Java8的lambda表示式,就是匿名函式,具體可以參考該文章https://segmentfault.com/q/1010000007518474。
try {
for (MessageExt messageExt : list) {
System.out.println("messageExt: " + messageExt);//輸出訊息內容
String messageBody = new String(messageExt.getBody());
System.out.println("消費響應：msgId : " + messageExt.getMsgId() + ", msgBody : " + messageBody);//輸出訊息內容
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍後再試
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消費成功
});
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}
}
6.TestController
package com.gwd.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.gwd.rocketmq.RocketMQProvider;
/**
* @FileName TestController.java
* @Description:TODO
* @author JackHisen(gu.weidong)
* @version V1.0
* @createtime 2018年3月22日 下午7:07:24
* 修改歷史：
* 時間 作者 版本 描述
*====================================================
*
*/
@RestController
public class TestController {
@Autowired
RocketMQProvider rocketMQProvider;
@RequestMapping("/testMQ")
public String testMq() {
rocketMQProvider.defaultMQProducer();
return null;
}
}
測試結果
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D269，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=334, sysFlag=0, bornTimestamp=1521719131994, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133733, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D269, commitLogOffset=49402473, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=335, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D269, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D2FE，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=335, sysFlag=0, bornTimestamp=1521719132002, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133739, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D2FE, commitLogOffset=49402622, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=336, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D2FE, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D393，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=336, sysFlag=0, bornTimestamp=1521719132008, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133745, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D393, commitLogOffset=49402771, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=337, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D393, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D428，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=337, sysFlag=0, bornTimestamp=1521719132014, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133750, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D428, commitLogOffset=49402920, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=338, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D428, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D4BD，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=338, sysFlag=0, bornTimestamp=1521719132019, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133755, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D4BD, commitLogOffset=49403069, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=339, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D4BD, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D552，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=339, sysFlag=0, bornTimestamp=1521719132024, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133761, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D552, commitLogOffset=49403218, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=340, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D552, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D5E7，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=340, sysFlag=0, bornTimestamp=1521719132036, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133773, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D5E7, commitLogOffset=49403367, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=341, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D5E7, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D67C，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=341, sysFlag=0, bornTimestamp=1521719132044, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133781, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D67C, commitLogOffset=49403516, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=342, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D67C, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D711，傳送狀態:SEND_OK
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=342, sysFlag=0, bornTimestamp=1521719132050, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133788, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D711, commitLogOffset=49403665, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=343, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D711, msgBody : 傳送訊息----zhisheng-----
傳送響應：MsgId:6A0EFA7200002AB20000000002F1D7A6，傳送狀態:SEND_OK
----------------傳送十條訊息耗時：79
messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=343, sysFlag=0, bornTimestamp=1521719132057, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133794, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D7A6, commitLogOffset=49403814, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=344, WAIT=true, TAGS=push}, body=29]]
消費響應：msgId : 6A0EFA7200002AB20000000002F1D7A6, msgBody : 傳送訊息----zhisheng-----