1:基礎知識梳理
Message對象
- topic: 主題名稱
- tag: 標簽，用於過濾
- key: 消息唯一標示，可以是業務字段組合
- body: 消息體,字節數組
注意 發送消息到Broker，需要判斷是否有此topic（在broker中），本地環境建議開啟自動創建topic，生產環境建議關閉自動化創建topic
建議先手工創建Topic，如果靠程序自動創建，然后再投遞消息，會出現延遲情況
概念模型: 一個topic下面對應多個queue,可以在創建Topic時指定，如訂單類topic
通過可視化管理后台查看消息
2:整合流程
1.pom.xml添加RocketMq依賴
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.2.6</version>
</dependency>
2.rocketmq.properties中添加如下信息
###producer
#該應用是否啟用生產者
rocketmq.producer.isOnOff=on
#發送同一類消息的設置為同一個group，保證唯一,默認不需要設置，rocketmq會使用ip@pid(pid代表jvm名字)作為唯一標示
rocketmq.producer.groupName=${spring.application.name}
#mq的nameserver地址
rocketmq.producer.namesrvAddr=127.0.0.1:9876
#消息最大長度 默認1024*4(4M)
rocketmq.producer.maxMessageSize=4096
#發送消息超時時間,默認3000
rocketmq.producer.sendMsgTimeout=3000
#發送消息失敗重試次數，默認2
rocketmq.producer.retryTimesWhenSendFailed=2
###consumer
##該應用是否啟用消費者
rocketmq.consumer.isOnOff=on
rocketmq.consumer.groupName=${spring.application.name}
#mq的nameserver地址
rocketmq.consumer.namesrvAddr=127.0.0.1:9876
#該消費者訂閱的主題和tags("*"號表示訂閱該主題下所有的tags),格式：topic~tag1||tag2||tag3;topic2~*;
rocketmq.consumer.topics=DemoTopic~*;
rocketmq.consumer.consumeThreadMin=20
rocketmq.consumer.consumeThreadMax=64
#設置一次消費消息的條數，默認為1條
rocketmq.consumer.consumeMessageBatchMaxSize=1
3.生產者Bean配置
package com.clouds.common.rocketmq.producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.util.StringUtils;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.clouds.common.rocketmq.constants.RocketMQErrorEnum;
import com.clouds.common.rocketmq.exception.RocketMQException;
/**
* 生產者配置
* .<br/>
*
* Copyright: Copyright (c) 2017 zteits
*
* @ClassName: MQProducerConfiguration
* @Description:
* @version: v1.0.0
*/
@Configuration
@PropertySource(value = "classpath:rocketmq.properties")
public class MQProducerConfiguration {
public static final Logger LOGGER = LoggerFactory.getLogger(MQProducerConfiguration.class);
/**
* 發送同一類消息的設置為同一個group，保證唯一,默認不需要設置，rocketmq會使用ip@pid(pid代表jvm名字)作為唯一標示
*/
@Value("${rocketmq.producer.groupName}")
private String groupName;
@Value("${rocketmq.producer.namesrvAddr}")
private String namesrvAddr;
/**
* 消息最大大小，默認4M
*/
@Value("${rocketmq.producer.maxMessageSize}")
private Integer maxMessageSize ;
/**
* 消息發送超時時間，默認3秒
*/
@Value("${rocketmq.producer.sendMsgTimeout}")
private Integer sendMsgTimeout;
/**
* 消息發送失敗重試次數，默認2次
*/
@Value("${rocketmq.producer.retryTimesWhenSendFailed}")
private Integer retryTimesWhenSendFailed;
@Bean
public DefaultMQProducer getRocketMQProducer() throws RocketMQException {
if (StringUtils.isEmpty(this.groupName)) {
throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"groupName is blank",false);
}
if (StringUtils.isEmpty(this.namesrvAddr)) {
throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"nameServerAddr is blank",false);
}
DefaultMQProducer producer;
producer = new DefaultMQProducer(this.groupName);
producer.setNamesrvAddr(this.namesrvAddr);
//如果需要同一個jvm中不同的producer往不同的mq集群發送消息，需要設置不同的instanceName
//producer.setInstanceName(instanceName);
if(this.maxMessageSize!=null){
producer.setMaxMessageSize(this.maxMessageSize);
}
if(this.sendMsgTimeout!=null){
producer.setSendMsgTimeout(this.sendMsgTimeout);
}
//如果發送消息失敗，設置重試次數，默認為2次
if(this.retryTimesWhenSendFailed!=null){
producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
}
try {
producer.start();
LOGGER.info(String.format("producer is start ! groupName:[%s],namesrvAddr:[%s]"
, this.groupName, this.namesrvAddr));
} catch (MQClientException e) {
LOGGER.error(String.format("producer is error {}"
, e.getMessage(),e));
throw new RocketMQException(e);
}
return producer;
}
}
4.消費者Bean配置
package com.clouds.common.rocketmq.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.util.StringUtils;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.clouds.common.rocketmq.constants.RocketMQErrorEnum;
import com.clouds.common.rocketmq.consumer.processor.MQConsumeMsgListenerProcessor;
import com.clouds.common.rocketmq.exception.RocketMQException;
/**
* 消費者Bean配置
* .<br/>
*
* Copyright: Copyright (c) 2017 zteits
*
* @ClassName: MQConsumerConfiguration
* @Description:
* @version: v1.0.0
*/
@Configuration
@PropertySource(value = "classpath:rocketmq.properties")
public class MQConsumerConfiguration {
public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfiguration.class);
@Value("${rocketmq.consumer.namesrvAddr}")
private String namesrvAddr;
@Value("${rocketmq.consumer.groupName}")
private String groupName;
@Value("${rocketmq.consumer.consumeThreadMin}")
private int consumeThreadMin;
@Value("${rocketmq.consumer.consumeThreadMax}")
private int consumeThreadMax;
@Value("${rocketmq.consumer.topics}")
private String topics;
@Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
private int consumeMessageBatchMaxSize;
@Autowired
private MQConsumeMsgListenerProcessor mqMessageListenerProcessor;
@Bean
public DefaultMQPushConsumer getRocketMQConsumer() throws RocketMQException {
if (StringUtils.isEmpty(groupName)){
throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"groupName is null !!!",false);
}
if (StringUtils.isEmpty(namesrvAddr)){
throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"namesrvAddr is null !!!",false);
}
if(StringUtils.isEmpty(topics)){
throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"topics is null !!!",false);
}
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
//集群消費模式
consumer.setMessageModel(MessageModel.CLUSTERING);
consumer.setNamesrvAddr(namesrvAddr);
consumer.setConsumeThreadMin(consumeThreadMin);
consumer.setConsumeThreadMax(consumeThreadMax);
consumer.registerMessageListener(mqMessageListenerProcessor);
/**
* 設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費
* 如果非第一次啟動，那么按照上次消費的位置繼續消費
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
/**
* 設置消費模型，集群還是廣播，默認為集群
*/
//consumer.setMessageModel(MessageModel.CLUSTERING);
/**
* 設置一次消費消息的條數，默認為1條
*/
consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
try {
/**
* 設置該消費者訂閱的主題和tag，如果是訂閱該主題下的所有tag，則tag使用*；如果需要指定訂閱該主題下的某些tag，則使用||分割，例如tag1||tag2||tag3
*/
String[] topicTagsArr = topics.split(";");
for (String topicTags : topicTagsArr) {
String[] topicTag = topicTags.split("~");
consumer.subscribe(topicTag[0],topicTag[1]);
}
consumer.start();
LOGGER.info("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr);
}catch (MQClientException e){
LOGGER.error("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr,e);
throw new RocketMQException(e);
}
return consumer;
}
}
5.消費者消息監聽處理器
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
/**
* 消費者消費消息路由
* .<br/>
*
* 注意，注意，注意： 實現了MessageListenerConcurrently 接口，多線程消費，並不能適用於順序有序消費, 若順序有序消費，請參照 rocketMQ(5) 第5章節
*
* @ClassName: RocketMQMessageListenerConcurrentlyProcessor
* @Description:
* @version: v1.0.0
*/
@Component
public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently{
private static final Logger logger = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);
/**
* 默認msgs里只有一條消息，可以通過設置consumeMessageBatchMaxSize參數來批量接收消息<br/>
* 不要拋異常，如果沒有return CONSUME_SUCCESS ，consumer會重新消費該消息，直到return CONSUME_SUCCESS
*/
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
if(CollectionUtils.isEmpty(msgs)){
logger.info("接受到的消息為空，不處理，直接返回成功");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
try{
MessageExt messageExt = msgs.get(0);
// topic
if(messageExt.getTopic().equals("TopicTest")){
// tag
if(messageExt.getTags().equals("topicTest")){
//TODO 判斷該消息是否重復消費（RocketMQ不保證消息不重復，如果你的業務需要保證嚴格的不重復消息，需要你自己在業務端去重）
//TODO 獲取該消息重試次數
int reconsume = messageExt.getReconsumeTimes();
if(reconsume >=3){//消息已經重試了3次，如果不需要再次消費，則返回成功
//TODO 不再消費，人工補償機制，記錄數據庫,發短信通知運維人員
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
//TODO 處理對應的業務邏輯
System.out.println("messageExt信息被接收:" + new String(messageExt.getBody()).toString());
}
}
// 如果沒有return success ，consumer會重新消費該消息，直到return success
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}catch (Exception e){
e.printStackTrace();
//TODO 拋出異常，需要重新消費
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
}
至此RocketMq配置已經全部完成，現在編寫測試用例測試下。
6.生產者Test
package com.clouds.common.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
@RunWith(SpringRunner.class)
@SpringBootTest
public class DefaultProductTest {
private static final Logger logger = LoggerFactory.getLogger(DefaultProductTest.class);
/**使用RocketMq的生產者*/
@Autowired
private DefaultMQProducer defaultMQProducer;
/**
* 發送消息
*
* 2018年3月3日 zhaowg
* @throws InterruptedException
* @throws MQBrokerException
* @throws RemotingException
* @throws MQClientException
*/
@Test
public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
String msg = "demo msg test";
logger.info("開始發送消息："+msg);
Message sendMsg = new Message("DemoTopic","DemoTag",msg.getBytes());
//默認3秒超時
SendResult sendResult = defaultMQProducer.send(sendMsg);
logger.info("消息發送響應信息："+sendResult.toString());
}
}
現在啟動SpringBootRocketMqApplication.java.
運行DefaultProductTest類的send方法發送消息，可以看到在消費者監聽器中已經獲取到消息了。
注意：發送的topic和tag，如果希望該應用收到消息，要現在application.properties中配置好rocketmq.consumer.topics值。
打印信息如下：
2018-03-03 00:52:39.785 INFO 14592 --- [ main] c.clouds.common.test.DefaultProductTest : 開始發送消息：demo msg test
2018-03-03 00:52:39.836 INFO 14592 --- [ main] c.clouds.common.test.DefaultProductTest : 消息發送響應信息：SendResult [sendStatus=SEND_OK, msgId=2F61081600002A9F00000000000A35F6, messageQueue=MessageQueue [topic=DemoTopic, brokerName=iZbp1g7kmh3b0jp1moarjoZ, queueId=0], queueOffset=2]
2018-03-03 00:52:39.853 INFO 14592 --- [ Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@3e694b3f: startup date [Sat Mar 03 00:52:37 CST 2018]; root of context hierarchy
2018-03-03 00:52:39.862 INFO 14592 --- [MessageThread_2] .c.c.r.c.p.MQConsumeMsgListenerProcessor : 接受到的消息為：MessageExt [queueId=0, storeSize=136, queueOffset=2, sysFlag=0, bornTimestamp=1520009559787, bornHost=/222.35.171.182:60888, storeTimestamp=1520009560327, storeHost=/47.97.8.22:10911, msgId=2F61081600002A9F00000000000A35F6, commitLogOffset=669174, bodyCRC=1945576729, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=DemoTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=3, WAIT=true, TAGS=DemoTag}, body=13]]
7.項目源代碼
https://gitee.com/zhaowg3/springboot-rocketmq/tree/simple/
3.啟動錯誤梳理
1.常見錯誤一
org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException:
sendDefaultImpl call timeout
解決方案
原因：阿里雲存在多網卡，rocketmq都會根據當前網卡選擇一個IP使用，當你的機器有多塊網卡時，很有可能會有問題。比如，我遇到的問題是我機器上有兩個IP，一個公網IP，一個私網IP, 因此需要配置broker.conf 指定當前的公網ip, 然后重新啟動broker
新增配置：conf/broker.conf (屬性名稱brokerIP1=broker所在的公網ip地址 )
新增這個配置：brokerIP1=120.76.62.13
啟動命令：nohup sh bin/mqbroker -n localhost:9876 -c ./conf/broker.conf &
2.常見錯誤二
MQClientException: No route info of this topic, TopicTest1
原因：Broker 禁止自動創建 Topic，且用戶沒有通過手工方式創建 此Topic, 或者broker和Nameserver網絡不通
解決：
通過 sh bin/mqbroker -m 查看配置
autoCreateTopicEnable=true 則自動創建topic
解決方案
Centos7關閉防火牆 systemctl stop firewalld
3.常見錯誤三
控制台查看不了數據，提示連接 10909錯誤
解決方案
原因：Rocket默認開啟了VIP通道，VIP通道端口為10911-2=10909
解決：阿里雲安全組需要增加一個端口 10909
4.常見錯誤4
DESC: service not available now, maybe disk full, CL:
解決方案
解決：修改啟動腳本runbroker.sh，在里面增加一句話即可：
JAVA_OPT="${JAVA_OPT} -Drocketmq.broker.diskSpaceWarningLevelRatio=0.98"
（磁盤保護的百分比設置成98%，只有磁盤空間使用率達到98%時才拒絕接收producer消息）
5.常見錯誤5
自動創建topic： autoCreateTopicEnable=true 無效原因
解決方案
客戶端版本要和服務端版本保持一致
6.常見錯誤6
RocketMQ客戶端連接提示service not available now, maybe disk full, CL: 0.87 CQ: 0.87 INDEX: 0.87, maybe your broker machine memory too small
解決方案
1: 應該是磁盤空間的問題，先看看磁盤空間
已用95%，查閱百度之后發現rocketmq源碼的DefaultMessageStore類里，默認會把剩余磁盤的比率不足75%(rocketmq版本不同這個比率好像不一樣)當做磁盤空間不足處理，看來磁盤是有點不夠了。
2.既然磁盤空間不夠，那就配置一下，把默認磁盤比率放大一些
先cd到rocketmq配置文件的路徑，我這里配置的是雙主雙從同步的模式，所以cd到配置文件（根據配置的不同文件夾的路徑不一樣，但都在/conf下）。
cd rocketmq-all-4.7.0-bin-release/conf/2m-2s-sync/
vim broker-a.properties
在最后加一行diskMaxUsedSpaceRatio=99(所有節點的配置文件都加一下)，表示剩余磁盤比例不足99才報錯
錯誤參考：https://www.cnblogs.com/shenrong/p/12670555.html
7.其他錯誤
https://blog.csdn.net/qq_14853889/article/details/81053145
https://blog.csdn.net/wangmx1993328/article/details/81588217#%E5%BC%82%E5%B8%B8%E8%AF%B4%E6%98%8E
https://www.jianshu.com/p/bfd6d849f156
https://blog.csdn.net/wangmx1993328/article/details/81588217
參考：https://www.jianshu.com/p/8c4c2c2ab62e
注意:部分資料從網絡收集整理，若有錯誤，請指正，多謝。