1.用戶角色配置
自帶的guest/guest 超級管理員
五中不同角色配置：
- 普通管理者(management)：僅可登陸管理控制臺，無法看到節點資訊，也無法對策略進行管理，
- 策略制定者(policymaker)：可登陸管理控制臺, 同時可以對policy進行管理，但無法查看節點的相關資訊，
- 監控者 (monitoring)：登錄管理控制臺 查看所有的資訊(Rabbit的相關節點的資訊,記憶體使用資訊,磁盤的情況)
- 超級管理員 administrator：登錄管理控制臺 查看所有的資訊 對所有用戶的策略進行操作
- 其他：無法登陸管理控制臺，通常就是普通的生產者和消費者(僅僅接收資訊或者發送訊息)，
2.Virtual Hosts配置
每一個 Virtual Hosts相互隔離, 就相當于一個相對獨立的RabbitMQ,===> exchange queue message不是互通
可以理解為：一個Mysql又很多資料庫,每一個資料庫就相當于一個Virtual Hosts
2.1創建Virtual Hosts
2.2權限的分配
點擊對應Virtual Hostsv 名字,進入配置頁面
該權限配置引數說明：
- user：用戶名
- configure ：一個正則運算式，用戶對符合該正則運算式的所有資源擁有 configure 操作的權限
- write：一個正則運算式，用戶對符合該正則運算式的所有資源擁有 write 操作的權限
- read：一個正則運算式，用戶對符合該正則運算式的所有資源擁有 read 操作的權限
3.入門案例
使用思路
生產者: --->按照JDBC思路
1. 創建連接工廠
2. 設定的rabbitMq的服務的主機 默認localhost
3. 設定的rabbitMq的服務的埠 默認5672
4. 設定的rabbitMq的服務的虛擬主機
5. 設定用戶和密碼
6. 創建連接
7. 創建頻道
8. 宣告佇列
9. 創建訊息
10. 發送訊息
11.關閉資源
（觀察者模式）
RabbitMQ 使用的是發布訂閱模式
所需依賴
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
對Rabbit的連接進行封裝
public class RabbitMQConnectionUtils {
/**
* 獲得Rabbit的連接
* 類比JDBC
*/
public static Connection getConection() throws IOException, TimeoutException {
//1.創建連接工廠
ConnectionFactory connectionFactory=new ConnectionFactory();
//2.設定的rabbitMq的服務的主機，默認localhost
connectionFactory.setHost("localhost");
//3.設定的rabbitMq的服務的埠，默認為5672
connectionFactory.setPort(5672);
//4.設定的rabbitMq的服務的虛擬主機
connectionFactory.setVirtualHost("testmq");
//5.設定用戶和密碼
connectionFactory.setUsername("chenbuting");
connectionFactory.setPassword("123");
//6.創建連接
Connection connection=connectionFactory.newConnection();
return connection;
}
}
基本訊息佇列
生產者Product相關代碼
public class Product {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection=RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel= connection.createChannel();
//8.宣告佇列（創建佇列）
/*
* 引數1：指定佇列的名稱
* 引數2：指定訊息是否持久化，一般設定為true,是否保存到磁盤當中去
* 引數3：指定是否獨占通道
* 引數4：是否自動洗掉
* 引數5：指定額外引數
*/
channel.queueDeclare("simple_01",true,false,false,null);
//9.創建訊息
String message="hello ,my name is chenbuting";
//10.發送訊息
/*
* 引數1：指定交換機 簡單模式中使用默認交換機 指定空字串
* 引數2: 指定routingkey 簡單模式 只需要指定佇列名稱即可
* 引數3： 指定攜帶的額外的引數 null
* 引數4：要發送的訊息本身（位元組陣列）
*/
channel.basicPublish("","simple_01",null,message.getBytes());
//11.關閉資源
channel.close();
connection.close();
}
}
消費者Consumer相關代碼
public class Consumer {
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection= RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel=connection.createChannel();
//8.宣告佇列
channel.queueDeclare("simple_01",true,false,false,null);
//9.接受訊息
//處理訊息
DefaultConsumer consumer = new DefaultConsumer(channel){
/*
*consumerTag: 消費者標簽，用于標識特定的消費者，每個消費者都有一個唯一的標簽，它可以用于取消訂閱或標識訊息是由哪個消費者處理的，
* envelope: 信封物件，包含與訊息傳遞相關的元資料，如交換機、路由鍵、傳遞標簽等，
* properties: AMQP 的基本屬性，包含附加的訊息屬性，如訊息持久性、優先級、時間戳等，
* body: 訊息的正文，以位元組陣列的形式提供，
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("訊息本身"+new String(body,"utf-8"));
System.out.println("exchange"+envelope.getExchange());
System.out.println("RoutingKey"+envelope.getRoutingKey());
System.out.println("訊息的序號"+envelope.getDeliveryTag());
}
};
//監聽器----->觀察者設計模式
channel.basicConsume("simple_01",true,consumer);
//10.建議不要關閉資源 建議一直監聽訊息
}
}
idea允許開啟多個實體的方式
作業訊息佇列
生產者相關代碼
public class WorkProduct {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection=RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel= connection.createChannel();
//8.宣告佇列（創建佇列）
/*
* 引數1：指定佇列的名稱
* 引數2：指定訊息是否持久化，一般設定為true,是否保存到磁盤當中去
* 引數3：指定是否獨占通道
* 引數4：是否自動洗掉
* 引數5：指定額外引數
*/
channel.queueDeclare("working_02",true,false,false,null);
for (int i=0;i<30;i++) {
//9.創建訊息
String message = "hello ,my name is CHENBUTING"+i;
//10.發送訊息
/*
* 引數1：指定交換機 簡單模式中使用默認交換機 指定空字串
* 引數2: 指定routingkey 簡單模式 只需要指定佇列名稱即可
* 引數3： 指定攜帶的額外的引數 null
* 引數4：要發送的訊息本身（位元組陣列）
*/
channel.basicPublish("", "working_02", null, message.getBytes());
}
//11.關閉資源
channel.close();
connection.close();
}
}
消費者相關代碼
WorkConsumer_1和WorkConsumer_2代碼內容
public class WorkConsumer_1 {
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection= RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel=connection.createChannel();
//8.宣告佇列
channel.queueDeclare("working_02",true,false,false,null);
//9.接受訊息
//處理訊息
DefaultConsumer consumer = new DefaultConsumer(channel){
/*
*consumerTag: 消費者標簽，用于標識特定的消費者，每個消費者都有一個唯一的標簽，它可以用于取消訂閱或標識訊息是由哪個消費者處理的，
* envelope: 信封物件，包含與訊息傳遞相關的元資料，如交換機、路由鍵、傳遞標簽等，
* properties: AMQP 的基本屬性，包含附加的訊息屬性，如訊息持久性、優先級、時間戳等，
* body: 訊息的正文，以位元組陣列的形式提供，
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("訊息本身"+new String(body,"utf-8"));
System.out.println("exchange"+envelope.getExchange());
System.out.println("RoutingKey"+envelope.getRoutingKey());
System.out.println("訊息的序號"+envelope.getDeliveryTag());
}
};
//監聽器----->觀察者設計模式
channel.basicConsume("working_02",true,consumer);
//10.建議不要關閉資源 建議一直監聽訊息
}
}
結果
訂閱模式
Fanout模式
Fanout生產者的相關代碼
public class FanoutProduct {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection=RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel= connection.createChannel();
//創建交換機
channel.exchangeDeclare("exchange_fanout", BuiltinExchangeType.FANOUT);
//8.宣告佇列（創建佇列）
/*
* 引數1：指定佇列的名稱
* 引數2：指定訊息是否持久化，一般設定為true,是否保存到磁盤當中去
* 引數3：指定是否獨占通道
* 引數4：是否自動洗掉
* 引數5：指定額外引數
*/
channel.queueDeclare("fanout_queue1",true,false,false,null);
channel.queueDeclare("fanout_queue2",true,false,false,null);
//9.創建訊息
String message = "my name is Fanout";
//10.發送訊息
/*
*這段代碼使用 RabbitMQ 的 Java 客戶端庫中的 `queueBind` 方法來將一個佇列系結到一個 fanout 型別的交換機上，下面是各個引數的介紹：
1. `fanout_queue1`: 佇列名稱，表示要系結的佇列的名稱，
2. `exchange_fanout`: 交換機名稱，表示要進行系結的交換機的名稱，
3. `""`: 路由鍵，表示要使用的路由鍵，
*/
channel.queueBind("fanout_queue1","exchange_fanout","");
channel.queueBind("fanout_queue2","exchange_fanout","");
/*
* 引數1：指定交換機 簡單模式中使用默認交換機 指定空字串
* 引數2: 指定routingkey 簡單模式 只需要指定佇列名稱即可
* 引數3： 指定攜帶的額外的引數 null
* 引數4：要發送的訊息本身（位元組陣列）
*/
channel.basicPublish("exchange_fanout", "", null, message.getBytes());
//11.關閉資源
channel.close();
connection.close();
}
}
Fanout消費者的相關代碼
public class FanoutConsumer_1 {
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection= RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel=connection.createChannel();
//8.宣告佇列
channel.queueDeclare("fanout_queue1",true,false,false,null);
//9.接受訊息
//處理訊息
DefaultConsumer consumer = new DefaultConsumer(channel){
/*
*consumerTag: 消費者標簽，用于標識特定的消費者，每個消費者都有一個唯一的標簽，它可以用于取消訂閱或標識訊息是由哪個消費者處理的，
* envelope: 信封物件，包含與訊息傳遞相關的元資料，如交換機、路由鍵、傳遞標簽等，
* properties: AMQP 的基本屬性，包含附加的訊息屬性，如訊息持久性、優先級、時間戳等，
* body: 訊息的正文，以位元組陣列的形式提供，
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("訊息本身"+new String(body,"utf-8"));
System.out.println("exchange"+envelope.getExchange());
System.out.println("RoutingKey"+envelope.getRoutingKey());
System.out.println("訊息的序號"+envelope.getDeliveryTag());
}
};
//監聽器----->觀察者設計模式
channel.basicConsume("fanout_queue1",true,consumer);
//10.建議不要關閉資源 建議一直監聽訊息
}
}
direct模式
direct生產者模式
public class DirectProduct {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection=RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel= connection.createChannel();
//創建交換機
channel.exchangeDeclare("exchange_direct", BuiltinExchangeType.DIRECT);
//8.宣告佇列（創建佇列）
/*
* 引數1：指定佇列的名稱
* 引數2：指定訊息是否持久化，一般設定為true,是否保存到磁盤當中去
* 引數3：指定是否獨占通道
* 引數4：是否自動洗掉
* 引數5：指定額外引數
*/
channel.queueDeclare("direct_queue1",true,false,false,null);
channel.queueDeclare("direct_queue2",true,false,false,null);
//9.創建訊息
String message1 = "this is add";
String message2 = "this is select";
//10.發送訊息
/*
*這段代碼使用 RabbitMQ 的 Java 客戶端庫中的 `queueBind` 方法來將一個佇列系結到一個 fanout 型別的交換機上，下面是各個引數的介紹：
1. `fanout_queue1`: 佇列名稱，表示要系結的佇列的名稱，
2. `exchange_fanout`: 交換機名稱，表示要進行系結的交換機的名稱，
3. `""`: 路由鍵，表示要使用的路由鍵，
*/
channel.queueBind("direct_queue1","exchange_direct","user.add");
channel.queueBind("direct_queue2","exchange_direct","user.select");
/*
* 引數1：指定交換機 簡單模式中使用默認交換機 指定空字串
* 引數2: 指定routingkey 簡單模式 只需要指定佇列名稱即可
* 引數3： 指定攜帶的額外的引數 null
* 引數4：要發送的訊息本身（位元組陣列）
*/
channel.basicPublish("exchange_direct", "user.add", null, message1.getBytes());
channel.basicPublish("exchange_direct", "user.select", null, message2.getBytes());
//11.關閉資源
channel.close();
connection.close();
}
}
direct消費者模式
public class DirectConsumer_1 {
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection= RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel=connection.createChannel();
//8.宣告佇列
channel.queueDeclare("direct_queue1",true,false,false,null);
//9.接受訊息
//處理訊息
DefaultConsumer consumer = new DefaultConsumer(channel){
/*
*consumerTag: 消費者標簽，用于標識特定的消費者，每個消費者都有一個唯一的標簽，它可以用于取消訂閱或標識訊息是由哪個消費者處理的，
* envelope: 信封物件，包含與訊息傳遞相關的元資料，如交換機、路由鍵、傳遞標簽等，
* properties: AMQP 的基本屬性，包含附加的訊息屬性，如訊息持久性、優先級、時間戳等，
* body: 訊息的正文，以位元組陣列的形式提供，
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("訊息本身"+new String(body,"utf-8"));
System.out.println("exchange"+envelope.getExchange());
System.out.println("RoutingKey"+envelope.getRoutingKey());
System.out.println("訊息的序號"+envelope.getDeliveryTag());
}
};
//監聽器----->觀察者設計模式
channel.basicConsume("direct_queue1",true,consumer);
//10.建議不要關閉資源 建議一直監聽訊息
}
}
Topic模式
生產模式相關代碼
public class TopicProduct {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection=RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel= connection.createChannel();
//創建交換機
channel.exchangeDeclare("exchange_topic", BuiltinExchangeType.TOPIC);
//8.宣告佇列（創建佇列）
/*
* 引數1：指定佇列的名稱
* 引數2：指定訊息是否持久化，一般設定為true,是否保存到磁盤當中去
* 引數3：指定是否獨占通道
* 引數4：是否自動洗掉
* 引數5：指定額外引數
*/
channel.queueDeclare("topic_queue1",true,false,false,null);
channel.queueDeclare("topic_queue2",true,false,false,null);
//9.創建訊息
String message1 = "this is add";
String message2 = "this is select";
String message3 = "this is update";
String message4 = "this is delete";
//10.發送訊息
/*
*這段代碼使用 RabbitMQ 的 Java 客戶端庫中的 `queueBind` 方法來將一個佇列系結到一個 fanout 型別的交換機上，下面是各個引數的介紹：
1. `fanout_queue1`: 佇列名稱，表示要系結的佇列的名稱，
2. `exchange_fanout`: 交換機名稱，表示要進行系結的交換機的名稱，
3. `""`: 路由鍵，表示要使用的路由鍵，
*/
channel.queueBind("topic_queue1","exchange_topic","user.*");
channel.queueBind("topic_queue2","exchange_topic","item.*");
/*
* 引數1：指定交換機 簡單模式中使用默認交換機 指定空字串
* 引數2: 指定routingkey 簡單模式 只需要指定佇列名稱即可
* 引數3： 指定攜帶的額外的引數 null
* 引數4：要發送的訊息本身（位元組陣列）
*/
channel.basicPublish("exchange_topic", "user.add", null, message1.getBytes());
channel.basicPublish("exchange_topic", "user.select", null, message2.getBytes());
channel.basicPublish("exchange_topic", "item.update", null, message3.getBytes());
channel.basicPublish("exchange_topic", "item.delete", null, message4.getBytes());
//11.關閉資源
channel.close();
connection.close();
}
}
消費者相關代碼
public class TopicConsumer_1 {
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection= RabbitMQConnectionUtils.getConection();
//7.創建頻道
Channel channel=connection.createChannel();
//8.宣告佇列
channel.queueDeclare("topic_queue1",true,false,false,null);
//9.接受訊息
//處理訊息
DefaultConsumer consumer = new DefaultConsumer(channel){
/*
*consumerTag: 消費者標簽，用于標識特定的消費者，每個消費者都有一個唯一的標簽，它可以用于取消訂閱或標識訊息是由哪個消費者處理的，
* envelope: 信封物件，包含與訊息傳遞相關的元資料，如交換機、路由鍵、傳遞標簽等，
* properties: AMQP 的基本屬性，包含附加的訊息屬性，如訊息持久性、優先級、時間戳等，
* body: 訊息的正文，以位元組陣列的形式提供，
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("訊息本身"+new String(body,"utf-8"));
System.out.println("exchange"+envelope.getExchange());
System.out.println("RoutingKey"+envelope.getRoutingKey());
System.out.println("訊息的序號"+envelope.getDeliveryTag());
}
};
//監聽器----->觀察者設計模式
channel.basicConsume("topic_queue1",true,consumer);
//10.建議不要關閉資源 建議一直監聽訊息
}
}
主意：
不要忘記在使用訂閱模式時修改該處的交換機分發策略
4. springboot整合RabbitMQ
本次整合需要創建2個springboot專案，一個為生產者，一個為消費者，
direct exchange(直連型交換機）
生產者相關的整合
相關依賴
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
application相關配置
spring:
#給專案來個名字
application:
name: rabbitmq-provider
#配置rabbitMq 服務器
rabbitmq:
host: localhost
port: 5672
username: chenbuting
password: root
#虛擬host 可以不設定,使用server默認host
#virtual-host: JCcccHost
配置類
@Configuration
public class DirectRabbitConfig {
//佇列 起名：TestDirectQueue
@Bean
public Queue TestDirectQueue() {
// durable:是否持久化,默認是false,持久化佇列：會被存盤在磁盤上，當訊息代理重啟時仍然存在，暫存佇列：當前連接有效
// exclusive:默認也是false，只能被當前創建的連接使用，而且當連接關閉后佇列即被洗掉，此參考優先級高于durable
// autoDelete:是否自動洗掉，當沒有生產者或者消費者使用此佇列，該佇列會自動洗掉，
// return new Queue("TestDirectQueue",true,true,false);
//一般設定一下佇列的持久化就好,其余兩個就是默認false
return new Queue("TestDirectQueue",true);
}
//Direct交換機 起名：TestDirectExchange
@Bean
DirectExchange TestDirectExchange() {
// return new DirectExchange("TestDirectExchange",true,true);
return new DirectExchange("TestDirectExchange",true,false);
}
//系結 將佇列和交換機系結, 并設定用于匹配鍵：TestDirectRouting
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
}
//單獨的 Direct 型別的交換機
@Bean
DirectExchange lonelyDirectExchange() {
return new DirectExchange("lonelyDirectExchange");
}
}
controller層
@RestController
public class SendMessageController {
@Autowired
RabbitTemplate rabbitTemplate; //使用RabbitTemplate,這提供了接收/發送等等方法
@GetMapping("/sendDirectMessage")
public String sendDirectMessage() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "https://www.cnblogs.com/chenbuting/archive/2023/07/08/test message, hello!";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String,Object> map=new HashMap<>();
map.put("messageId",messageId);
map.put("messageData",messageData);
map.put("createTime",createTime);
//將訊息攜帶系結鍵值：TestDirectRouting 發送到交換機TestDirectExchange
rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
return "ok";
}
}
消費者相關的整合
所需pom依賴
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
application相關配置
spring:
#給專案來個名字
application:
name: rabbitmq-provider
#配置rabbitMq 服務器
rabbitmq:
host: localhost
port: 5672
username: chenbuting
password: root
#虛擬host 可以不設定,使用server默認host
#virtual-host: JCcccHost
相關配置類（消費者單獨使用，可以不用添加配置，直接監聽就行，直接使用注釋監聽器監聽對應的佇列即可，配置的話，則消費者也可當做生產者的身份，這樣也就可以推送訊息了）
@Configuration
public class DirectRabbitConfig {
//佇列 起名：TestDirectQueue
@Bean
public Queue TestDirectQueue() {
return new Queue("TestDirectQueue",true);
}
//Direct交換機 起名：TestDirectExchange
@Bean
DirectExchange TestDirectExchange() {
return new DirectExchange("TestDirectExchange");
}
//系結 將佇列和交換機系結, 并設定用于匹配鍵：TestDirectRouting
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
}
}
創建訊息監聽
@Component
@RabbitListener(queues = "TestDirectQueue")//監聽的佇列名稱 TestDirectQueue
public class DirectReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("DirectReceiver消費者收到訊息 : " + testMessage.toString());
}
}
Topic Exchange主題交換機
生產者相關整合
相關配置類
@Configuration
public class TopicRabbitConfig {
//系結鍵
public final static String man = "topic.man";
public final static String woman = "topic.woman";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.man);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.woman);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//將firstQueue和topicExchange系結,而且系結的鍵值為topic.man
//這樣只要是訊息攜帶的路由鍵是topic.man,才會分發到該佇列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
}
//將secondQueue和topicExchange系結,而且系結的鍵值為用上通配路由鍵規則topic.#
// 這樣只要是訊息攜帶的路由鍵是以topic.開頭,都會分發到該佇列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
}
}
添加兩個介面，用于將訊息推送到交換機上
@GetMapping("/sendTopicMessage1")
public String sendTopicMessage1() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "https://www.cnblogs.com/chenbuting/archive/2023/07/08/message: M A N";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> manMap = new HashMap<>();
manMap.put("messageId", messageId);
manMap.put("messageData", messageData);
manMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.man", manMap);
return "ok";
}
@GetMapping("/sendTopicMessage2")
public String sendTopicMessage2() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "https://www.cnblogs.com/chenbuting/archive/2023/07/08/message: woman is all";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> womanMap = new HashMap<>();
womanMap.put("messageId", messageId);
womanMap.put("messageData", messageData);
womanMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
return "ok";
}
}
消費者相關整合
創建TopicManReceiver
@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("TopicManReceiver消費者收到訊息 : " + testMessage.toString());
}
}
創建TopicTotalReceiver
@Component
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("TopicTotalReceiver消費者收到訊息 : " + testMessage.toString());
}
}
其他類似
本文來自博客園，作者：陳步汀，轉載請注明原文鏈接：https://www.cnblogs.com/chenbuting/p/17537371.html
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/556840.html
標籤：其他
上一篇：SpringBoot與MyBatis零XML配置集成和集成測驗
下一篇：返回列表