一、RabbitMQ之简单模式
模式说明
生产者代码实现
基本步骤
消费者代码实现
基本步骤
官方文档
二、RabbitMQ之WorkQueues模式
模式说明
生产者代码实现
2个消费者代码实现
官方文档
三、RabbitMQ之Pub/Sub模式
模式说明
生产者代码实现
2个消费者代码实现
发布订阅模式与工作队列模式的区别
官方文档
四、RabbitMQ之Routing模式
模式说明
需求描述
生产者代码实现
2个消费者代码实现
官方文档
五、RabbitMQ之Topics模式
模式说明
生产者代码实现
2个消费者代码实现
官方文档
一、RabbitMQ之简单模式
模式说明
P：生产者，也就是要发送消息的程序
C：消费者：消息的接收者，会一直等待消息到来
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息
生产者代码实现
基本步骤
1. 创建连接工厂ConnectionFactory
2. 设置工厂的参数
3. 创建连接 Connection
4. 创建管道 Channel
5. 简单模式中没有交换机exchange，所以不用创建（RabbitMQ会使用默认的交换机！）
6. 创建队列 queue
7. 设置发送内容，使用channal.basicPublish()发送
8. 释放资源
public class Producer_01_HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection -> newConnection()
Connection connection = factory.newConnection();
// 4.创建Channel -> createChannel()
Channel channel = connection.createChannel();
// 5.(最简单的模型没有exchange)创建queue
channel.queueDeclare("hello_word666",false,false,false,null);
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/*
queue: 队列名称
durable: 是否持久化，即mq重启后还在！
exclusive: 是否独占 -> 只能有一个消费者监听这个队列
当Connection关闭时，是否删除队列
autoDelete: 是否自动删除 -> 当没有消费者时，就自动删除
arguments: 参数
*/
String body = "Hello RabbitMQ........";
// 6.发送消息
channel.basicPublish("","hello_word",null,body.getBytes());
//basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
/*
exchange: 交换机名称，在简单模式下会使用默认的，"" -> 表示默认
routingKey: 路由名称
props: 配置信息
body: 发送的信息数据
*/
// 7.关闭连接对象
//channel.close();
//connection.close();
}
}
消费者代码实现
基本步骤
1. 创建连接工厂ConnectionFactory
2. 设置工厂的参数
3. 创建连接 Connection
4. 创建管道 Channel
=====================================================================
前四步代码基本是一致的，需要注意的是生产者与消费者的Channel是不同Connection中的！不是同一个对象。如下图所示：
5. 最简单的模型没有交换机exchange，所以此处RabbitMQ会使用默认的交换机
6. 接收消息，有一个回调方法 channel.basicConsume()
public class Consumer_01_HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
// 5.(最简单的模型没有exchange)创建queue
channel.queueDeclare("hello_word",true,false,false,null);
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumerTag:"+consumerTag);
System.out.println("envelope:"+envelope);
System.out.println("RoutingKey:"+envelope.getRoutingKey());
System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
channel.basicConsume("hello_word",false,consumer);
}
}
官方文档
RabbitMQ tutorial - "Hello World!" — RabbitMQ
https://www.rabbitmq.com/tutorials/tutorial-one-java.html
二、RabbitMQ之WorkQueues模式
模式说明
Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
生产者代码实现
public class Producer_02_WorkQueues {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection -> newConnection()
Connection connection = factory.newConnection();
// 4.创建Channel -> createChannel()
Channel channel = connection.createChannel();
// 5.(最简单的模型没有exchange)创建queue
channel.queueDeclare("work_queues",true,false,false,null);
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/*
queue: 队列名称
durable: 是否持久化，即mq重启后还在！
exclusive: 是否独占 -> 只能有一个消费者监听这个队列
当Connection关闭时，是否删除队列
autoDelete: 是否自动删除 -> 当没有消费者时，就自动删除
arguments: 参数
*/
for (int i = 0; i < 100; i++) {
String body = i + "Hello RabbitMQ........";
// 6.发送消息
channel.basicPublish("","work_queues",null,body.getBytes());
}
//basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
/*
exchange: 交换机名称，在简单模式下会使用默认的，"" -> 表示默认
routingKey: 路由名称
props: 配置信息
body: 发送的信息数据
*/
// 7.关闭连接对象
//channel.close();
//connection.close();
}
}2个消费者代码实现
public class Consumer_02_WorkQueues01 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
// 5.(最简单的模型没有exchange)创建queue
channel.queueDeclare("work_queues",true,false,false,null);
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("body:"+new String(body));
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume("work_queues",false,consumer);
}
}
public class Consumer_02_WorkQueues02 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
// 5.(最简单的模型没有exchange)创建queue
channel.queueDeclare("work_queues",true,false,false,null);
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body:"+new String(body));
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
channel.basicConsume("work_queues",false,consumer);
}
}
该模式有一个地方需要尤其注意，官方文档上面也给出了说明——轮循机制调度
主要就是下面的内容：
简单的来说，就是：channel.basicConsume("work_queues",false,consumer);
第二个参数如果是 false：表示若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...（触发重复消费！！！）但是保证了消息不会丢。
如何避免会在后续的文章介绍。
该模式的分配还是一种公平分配
官方文档
RabbitMQ tutorial - Work Queues — RabbitMQhttps://www.rabbitmq.com/tutorials/tutorial-two-python.html
三、RabbitMQ之Pub/Sub模式
模式说明
P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接收者，会一直等待消息到来
Queue：消息队列，接收消息、缓存消息
Exchange：交换机（X）一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、 递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
➢ Fanout：广播，将消息交给所有绑定到交换机的队列
➢ Direct：定向，把消息交给符合指定routing key 的队列
➢ Topic（常用）：通配符，把消息交给符合routing pattern（路由模式）的队列
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！
生产者代码实现
与之前的步骤相比，多了创建交换机和绑定交换机与队列的操作
public class Producer_03_PubSub {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection -> newConnection()
Connection connection = factory.newConnection();
// 4.创建Channel -> createChannel()
Channel channel = connection.createChannel();
// *5.创建交换机
/*
exchangeDeclare(String exchange, -- 交换机的名称
String type, -- 交换机的类型，4种
枚举(direct,fanout,topic,headers)
boolean durable, -- 持久化
boolean autoDelete, -- 自动删除
boolean internal, -- 内部使用，基本是false
Map<String, Object> arguments) -- 参数
*/
String exchangeName = "test_fanout";
channel.exchangeDeclare(exchangeName,
BuiltinExchangeType.FANOUT,
true, false, false, null);
// 6.创建队列 2个
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
// 7.绑定队列与交换机
channel.queueBind(queue1Name,exchangeName,"");
channel.queueBind(queue2Name,exchangeName,"");
// 8.发送消息
String body = "日志信息：张三调用了findAll()方法...日志级别：info...";
channel.basicPublish(exchangeName,"",null,body.getBytes());
// 9.释放资源
channel.close();
connection.close();
}
}2个消费者代码实现
public class Consumer_03_PubSub01 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("将日志信息打印到控制台...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue1Name,true,consumer);
}
}
public class Consumer_03_PubSub02 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("将日志信息保存到数据库...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue2Name,true,consumer);
}
}
两个不一样的系统，对同一条消息做不一样的处理
发布订阅模式与工作队列模式的区别
（1）工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机
（2）发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用 默认交换机)
（3）发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机
官方文档
RabbitMQ tutorial - Publish/Subscribe — RabbitMQhttps://www.rabbitmq.com/tutorials/tutorial-three-java.html
四、RabbitMQ之Routing模式
模式说明
队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息
需求描述
P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息
生产者代码实现
public class Producer_04_Routing {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection -> newConnection()
Connection connection = factory.newConnection();
// 4.创建Channel -> createChannel()
Channel channel = connection.createChannel();
// *5.创建交换机
/*
exchangeDeclare(String exchange, -- 交换机的名称
String type, -- 交换机的类型，4种
枚举(direct,fanout,topic,headers)
boolean durable, -- 持久化
boolean autoDelete, -- 自动删除
boolean internal, -- 内部使用，一半false
Map<String, Object> arguments) -- 参数
*/
String exchangeName = "test_direct";
channel.exchangeDeclare(exchangeName,
BuiltinExchangeType.DIRECT,
true, false, false, null);
// 6.创建队列 2个
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
// 7.绑定队列与交换机
channel.queueBind(queue1Name,exchangeName,"error");
channel.queueBind(queue2Name,exchangeName,"info");
channel.queueBind(queue2Name,exchangeName,"warning");
channel.queueBind(queue2Name,exchangeName,"error");
// 8.发送消息
//String body = "日志信息：张三调用了findAll()方法...日志级别：info...";
//channel.basicPublish(exchangeName,"info",null,body.getBytes());
String body = "日志信息：张三调用了delete()方法...日志级别：delete...";
channel.basicPublish(exchangeName,"info",null,body.getBytes());
// 9.释放资源
channel.close();
connection.close();
}
}在生产者中指定了，路由规则：
error 级别的信息会发给test_direct_queue1队列
info、warning、error级别的信息会发给test_direct_queue2队列
2个消费者代码实现
public class Consumer_04_Routing01 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("将日志信息打印到控制台...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue1Name,true,consumer);
}
}
public class Consumer_04_Routing02 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("将日志信息保存到数据库...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue2Name,true,consumer);
}
}
总的来说就一句话：
Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。
官方文档
RabbitMQ tutorial - Routing — RabbitMQhttps://www.rabbitmq.com/tutorials/tutorial-four-java.html
五、RabbitMQ之Topics模式
模式说明
通配符规则：# 匹配一个或多个词，* 匹配有且仅有1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert
红色 Queue：绑定的是 usa.# ，因此凡是以 usa.开头的 routing key 都会被匹配到
黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配
Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
生产者代码实现
public class Producer_05_Topics {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection -> newConnection()
Connection connection = factory.newConnection();
// 4.创建Channel -> createChannel()
Channel channel = connection.createChannel();
// *5.创建交换机
/*
exchangeDeclare(String exchange, -- 交换机的名称
String type, -- 交换机的类型，4种
枚举(direct,fanout,topic,headers)
boolean durable, -- 持久化
boolean autoDelete, -- 自动删除
boolean internal, -- 内部使用，一半false
Map<String, Object> arguments) -- 参数
*/
// direct ,fanout ,topic , headers
String exchangeName = "test_topics";
channel.exchangeDeclare(exchangeName,
BuiltinExchangeType.TOPIC,
true, false, false, null);
// 6.创建队列 2个
String queue1Name = "test_topics_queue1";
String queue2Name = "test_topics_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
// 7.绑定队列与交换机
// * 代表一个词
// # 代表0个和多个单词
// routing key 系统名称.日志级别
// 所有error级别的日志 与 order系统的日志存入数据库
channel.queueBind(queue1Name,exchangeName,"#.error");
channel.queueBind(queue1Name,exchangeName,"order.*");
// 所有的信息打印控制台上
channel.queueBind(queue2Name,exchangeName,"*.*");
// 8.发送消息
String body = "1234567890";
channel.basicPublish(exchangeName,"test.info",null,body.getBytes());
// 9.释放资源
channel.close();
connection.close();
}
}2个消费者代码实现
public class Consumer_05_Topics01 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue1Name = "test_topics_queue1";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("database...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue1Name,true,consumer);
}
}public class Consumer_05_Topics02 {
public static void main(String[] args) throws IOException, TimeoutException {
// 1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置参数
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/HarmonyOS");
factory.setUsername("Harmony");
factory.setPassword("888888");
// 3.创建连接 Connection
Connection connection = factory.newConnection();
// 4.创建Channel
// 生产者 与 消费者的 Channel是不同Connection中的！不是同一个对象
Channel channel = connection.createChannel();
String queue2Name = "test_topics_queue2";
// 6.接受消息
Consumer consumer = new DefaultConsumer(channel) {
/*
回调方法：当收到消息时，会自动执行该方法
consumerTag: 标识
envelope: 获取一些信息，如：交换机、路由key
properties: 配置信息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
System.out.println("body:"+new String(body));
System.out.println("control...");
}
};
//basicConsume(String queue, boolean autoAck, Consumer callback)
/*
queue: 队列名称
autoAck: 是否自动确认
callback: 回调对象
*/
// 自动确认 --> false: 若一个消费者死了，之前分配到其身上的所有资源会重新分配，交给其他消费者...
channel.basicConsume(queue2Name,true,consumer);
}
}
总述：topics模式比routing模式要更加灵活，笼统的说就是routing模式加上通配符
官方文档
RabbitMQ tutorial - Routing — RabbitMQhttps://www.rabbitmq.com/tutorials/tutorial-four-java.html
相关文章
暂无评论...
致力于称为最好的程序员导航网站