程序员经验分享-hwhale

快速入门RabbitMQ（详细）第二篇：RabbitMQ五种工作模式的使用及总结

2023-05-16

4. RabbitMQ工作模式

4.1. Work queues工作队列模式

在这里插入图片描述
Work Queues 与入门程序的 简单模式 相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

4.1.2. 代码

Work Queues 与入门程序的 简单模式 的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试。

1）生产者

import com.lxs.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
for (int i = 1; i <= 30; i++) {
// 发送信息
String message = "你好；小兔子！work模式--" + i;
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息：" + message);
}
// 关闭资源
channel.close();
connection.close();
}
}

2）消费者1和2 (都一样)

import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 声明（创建）队列
/**
3）消费者2
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//一次只能接收并处理一个消息
channel.basicQos(1);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
Thread.sleep(1000);
//确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.QUEUE_NAME, false, consumer);
}
}

4.1.3. 测试

在这里插入图片描述

4.1.4. 小结

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。

4.2. 订阅模式概述

在这里插入图片描述
前面2个案例中，只有3个角色：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分
    而在订阅模型中，多了一个exchange角色，而且过程略有变化：
  • P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  • C：消费者，消息的接受者，会一直等待消息到来。
  • Queue：消息队列，接收消息、缓存消息。
  • Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息
    • Fanout：广播，将消息交给所有绑定到交换机的队列
    • Direct：定向，把消息交给符合指定routing key 的队列
    • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
  • Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

4.3. Publish/Subscribe发布与订阅模式

4.3.1. 模式说明

发布订阅模式： 1、每个消费者监听自己的队列。 2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收 到消息
在这里插入图片描述

4.3.2. 代码

1）生产者

package com.lxs.rabbitmq.ps;
import com.lxs.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 发布与订阅使用的交换机类型为：fanout
*/
public class Producer {
//交换机名称
static final String FANOUT_EXCHAGE = "fanout_exchange";
//队列名称
static final String FANOUT_QUEUE_1 = "fanout_queue_1";
//队列名称
static final String FANOUT_QUEUE_2 = "fanout_queue_2";
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
/**
* 声明交换机
* 参数1：交换机名称
* 参数2：交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
//队列绑定交换机
channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
for (int i = 1; i <= 10; i++) {
// 发送信息
String message = "你好；小兔子！发布订阅模式--" + i;
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
System.out.println("已发送消息：" + message);
}
// 关闭资源
channel.close();
connection.close();
}
}

2）消费者1

package com.lxs.rabbitmq.ps;
import com.lxs.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
}
}

3）消费者2

public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, "");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
}
}

4.3.3. 测试

启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达广播的效果

4.3.4. 小结

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。

发布订阅模式与工作队列模式的区别

  • 1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
  • 2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换
    机)。
  • 3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 。

4.4. Routing路由模式

4.4.1. 模式说明

路由模式特点：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 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 的消息

4.4.2. 代码

在这里插入图片描述
在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key

1）生产者

/**
* 路由模式的交换机类型为：direct
*/
public class Producer {
//交换机名称
static final String DIRECT_EXCHAGE = "direct_exchange";
//队列名称
static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
//队列名称
static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
/**
* 声明交换机
* 参数1：交换机名称
* 参数2：交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
//队列绑定交换机
channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
// 发送信息
String message = "新增了商品。路由模式；routing key 为 insert " ;
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "修改了商品。路由模式；routing key 为 update" ;
/**
* 参数1：交换机名称，如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 关闭资源
channel.close();
connection.close();
}
}

2）消费者1

public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer);
}
}

3）消费者2

public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer);
}
}

4.4.3. 测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果。
在这里插入图片描述

4.4.4. 小结

Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

4.5. Topics通配符模式

4.5.1. 模式说明

在这里插入图片描述
举例：
item.# ：能够匹配 item.insert.abc 或者 item.insert
item.* ：只能匹配 item.insert
在这里插入图片描述
在这里插入图片描述
图解：

  • 红色Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到
  • 黄色Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

4.5.2. 代码

在这里插入图片描述

1）生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.insert 、 item.update 、 item.delete ：

/**
* 通配符Topic的交换机类型为：topic
*/
public class Producer {
//交换机名称
static final String TOPIC_EXCHAGE = "topic_exchange";
//队列名称
static final String TOPIC_QUEUE_1 = "topic_queue_1";
//队列名称
static final String TOPIC_QUEUE_2 = "topic_queue_2";
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
/**
* 声明交换机
* 参数1：交换机名称
* 参数2：交换机类型，fanout、topic、topic、headers
*/
channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
// 发送信息
String message = "新增了商品。Topic模式；routing key 为 item.insert " ;
channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "修改了商品。Topic模式；routing key 为 item.update" ;
channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "删除了商品。Topic模式；routing key 为 item.delete" ;
channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 关闭资源
channel.close();
connection.close();
}
}

2）消费者1

接收两种类型的消息：更新商品和删除商品

public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");
channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
}
}

3）消费者2:

接收所有类型的消息：新增商品，更新商品和删除商品。

public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
// 声明（创建）队列
/**
* 参数1：队列名称
* 参数2：是否定义持久化队列
* 参数3：是否独占本次连接
* 参数4：是否在不使用的时候自动删除队列
* 参数5：队列其它参数
*/
channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消
息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//收到的消息
System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
/**
* 参数1：队列名称
* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，
设置为false则需要手动确认
* 参数3：消息接收到后回调
*/
channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer);
}
}

4.5.3. 测试

启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果；并且这些routing key可以使用通配符
在这里插入图片描述

4.5.4. 小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

4.6. 模式总结

RabbitMQ工作模式：

  • 1、简单模式 : 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
  • 2、工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
  • 3、发布订阅模式 P/S 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
  • 4、路由模式 Routing 需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
  • 5、通配符模式 Topic 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routingkey，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

快速入门RabbitMQ（详细）第二篇：RabbitMQ五种工作模式的使用及总结 的相关文章

随机推荐

  • 中国计算机设计大赛作品（附代码与设计书，答辩PPT）

    1 主界面 2 目录界面 3 离散信号的产生 4 离散信号的基本变换及运算 变换 运算 5 离散信号的卷积运算 6 离散信号的卷积运算 7 系统稳定性分析 8 傅立叶变换 9 FIR滤波器设计 10 IIR滤波器设计 11 语音去噪处理 录

  • 更改pip镜像源的多种方法

    目前可用的pip国内镜像源有下面这些 xff1a 阿里云 http mirrors aliyun com pypi simple 中国科技大学 https pypi mirrors ustc edu cn simple 豆瓣 http py

  • 详细介绍MATLAB导入文本文件、excel等数据文件

    一 通过导入工具导入 选择导入工具 选择文件 二 通过函数uiimport导入 xff08 推荐 xff09 使用uiimport导入数据比较方便 xff0c 输入后会自动弹出对话框选择文件或者剪贴板内容 进阶命令如下 xff1a uiim

  • 机器学习（一）：概述

    机器学习近年来发展迅速 xff0c 那什么是机器学习呢 xff1f 其实机器学习在我们每天的生活 工作中都随处可见机器学习的应用 比如你每天打开手机 xff0c 无论是打开短视频软件 xff0c 还是逛购物软件 xff0c 这些里面都是包含

  • 机器学习（二）：聚类算法1——K-means算法

    Kmeans是一种经典的聚类算法 xff0c 所谓聚类 xff0c 是指在没有给出目标的情况下 xff0c 将样本根据某种关系分为某几类 那在kmeans中 xff0c 是根据样本点间的距离 xff0c 将样本n分为k个类 K means实

  • 调制中的归一化因子如何计算？

    归一化因子的作用 xff1a 将功率 xff08 能量 xff09 进行归一化处理 添加功率归一化因子 xff0c 目的在于使得不同调制方式 xff08 或者说对于所有映射方式 xff09 都能够取得相同的平均功率 归一化因子 计算公式 x

  • 画图软件origin-柱状图断点设置

    OriginLab为中国学生提供了 免费半年正版中文 OriginPro xff1b 注册完毕后登录学校邮箱 xff0c 请根据邮件提示完成安装注册即可 xff0c 然后会发送安装教程给你 下载 xff0c 并安装典型使用方法 2 1 柱状

  • plot无法画图---已解决

    项目场景 xff1a plot画图 BUG两则 记录小bug xff1a 问题描述 用vscode开发中 xff0c 需要进行plot输出 xff0c 之前一直没问题 xff0c 近日plot无输出 xff0c 程序直接结束 xff0c 不

  • 条件判断中 两个等号（==）和三个等号（===）的区别

    1 两个等号 61 61 是抽象相等运算符 xff1b 三个等号 61 61 61 是 严格相等运算符 2 两个等号 61 61 运算符是在进行必要的类型转换后 xff0c 再比较 xff1b 如果比较的值 xff0c 一个是字符串 xff

  • Windows11连接共享打印机指定的网络名不再可用

    连接的设备是Windows11连接的Windows10打印机 一 共享网络打印机 xff08 附上共享打印机的教程 xff09 共享打印机官网教程 二 共享后指定的网络名不再可用 网上搜索后 xff0c 大体的步骤就是检查几个服务是否打开

  • 「计网四」网络层（上篇）

    文章目录 一 网络层概述二 网络层提供的两种服务三 IPv43 1 IPv4地址概述3 2 分类编址的IPv4地址3 3 划分子网的IPv4地址3 4 无分类编址的IPv4地址3 5 IPv4地址的应用规划 四 IP数据报的发送和转发过程五

  • Navicat连接MySQL出错：1251 - Client does not support authentication protocol requested by server……

    1251 Client does not support authentication protocol requested by server 游海东的技术专栏 CSDN博客 Navicat连接不上MySQL8 0问题的解决方法 终于对了

  • Huawei2288H V5 服务器重做RAID0

    Huawei服务器v5版本重做RAID0 xff1a 因为环境需求需要对三台Huawei服务器重做RAID xff0c 但是试过了百度的ctrl 43 C ctrl 43 H ctrl 43 A都是进 不去 xff0c 和网上百度的图片对比

  • 大数据之（一）Mapreduce

    MapReduce xff08 一 xff09 1 Mapreduce概述1 1定义1 2优缺点1 3核心思想1 4MapReduce进程1 5常用数据序列化类型1 6MapReduce编程规范1 7WordCount案例实操 2 Hado

  • 在Docker Debian容器中安装ps、top等命令

    有些debian镜像默认没有包括进程管理相关工具 xff0c 在实际使用时可能有些麻烦 xff0c 如果需要也可以自己安装 xff0c 使用如下命令 span class token function apt get span update

  • LCD液晶屏没有点亮，为什么会自动出现笔段、走线、乱码等现象？

    一般LCD段码液晶屏在不触碰的情况下 xff0c 是不会产生这种自动出现笔段 走线 乱码等现象 只有在物体或手碰到液晶屏的时候才会或多或少的出现这种问题 xff0c 而造成这种现象的是 静电 静电是一种处于静止状态的电荷 在干燥和多风的秋天

  • 用Java解决牛客网华为机试题目HJ9

    题目描述 输入一个int型整数 xff0c 按照从右向左的阅读顺序 xff0c 返回一个不含重复数字的新的整数 输入描述 输入一个int型整数 输出描述 按照从右向左的阅读顺序 xff0c 返回一个不含重复数字的新的整数 import ja

  • 关于word中的插件Grammarly使用

    Grammarly插件嵌入到word中 xff0c 可以帮助检查语法错误 没有账号的话注册 1 xff09 下载 链接 xff1a https www grammarly com office addin xff08 第二个 xff09 注

  • Dev C++ 5.11版 「时代 の 眼泪」安装指南

    介绍 Dev C 43 43 是一款轻量化的 C C 43 43 桌面 IDE xff0c 由于不需要额外配置 GCC 即开即用的特点 xff0c 因此常常用于 C C 43 43 语言基础的学习 目前官方最新版为 Dev C 43 43

  • 快速入门RabbitMQ（详细）第二篇：RabbitMQ五种工作模式的使用及总结

    4 RabbitMQ工作模式 4 1 Work queues工作队列模式 Work Queues 与入门程序的 简单模式 相比 xff0c 多了一个或一些消费端 xff0c 多个消费端共同消费同一个队列中的消息 应用场景 xff1a 对于

热门标签