目錄
一、pom文件引入依賴
二、application.yml文件添加內(nèi)容
三、創(chuàng)建producer生產(chǎn)者
四、創(chuàng)建Consumer消費(fèi)者（創(chuàng)建兩個消費(fèi)者，所屬一個Topic）
五、啟動項目測試
RocketMQ 支持3 種消息發(fā)送方式： 同步 （sync）、異步（async）、單向（oneway）。
- 同步：發(fā)送者向 MQ 執(zhí)行發(fā)送消息API 時，同步等待，直到消息服務(wù)器返回發(fā)送結(jié)果。
- 異步：發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時，指定消息發(fā)送成功后的回調(diào)函數(shù)，然后調(diào)用消息發(fā)送API 后，立即返回，消息發(fā)送者線程不阻塞，直到運(yùn)行結(jié)束，消息發(fā)送成功或失敗的回調(diào)任務(wù)在一個新的線程中返回。
- 單向：消息發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時，直接返回，不等待消息服務(wù)器的結(jié)果，也不注冊回調(diào)函數(shù)，只管發(fā)，不管是否成功存儲在消息服務(wù)器上。
前提：
運(yùn)行項目需要具備RocketMQ環(huán)境，參考Docker搭建RocketMQ集群
一、pom文件引入依賴
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.2</version>
</dependency>二、application.yml文件添加內(nèi)容
rocketmq:
name-server: IP:9876 #IP為rocketmq訪問的地址
producer:
group: first1-group #事務(wù)消息才會用到三、創(chuàng)建producer生產(chǎn)者
package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class RocketController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@RequestMapping(value = "/rocket", method = RequestMethod.GET)
public void noTag() {
// convertAndSend() 發(fā)送普通字符串消息
rocketMQTemplate.convertAndSend("sendMessage_topic", "Hello Word");
}
@RequestMapping(value = "/tagA", method = RequestMethod.GET)
public void tagA() {
rocketMQTemplate.convertAndSend("sendMessage_topic:tagA", "hello world tagA");
}
@RequestMapping(value = "/tagB", method = RequestMethod.GET)
public void tagB() {
rocketMQTemplate.convertAndSend("sendMessage_topic:tagB", "hello world tagB");
}
@RequestMapping(value = "/syncSend", method = RequestMethod.GET)
public void syncSend() {
String json = "發(fā)送同步消息";
SendResult sendResult = rocketMQTemplate.syncSend("sendMessage_topic:1", json);
System.out.println(sendResult);
}
@RequestMapping(value = "/aSyncSend", method = RequestMethod.GET)
public void aSyncSend() {
String json = "發(fā)送異步消息";
SendCallback callback = new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("發(fā)送消息成功");
}
@Override
public void onException(Throwable throwable) {
System.out.println("發(fā)送消息失敗");
}
};
rocketMQTemplate.asyncSend("sendMessage_topic", json, callback);
}
@RequestMapping(value = "/sendOneWay", method = RequestMethod.GET)
public void sendOneWay() {
rocketMQTemplate.sendOneWay("sendMessage_topic", "發(fā)送單向消息");
}
}
四、創(chuàng)建Consumer消費(fèi)者（創(chuàng)建兩個消費(fèi)者，所屬一個Topic）
Consumer1:
package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
* topic 是主題
* consumerGroup 是消費(fèi)者組，一條消息只能被同一個消費(fèi)者組里的一個消費(fèi)者消費(fèi)。
* selectorExpression 是用于消息過濾的，以 TAG 方式為例：
* 默認(rèn)為 "*"，表示不過濾，消費(fèi)此 topic 下所有消息
* 配置為 "tagA"，表示只消費(fèi)此 topic 下 TAG = tagA 的消息
* 配置為 "tagA || tagB"，表示消費(fèi)此 topic 下 TAG = tagA 或 TAG = tagB 的消息，以此類推
* 消費(fèi)模式：默認(rèn) CLUSTERING （ CLUSTERING：負(fù)載均衡 ）（ BROADCASTING：廣播機(jī)制 ）
*/
@RocketMQMessageListener(topic = "sendMessage_topic",
consumerGroup = "consumer-group-test1",
// selectorExpression = "tagA || tagB",
messageModel = MessageModel.CLUSTERING)
@Component
public class DemoConsumer1 implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("receive message1:" + s);
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("處理完成");
}
}
Consumer2:
package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
* topic 是主題
* consumerGroup 是消費(fèi)者組，一條消息只能被同一個消費(fèi)者組里的一個消費(fèi)者消費(fèi)。
* selectorExpression 是用于消息過濾的，以 TAG 方式為例：
* 默認(rèn)為 "*"，表示不過濾，消費(fèi)此 topic 下所有消息
* 配置為 "tagA"，表示只消費(fèi)此 topic 下 TAG = tagA 的消息
* 配置為 "tagA || tagB"，表示消費(fèi)此 topic 下 TAG = tagA 或 TAG = tagB 的消息，以此類推
* 消費(fèi)模式：默認(rèn) CLUSTERING （ CLUSTERING：負(fù)載均衡 ）（ BROADCASTING：廣播機(jī)制 ）
*/
@RocketMQMessageListener(topic = "sendMessage_topic",
consumerGroup = "consumer-group-test1",
messageModel = MessageModel.CLUSTERING)
@Component
public class DemoConsumer2 implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("receive message2:" + s);
try {
Thread.sleep(8000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("處理完成");
}
}
五、啟動項目測試
1、發(fā)送同步消息：
?以上輸出可以看到：同步消息發(fā)送后，消息發(fā)送到broker后就返回結(jié)果了，消費(fèi)端還未處理完，兩者互互不影響。
2、發(fā)送異步消息：
?以上輸出：發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時，指定消息發(fā)送成功后的回調(diào)函數(shù)，然后調(diào)用消息發(fā)送API 后，立即返回，消息發(fā)送者線程不阻塞，直到運(yùn)行結(jié)束，消息發(fā)送成功或失敗的回調(diào)任務(wù)在一個新的線程中返回。
3、發(fā)送單向消息：
文章來源：http://www.zghlxwxcb.cn/news/detail-637819.html
?以上輸出：消息發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時，直接返回，不等待消息服務(wù)器的結(jié)果，也不注冊回調(diào)函數(shù)，只管發(fā)，不管是否成功存儲在消息服務(wù)器上。文章來源地址http://www.zghlxwxcb.cn/news/detail-637819.html
到了這里，關(guān)于SpringBoot集成RocketMQ實現(xiàn)三種消息發(fā)送方式的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
