目錄
- SpringBoot如何優雅的使用RocketMQ
- 什麼是RocketMQ？
- RocketMQ環境安裝
- SpringBoot環境中使用RocketMQ
SpringBoot如何優雅的使用RocketMQ
MQ，是一種跨進程的通信機制，用於上下游傳遞消息。在傳統的互聯網架構中通常使用MQ來對上下游來做解耦合。
舉例：當A系統對B系統進行消息通訊，如A系統發佈一條系統公告，B系統可以訂閱該頻道進行系統公告同步，整個過程中A系統並不關係B系統會不會同步，由訂閱該頻道的系統自行處理。
什麼是RocketMQ？
官方說明:
隨著使用越來越多的隊列和虛擬主題，ActiveMQ IO模塊遇到了瓶頸。我們盡力通過節流，斷路器或降級來解決此問題，但效果不佳。因此，我們那時開始關注流行的消息傳遞解決方案Kafka。不幸的是，Kafka不能滿足我們的要求，特別是在低延遲和高可靠性方面。
看到這裡可以很清楚的知道RcoketMQ 是一款低延遲、高可靠、可伸縮、易於使用的消息中間件。
具有以下特性：
- 支持發佈/訂閱（Pub/Sub）和點對點（P2P）消息模型
- 能夠保證嚴格的消息順序，在一個隊列中可靠的先進先出（FIFO）和嚴格的順序傳遞
- 提供豐富的消息拉取模式，支持拉（pull）和推（push）兩種消息模式
- 單一隊列百萬消息的堆積能力，億級消息堆積能力
- 支持多種消息協議，如 JMS、MQTT 等
- 分佈式高可用的部署架構,滿足至少一次消息傳遞語義
RocketMQ環境安裝
下載地址：https://rocketmq.apache.org/dowloading/releases/
從官方下載二進制或者源碼來進行使用。源碼編譯需要Maven3.2x，JDK8
在根目錄進行打包:
<code>Copymvn -Prelease-all -DskipTests clean packager -U/<code>
distribution/target/apache-rocketmq文件夾中會存在一個文件夾版,zip,tar三個可運行的完整程序。
使用rocketmq-4.6.0.zip:
- 啟動名稱服務 mqnamesrv.cmd
- 啟動數據中心 mqbroker.cmd -n localhost:9876
SpringBoot環境中使用RocketMQ
SpringBoot 入門：https://www.cnblogs.com/SimpleWu/p/10027237.htmlSpringBoot 常用start:https://www.cnblogs.com/SimpleWu/p/9798146.html當前環境版本為:
- SpringBoot 2.0.6.RELEASE
- SpringCloud Finchley.RELEASE
- SpringCldod Alibaba 0.2.1.RELEASE
- RocketMQ 4.3.0
在項目工程中導入:
<code>
<dependency>
<groupid>org.apache.rocketmq/<groupid>
<artifactid>rocketmq-client/<artifactid>
<version>${rocketmq.version}/<version>
/<dependency>
/<code>
由於我們這邊已經有工程了所以就不在進行創建這種過程了。主要是看看如何使用RocketMQ。
創建RocketMQProperties配置屬性類,類中內容如下:
<code>@ConfigurationProperties(prefix = "rocketmq")
public class RocketMQProperties {
private boolean isEnable = false;
private String namesrvAddr = "localhost:9876";
private String groupName = "default";
private int producerMaxMessageSize = 1024;
private int producerSendMsgTimeout = 2000;
private int producerRetryTimesWhenSendFailed = 2;
private int consumerConsumeThreadMin = 5;
private int consumerConsumeThreadMax = 30;
private int consumerConsumeMessageBatchMaxSize = 1;
//省略get set
}/<code>
現在我們所有子系統中的生產者，消費者對應：
isEnable 是否開啟mq
namesrvAddr 集群地址
groupName 分組名稱
設置為統一已方便系統對接,如有其它需求在進行擴展,類中我們已經給了默認值也可以在配置文件或配置中心中獲取配置，配置如下:
<code>#發送同一類消息的設置為同一個group，保證唯一,默認不需要設置，rocketmq會使用ip@pid(pid代表jvm名字)作為唯一標示
rocketmq.groupName=please_rename_unique_group_name
#是否開啟自動配置
rocketmq.isEnable=true
#mq的nameserver地址
rocketmq.namesrvAddr=127.0.0.1:9876
#消息最大長度 默認1024*4(4M)
rocketmq.producer.maxMessageSize=4096
#發送消息超時時間,默認3000
rocketmq.producer.sendMsgTimeout=3000
#發送消息失敗重試次數，默認2
rocketmq.producer.retryTimesWhenSendFailed=2
#消費者線程數量
rocketmq.consumer.consumeThreadMin=5
rocketmq.consumer.consumeThreadMax=32
#設置一次消費消息的條數，默認為1條
rocketmq.consumer.consumeMessageBatchMaxSize=1/<code>
創建消費者接口 RocketConsumer.java 該接口用戶約束消費者需要的核心步驟:
<code>/**
* 消費者接口
*
* @author SimpleWu
*
*/
public interface RocketConsumer {
/**
* 初始化消費者
*/
public abstract void init();
/**
* 註冊監聽
*
* @param messageListener
*/
public void registerMessageListener(MessageListener messageListener);
}/<code>
創建抽象消費者 AbstractRocketConsumer.java：
<code>/**
* 消費者基本信息
*
* @author SimpelWu
*/
public abstract class AbstractRocketConsumer implements RocketConsumer {
protected String topics;
protected String tags;
protected MessageListener messageListener;
protected String consumerTitel;
protected MQPushConsumer mqPushConsumer;
/**
* 必要的信息
*
* @param topics
* @param tags
* @param consumerTitel
*/
public void necessary(String topics, String tags, String consumerTitel) {
this.topics = topics;
this.tags = tags;
this.consumerTitel = consumerTitel;
}
public abstract void init();
@Override
public void registerMessageListener(MessageListener messageListener) {
this.messageListener = messageListener;
}
}/<code>
在類中我們必須指定這個topics,tags與消息監聽邏輯
<code>public abstract void init();該方法是用於初始化消費者，由子類實現。/<code>
接下來我們編寫自動配置類RocketMQConfiguation.java,該類用戶初始化一個默認的生產者連接，以及加載所有的消費者。
@EnableConfigurationProperties({ RocketMQProperties.class }) 使用該配置文件
@Configuration 標註為配置類
@ConditionalOnProperty(prefix = "rocketmq", value = "isEnable", havingValue = "true") 只有當配置中指定rocketmq.isEnable = true的時候才會生效
核心內容如下:
<code>/**
* mq配置
*
* @author SimpleWu
*/
@Configuration
@EnableConfigurationProperties({ RocketMQProperties.class })
@ConditionalOnProperty(prefix = "rocketmq", value = "isEnable", havingValue = "true")
public class RocketMQConfiguation {
private RocketMQProperties properties;
private ApplicationContext applicationContext;
private Logger log = LoggerFactory.getLogger(RocketMQConfiguation.class);
public RocketMQConfiguation(RocketMQProperties properties, ApplicationContext applicationContext) {
this.properties = properties;
this.applicationContext = applicationContext;
}
/**
* 注入一個默認的消費者
* @return
* @throws MQClientException
*/
@Bean
public DefaultMQProducer getRocketMQProducer() throws MQClientException {
if (StringUtils.isEmpty(properties.getGroupName())) {
throw new MQClientException(-1, "groupName is blank");
}
if (StringUtils.isEmpty(properties.getNamesrvAddr())) {
throw new MQClientException(-1, "nameServerAddr is blank");
}
DefaultMQProducer producer;
producer = new DefaultMQProducer(properties.getGroupName());
producer.setNamesrvAddr(properties.getNamesrvAddr());
// producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");
// 如果需要同一個jvm中不同的producer往不同的mq集群發送消息，需要設置不同的instanceName
// producer.setInstanceName(instanceName);
producer.setMaxMessageSize(properties.getProducerMaxMessageSize());
producer.setSendMsgTimeout(properties.getProducerSendMsgTimeout());
// 如果發送消息失敗，設置重試次數，默認為2次
producer.setRetryTimesWhenSendFailed(properties.getProducerRetryTimesWhenSendFailed());
try {
producer.start();
log.info("producer is start ! groupName:{},namesrvAddr:{}", properties.getGroupName(),
properties.getNamesrvAddr());
} catch (MQClientException e) {
log.error(String.format("producer is error {}", e.getMessage(), e));
throw e;
}
return producer;
}
/**
* SpringBoot啟動時加載所有消費者
*/
@PostConstruct
public void initConsumer() {
Map<string> consumers = applicationContext.getBeansOfType(AbstractRocketConsumer.class);
if (consumers == null || consumers.size() == 0) {
log.info("init rocket consumer 0");
}
Iterator<string> beans = consumers.keySet().iterator();
while (beans.hasNext()) {
String beanName = (String) beans.next();
AbstractRocketConsumer consumer = consumers.get(beanName);
consumer.init();
createConsumer(consumer);
log.info("init success consumer title {} , toips {} , tags {}", consumer.consumerTitel, consumer.tags,
consumer.topics);
}
}
/**
* 通過消費者信心創建消費者
*
* @param consumerPojo
*/
public void createConsumer(AbstractRocketConsumer arc) {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.properties.getGroupName());
consumer.setNamesrvAddr(this.properties.getNamesrvAddr());
consumer.setConsumeThreadMin(this.properties.getConsumerConsumeThreadMin());
consumer.setConsumeThreadMax(this.properties.getConsumerConsumeThreadMax());
consumer.registerMessageListener(arc.messageListenerConcurrently);
/**
* 設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費 如果非第一次啟動，那麼按照上次消費的位置繼續消費
*/
// consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
/**
* 設置消費模型，集群還是廣播，默認為集群
*/
// consumer.setMessageModel(MessageModel.CLUSTERING);
/**
* 設置一次消費消息的條數，默認為1條
*/
consumer.setConsumeMessageBatchMaxSize(this.properties.getConsumerConsumeMessageBatchMaxSize());
try {
consumer.subscribe(arc.topics, arc.tags);
consumer.start();
arc.mqPushConsumer=consumer;
} catch (MQClientException e) {
log.error("info consumer title {}", arc.consumerTitel, e);
}
}
}/<string>/<string>/<code>
然後在src/main/resources文件夾中創建目錄與文件META-INF/spring.factories裡面添加自動配置類即可開啟啟動配置，我們只需要導入依賴即可:
<code>org.springframework.boot.autoconfigure.EnableAutoConfiguration=\\
com.xcloud.config.rocketmq.RocketMQConfiguation/<code>
接下來在服務中導入依賴，然後通過我們的抽象類獲取所有必要信息對消費者進行創建，該步驟會在所有消費者初始化完成後進行，且只會管理是Spring Bean的消費者。
下面我們看看如何創建一個消費者，創建消費者的步驟非常簡單，只需要繼承AbstractRocketConsumer然後再加上Spring的@Component就能夠完成消費者的創建，我們可以在類中自定義消費的主題與標籤。
在項目可以根據需求當消費者創建失敗的時候是否繼續啟動工程。
創建一個默認的消費者 DefaultConsumerMQ.java
<code>@Component
public class DefaultConsumerMQ extends AbstractRocketConsumer {
/**
* 初始化消費者
*/
@Override
public void init() {
// 設置主題,標籤與消費者標題
super.necessary("TopicTest", "*", "這是標題");
//消費者具體執行邏輯
registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<messageext> msgs, ConsumeConcurrentlyContext context) {
msgs.forEach(msg -> {
System.out.printf("consumer message boyd %s %n", new String(msg.getBody()));
});
// 標記該消息已經被成功消費
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
}
}/<messageext>/<code>
super.necessary("TopicTest", "*", "這是標題"); 是必須要設置的，代表該消費者監聽TopicTest主題下所有tags,標題那個字段是我自己定義的，所以對於該配置來說沒什麼意義。
我們可以在這裡注入Spring的Bean來進行任意邏輯處理。
創建一個消息發送類進行測試
<code>@Override
public String qmtest(@PathVariable("name")String name) throws MQClientException, RemotingException, MQBrokerException, InterruptedException, UnsupportedEncodingException {
Message msg = new Message("TopicTest", "tags1", name.getBytes(RemotingHelper.DEFAULT_CHARSET));
// 發送消息到一個Broker
SendResult sendResult = defaultMQProducer.send(msg);
// 通過sendResult返回消息是否成功送達
System.out.printf("%s%n", sendResult);
return null;
}/<code>
我們來通過Http請求測試:
<code>http://localhost:10001/demo/base/mq/hello consumer message boyd hello
http://localhost:10001/demo/base/mq/嘿嘿嘿嘿嘿 consumer message boyd 嘿嘿嘿嘿嘿 /<code>
好了到這裡簡單的start算是設計完成了，後面還有一些：順序消息生產，順序消費消息，異步消息生產等一系列功能，官人可參照官方去自行處理。
- ActiveMQ 沒經過大規模吞吐量場景的驗證，社區不高不活躍。
- RabbitMQ 集群動態擴展麻煩，且與當前程序語言不至於難以定製化。
- kafka 支持主要的MQ功能，功能無法達到程序需求的要求，所以不使用,且與當前程序語言不至於難以定製化。
- rocketMQ 經過全世界的女人的洗禮，已經很強大;MQ功能較為完善，還是分佈式的，擴展性好;支持複雜MQ業務場景。（業務複雜可做首選）
來源：https://www.cnblogs.com/SimpleWu/p/12112351.html
閱讀更多 java樂園 的文章
