寫在前面
這裡介紹下Springboot 內建RocketMQ的三種方式
一、原生 jar（rocketmq-client）
1.1、producer
1.1.1、三個基本使用
- producerGroup，定義生産者組
- DefaultMQProducer，定義生産者配置
- TransactionMQProducer，定義支援事務生産者
1.1.2、三種基本發送方式：
- 同步發送
- 異步發送
- 單項發送
同步發送，代碼示例
/**
* 同步發送實體對象消息
* 可靠同步發送：同步發送是指消息發送方發出資料後，會在收到接收方發回響應之後才發下一個資料包的通訊方式；
* 特點：速度快；有結果回報；資料可靠；
* 應用場景：應用場景非常廣泛，例如重要通知郵件、報名短信通知、營銷短信系統等；
*
* @param topic
* @param tags
* @param body
* @return
* @throws InterruptedException
* @throws RemotingException
* @throws MQClientException
* @throws MQBrokerException
* @throws UnsupportedEncodingException
*/
public String syncSend(String topic, String tags, String body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
Message msg = new Message(topic /* Topic */,
tags /* Tag */,
("Hello RocketMQ ").getBytes() /* Message body */
);
// 發送消息到一個Broker
SendResult sendResult = producer.send(msg);
// 通過sendResult傳回消息是否成功送達
System.out.printf("%s%n", sendResult);
TimeUnit.SECONDS.sleep(1);
return "{\"MsgId\":\"" + sendResult.getMsgId() + "\"}";
} 異步發送，代碼示例
/**
* 異步發送消息
* 可靠異步發送：發送方發出資料後，不等接收方發回響應，接着發送下個資料包的通訊方式；
* 特點：速度快；有結果回報；資料可靠；
* 應用場景：異步發送一般用于鍊路耗時較長,對 rt響應時間較為敏感的業務場景,例如使用者視訊上傳後通知啟動轉碼服務,轉碼完成後通知推送轉碼結果等；
*
* @param topic
* @param tags
* @param body
* @return
* @throws InterruptedException
* @throws RemotingException
* @throws MQClientException
* @throws MQBrokerException
* @throws UnsupportedEncodingException
*/
public void asyncSend(String topic, String tags, String body) throws Exception {
Message msg = new Message(topic /* Topic */,
tags /* Tag */,
("Hello RocketMQ ").getBytes() /* Message body */
);
// 發送消息到一個Broker
producer.send(msg, new SendCallback() {
public void onSuccess(SendResult sendResult) {
System.out.println("發送結果 ： " + sendResult);
}
public void onException(Throwable throwable) {
System.out.println(throwable.getMessage());
}
});
TimeUnit.SECONDS.sleep(1);
} 單項發送，代碼示例
/**
* 單向發送
* 單向發送：隻負責發送消息，不等待伺服器回應且沒有回調函數觸發，即隻發送請求不等待應答；此方式發送消息的過程耗時非常短，一般在微秒級别；
* 特點：速度最快，耗時非常短，毫秒級别；無結果回報；資料不可靠，可能會丢失；
* 應用場景：适用于某些耗時非常短，但對可靠性要求并不高的場景，例如日志收集；
*
* @param topic
* @param tags
* @param body
* @throws InterruptedException
* @throws RemotingException
* @throws MQClientException
* @throws MQBrokerException
* @throws UnsupportedEncodingException
*/
public void oneway(String topic, String tags, String body) throws Exception {
Message msg = new Message(topic /* Topic */,
tags /* Tag */,
("Hello RocketMQ ").getBytes() /* Message body */
);
producer.sendOneway(msg);
TimeUnit.SECONDS.sleep(1);
} 1.1.3、其他發送特性
- 消息延遲
- 設定消息屬性，用于消費過濾
- 消息隊列選擇器Selector
- 事務監聽
消息延遲
/**
* 延遲 消費
*/
public void delayTestListener() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delayGroup");
consumer.setNamesrvAddr(namesrvAddr);
try {
// 訂閱PushTopic下Tag為push的消息,都訂閱消息
consumer.subscribe("delayPushMsg", "push");
// 程式第一次啟動從消息隊列頭擷取資料
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//可以修改每次消費消息的數量，預設設定是每次消費一條
consumer.setConsumeMessageBatchMaxSize(1);
//在此監聽中消費資訊，并傳回消費的狀态資訊
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
// 會把不同的消息分别放置到不同的隊列中
for (MessageExt msg : msgs) {
log.info("Receive message:msgId={},msgBody={},delay={} ms",
msg.getMsgId(),
new String(msg.getBody()),
(System.currentTimeMillis() - msg.getStoreTimestamp()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
} 設定消息屬性，用于消費過濾
/**
* todo 注意這裡 需要啟動 broker 前，設定 支援SQL92 Filter = enable
* sql filter
*
* @param topic
* @param tags
* @param body
* @param i
* @throws Exception
*/
public void filtersql(String topic, String tags, String body, int i) throws Exception {
//消息
Message message = new Message(topic, tags, body.getBytes());
//設定消息屬性
message.putUserProperty("i", String.valueOf(i));
//發送消息
SendResult sendresult = producer.send(message);
System.out.println("消息結果 ：" + sendresult);
TimeUnit.SECONDS.sleep(1);
} 消息隊列選擇器Selector
/**
* Order 測試
*
* @param topic
* @param tags
* @param body
* @param order
* @throws Exception
*/
public void orderPush(String topic, String[] tags, String body, Boolean order) throws Exception {
// 訂單清單
List<OrderStep> orderList = this.buildOrders();
Date date = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String dateStr = sdf.format(date);
for (int i = 0; i < 10; i++) {
if (order) {
log.info("有序的消費，根據隊列id，配置設定分組，啟動相應的唯一消費線程");
// 加個時間字首
String body1 = dateStr + body + orderList.get(i);
Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, body1.getBytes());
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Long id = (Long) arg; //根據訂單id選擇發送queue
long index = id % mqs.size();
return mqs.get((int) index);
}
}, orderList.get(i).getOrderId());//訂單id
System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
sendResult.getSendStatus(),
sendResult.getMessageQueue().getQueueId(),
body1));
} else {
log.info("無序的的消費，需等所有消息釋出完成，在配置設定，根據隊列id，啟動相應的唯一消費線程");
// 加個時間字首
String body1 = dateStr + body + orderList.get(i);
Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, body1.getBytes());
SendResult sendResult = producer.send(msg);
System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
sendResult.getSendStatus(),
sendResult.getMessageQueue().getQueueId(),
body1));
}
}
} 事務監聽
/**
* 事務測試
*
* @param topic
* @param tags
* @param body
* @throws Exception
*/
public void tasnsaction(String topic, String[] tags, String body) throws Exception {
//建立事務監聽器
TransactionListener listener = new TransactionListener() {
/**
* When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
*
* @param message Half(prepare) message
* @param o Custom business parameter
* @return Transaction state
*/
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
if ("Tag1".equals(message.getTags())) {
return LocalTransactionState.COMMIT_MESSAGE;
} else if ("Tag2".equals(message.getTags())) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else return LocalTransactionState.UNKNOW;
}
/**
* When no response to prepare(half) message. broker will send check message to check the transaction status, and this
* method will be invoked to get local transaction status.
*
* @param messageExt Check message
* @return Transaction state
*/
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
System.out.println(messageExt.getTags() + "消息回查！");
return LocalTransactionState.COMMIT_MESSAGE;
}
};
//set事務監聽器
transProducer.setTransactionListener(listener);
//發送消息
for (int i = 0; i < 3; i++) {
Message message = new Message(topic, tags[i], "KEY" + i, (body + i).getBytes());
SendResult sendResult = transProducer.sendMessageInTransaction(message, null);
System.out.println("發送結果 ：" + sendResult);
TimeUnit.SECONDS.sleep(2);
}
} 1.2、consumer
1.2.1、基本使用
- DefaultMQPushConsumer
1.2.2、監聽示例
基本配置
/**
* RocketMq配置監聽資訊
*/
public void messageListener() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SpringBootRocketMqGroup");
consumer.setNamesrvAddr(namesrvAddr);
try {
// 訂閱PushTopic下Tag為push的消息,都訂閱消息
consumer.subscribe("PushTopic", "push");
// 程式第一次啟動從消息隊列頭擷取資料
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//負載均衡模式消費
// consumer.setMessageModel(MessageModel.BROADCASTING);
//可以修改每次消費消息的數量，預設設定是每次消費一條
consumer.setConsumeMessageBatchMaxSize(1);
//在此監聽中消費資訊，并傳回消費的狀态資訊
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
// 會把不同的消息分别放置到不同的隊列中
for (Message msg : msgs) {
System.out.println("接收到了消息：" + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
} filter tag
/**
* filter tag 監聽
*/
public void filterTagListener() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tagFilterGroup");
consumer.setNamesrvAddr(namesrvAddr);
try {
//訂閱的topic與tag
consumer.subscribe("topic1", "tag1 || tag2");
//注冊消息監聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
log.info("收到消息:Keys->{},body->{}", msg.getKeys(), new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
} 條件過濾
/**
* sql92 條件過濾
*/
public void sqlFilterTagListener() {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sqlFilterGroup");
consumer.setNamesrvAddr(namesrvAddr);
try {
//設定訂閱條件
consumer.subscribe("topic2", MessageSelector.bySql("i > 5"));
//注冊監聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt msg : list) {
log.info("收到消息:Keys->{},body->{},i ->{}",
msg.getKeys(),
new String(msg.getBody()),
msg.getProperty("i"));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
} 二、rocketmq-spring-boot-starter
Maven 坐标
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>${rocketmq-spring-boot-starter-version}</version>
</dependency> 基本架構，不是很複雜，将RocketMQ client Jar 中的相關連接配接配置，Message 對象轉換成了Spring Bean
基本配置
rocketmq.name-server=localhost:9876
rocketmq.producer.group=boot-group1
rocketmq.producer.sendMessageTimeout=300000 2.1、producer
兩個模闆對象
- RocketMQTemplate，内置對象
- extRocketMQTemplate,需自定義的RocketMQTemplate
RocketMQTemplate，其實封裝了Spring message 和 rocket-client 的相關轉換，實作
同步發送，代碼示例
/**
* Send string
* localhost:10001/sendString
*/
@GetMapping("/sendString")
public void sendString() {
SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
System.out.printf("syncSend1 to topic %s sendResult=%s %n", springTopic, sendResult);
sendResult = rocketMQTemplate.syncSend(userTopic, new User().setUserAge((byte) 18).setUserName("Kitty"));
System.out.printf("syncSend1 to topic %s sendResult=%s %n", userTopic, sendResult);
sendResult = rocketMQTemplate.syncSend(userTopic, MessageBuilder.withPayload(
new User().setUserAge((byte) 21).setUserName("Lester")).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build());
System.out.printf("syncSend1 to topic %s sendResult=%s %n", userTopic, sendResult);
} 異步發送，代碼示例
/**
* 異步
* Send user-defined object
* localhost:10001/send-with-user-defined
*/
@GetMapping("/send-with-user-defined")
public void userDefined() {
rocketMQTemplate.asyncSend(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")), new SendCallback() {
@Override
public void onSuccess(SendResult var1) {
System.out.printf("async onSucess SendResult=%s %n", var1);
}
@Override
public void onException(Throwable var1) {
System.out.printf("async onException Throwable=%s %n", var1);
}
});
} 單向發送，代碼示例
/**
* 單向發送
*/
@GetMapping("/send-with-oneWay")
public void sendOneWay() {
rocketMQTemplate.sendOneWay(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")));
} 事務監聽處理
/**
* Send transactional messages using rocketMQTemplate
* localhost:10001/send-transactional-rocketMQTemplate
*/
@GetMapping("/send-transactional-rocketMQTemplate")
public void transactionalRocketMQTemplate() {
String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 10; i++) {
try {
Message msg = MessageBuilder.withPayload("rocketMQTemplate transactional message " + i).
setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
springTransTopic + ":" + tags[i % tags.length], msg, null);
System.out.printf("------rocketMQTemplate send Transactional msg body = %s , sendResult=%s %n",
msg.getPayload(), sendResult.getSendStatus());
Thread.sleep(10);
} catch (Exception e) {
e.printStackTrace();
}
}
} // todo 這裡還有很多種方式，具體操作場景、差别還有待研究，學習
2.2、consumer
這裡比較簡單，代碼示例如下
/**
* StringConsumer
*/
@Service
@RocketMQMessageListener(topic = "${demo.rocketmq.topic}",
consumerGroup = "string_consumer",
selectorExpression = "${demo.rocketmq.tag}")
public class StringConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.printf("------- StringConsumer received: %s \n", message);
}
} /**
* The consumer that replying String
*/
@Service
@RocketMQMessageListener(topic = "${demo.rocketmq.stringRequestTopic}",
consumerGroup = "${demo.rocketmq.stringRequestConsumer}",
selectorExpression = "${demo.rocketmq.tag}")
public class StringConsumerWithReplyString implements RocketMQReplyListener<String, String> {
@Override
public String onMessage(String message) {
System.out.printf("------- StringConsumerWithReplyString received: %s \n", message);
return "reply string";
}
} 其他，需自行學習…
三、阿裡雲 ONS
首先，對于測試學習，挺貴的哈，但是卻提供了非常可靠的消息機制
使用上也很簡單，引入 SDK
<!-- https://mvnrepository.com/artifact/com.aliyun.openservices/ons-client -->
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.6.Final</version>
</dependency> 3.1、配置
rocketmq:
producer:
producerId: GroupId#生産者id(舊版本是生産者id，新版本是groupid)，
msgTopic: Test #生産主題
accessKey: XXX #連接配接通道
secretKey: XXX #連接配接秘鑰
onsAddr: #生産者ons接入域名 3.2、使用
package com.tonels.spring.boot.rocketmq.producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Properties;
/**
* rocketmq生産者啟動初始化類
*
*/
@Component
public class RocketmqProducerInit {
@Value("${rocketmq.producer.producerId}")
private String producerId;
@Value("${rocketmq.producer.accessKey}")
private String accessKey;
@Value("${rocketmq.producer.secretKey}")
private String secretKey;
@Value("${rocketmq.producer.onsAddr}")
private String ONSAddr;
private static Producer producer;
@PostConstruct
public void init(){
System.out.println("初始化啟動生産者！");
// producer 執行個體配置初始化
Properties properties = new Properties();
//您在控制台建立的Producer ID
properties.setProperty(PropertyKeyConst.GROUP_ID, producerId);
// AccessKey 阿裡雲身份驗證，在阿裡雲伺服器管理控制台建立
properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
// SecretKey 阿裡雲身份驗證，在阿裡雲伺服器管理控制台建立
properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
//設定發送逾時時間，機關毫秒
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
// 設定 TCP 接入域名(此處以公共雲生産環境為例)，設定 TCP 接入域名，進入 MQ 控制台的消費者管理頁面，在左側操作欄單擊擷取接入點擷取
properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr);
producer = ONSFactory.createProducer(properties);
// 在發送消息前，初始化調用start方法來啟動Producer，隻需調用一次即可，當項目關閉時，自動shutdown
producer.start();
}
/**
* 初始化生産者
* @return
*/
public Producer getProducer(){
return producer;
}
} @Autowired
private RocketmqProducerInit producer;
public boolean sendMsg(String msg) {
Long startTime = System.currentTimeMillis();
Message message = new Message(msgTopic, tag, msg.getBytes());
SendResult sendResult = producer.getProducer().send(message);
if (sendResult != null) {
System.out.println(new Date() + " Send mq message success. Topic is:" + message.getTopic() + " msgId is: " + sendResult.getMessageId());
} else {
logger.warn(".sendResult is null.........");
}
Long endTime = System.currentTimeMillis();
System.out.println("單次生産耗時："+(endTime-startTime)/1000);
return true;
}