- 导入依赖创建消息生产者创建消息消费者
2.创建消费者com.rabbitmq amqp-client5.8.0 commons-io commons-io2.6
设置队列名字
创建一个连接工厂
设置连接到MQ的账户名和密码
创建一个新的连接
创建一个信道
生成一个队列
发送一条消息
注意：消息必须是byte[]数组类型。
public class Producer {
private static final String QUEUE_NAME = "Hello World";
public static void main(String[] args) throws Exception {
// 1. 创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2. 设置连接到MQ的账户、密码和主机地址
factory.setHost("xx.xx.xx.xx");
factory.setUsername("admin");
factory.setPassword("123");
// 3. 创建一个新连接并处理异常
Connection connection = factory.newConnection();
// 4. 创建一个新信道
Channel channel = connection.createChannel();
// 5. 生成一个队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String message = "Hello Pox";
// 6. 发送一个消息
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
}
}
3.创建消费者
设置队列名字
创建一个连接工厂
设置连接到MQ的账户名和密码
创建一个新的连接
创建一个信道
消费消息
注意：参数中有两个参数，通过函数式接口创建
public class Consumer {
private static final String QUEUE_NAME = "Hello World";
public static void main(String[] args) throws Exception{
// 连接到MQ
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xx.xx.xx");
factory.setUsername("admin");
factory.setPassword("123");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 推送的消息如何进行消费的接口回调
DeliverCallback deliverCallback=(consumerTag, delivery)->{
String message= new String(delivery.getBody());
System.out.println(message);
};
// 取消 消费的一个回调接口
CancelCallback cancelCallback=(consumerTag)->{
System.out.println("消息消费被中断");
};
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
}
}
2.Work Queues - 工作模式运行RabbitMQ 服务，编写测试程序，运行生产者再运行消费者，在消费者的控制台接收到消息。
工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将d出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
2.1 过程对于来的密集消息时，先将消息封装并发送到队列中。然后再将这些消息分发出去到具体的工作线程进行执行，避免大量消息的堆积。
创建工作线程
创建消息的发送方
启动多个工作线程进行处理
消息发送方发送多条消息
2.2 详解工作线程通过轮询的方式处理消息。
将连接到MQ的过程抽取出来编写为一个工具类。
public class RabbitMQUtil {
public static Channel getChannel() throws Exception{
//1.创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置连接的地址、MQ登录的账户和密码
factory.setHost("xx.xx.xx.xx");
factory.setUsername("admin");
factory.setPassword("123");
//3.创建一个连接
Connection connection = factory.newConnection();
//4.返回信道
return connection.createChannel();
}
}
1.创建工作线程
public class Worker {
// 定义队列的名字
private static final String QUEUE_NAME = "Hello World";
public static void run() throws Exception{
// 将连接到MQ的过程抽取出来编写为一个工具类
Channel channel = RabbitMQUtil.getChannel();
// 打印当前正在处理的线程名字
System.out.println("当前正在执行的线程： "+Thread.currentThread().getName());
// 接收消息并处理
channel.basicConsume(QUEUE_NAME,true,
(consumerTag, message)->{
System.out.println("接收到的消息："+Thread.currentThread().getName()+new String(message.getBody()));
},
(consumerTag)->{
System.out.println("消息接收被取消，执行该方法...");
});
}
}
2.创建消息的发送方
public class Task {
private static final String QUEUE_NAME = "Hello World";
public static void main(String[] args) throws Exception {
// 通过工具类获取信道
Channel channel = RabbitMQUtil.getChannel();
// 发送消息，并设定参数
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
// 连续发送消息
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息发送完成："+message);
}
}
}
3.启动多个工作线程进行消息接收
这里通过创建多个线程的方式，创建多个消息的接受方
public class Client {
public static void main(String[] args) {
new Thread(()->{
try {
Worker.run();
} catch (Exception e) {
e.printStackTrace();
}
},"线程一").start();
new Thread(()->{
try {
Worker.run();
} catch (Exception e) {
e.printStackTrace();
}
},"线程二").start();
}
}
4.结果
当前正在执行的线程： 线程一 当前正在执行的线程： 线程二 接收到的消息：pool-1-thread-4AA 接收到的消息：pool-2-thread-4BB 接收到的消息：pool-1-thread-5CC 接收到的消息：pool-2-thread-5DD
3.发布确认 3.1 原理可以看到，通过轮询的方式，两个消息接收方，不断接收消息。
生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。
confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。
3.2 发布确认策略 3.2.1 开启发布确认发布确认可以确保消息的持久化，通过开启相关参数开启。
发布确认默认是没有开启的，如果要开启需要调用方法/confirm/iSelect，每当你要想使用发布确认，都需要在channel上调用该方法：channel.confirmSelect();
确认的方式为：channel.waitForConfirms();，当队列确认了消息的持久化就会返回给生产者，让其发送下一条消息。
3.2.2 单个发布确认这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:**发布速度特别的慢，**因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。
3.2.3 批量发布确认确认一条发布下一条，没有确认不会发送下一条消息。
每发送一条消息就要去发布确认一次。
上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。
原理和上面一样，但是发布确认的方式为，当发布了一定的数量的消息以后，再调用发布确认方法，查询持久化。
// 当消息达到指定的数量以后，进行发布确认
if (outstandingMessageCount == batchSize) {
// 发布确认
channel.waitFor/confirm/is();
// 归0，开始下一次的批量确认
outstandingMessageCount = 0;
}
3.2.4 异步发布确认
异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。
消息的发送方只用关心将消息发送出去，消息的发布确认交给MQ来做。对于每一条发送出去的消息，MQ都可以进行应答，应答持久化成功或失败。
通过设置监听器和一个线程安全的List来保存发送的消息。
public static void messagePublishAsyn() throws Exception{
Channel channel = RabbitMQUtil.getChannel();
// 随机创建一个队列名字
String queueName = UUID.randomUUID().toString();
// 声明队列
channel.queueDeclare(queueName, false, false, false, null);
// 开启发布确认
channel./confirm/iSelect();
ConcurrentSkipListMap outstandingConfirms = new ConcurrentSkipListMap<>();
/confirm/iCallback ackCallback = (sequenceNumber, multiple) -> {
if (multiple) {
//返回的是小于等于当前序列号的未确认消息 是一个 map
ConcurrentNavigableMap confirmed = outstanding/confirm/is.headMap(sequenceNumber, true);
//清除该部分确认消息
/confirm/ied.clear();
}else{
//只清除当前序列号的消息
outstanding/confirm/is.remove(sequenceNumber);
}
};
/confirm/iCallback nackCallback = (sequenceNumber, multiple) -> {
String message = outstanding/confirm/is.get(sequenceNumber);
System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
};
channel.add/confirm/iListener(ackCallback, nackCallback);
for (int i = 0; i < 1000; i++) {
String message = "消息" + i;
outstanding/confirm/is.put(channel.getNextPublishSeqNo(), message);
// 发送消息
channel.basicPublish("", queueName, null, message.getBytes());
}
}
4.发布、订阅模式
4.1 概述
我们假设的是工作队列背后，每个任务都恰好交付给一个消费者(工作进程)。在这一部分中，我们将做一些完全不同的事情-我们将消息传达给多个消费者。这种模式称为 ”发布/订阅”.
为了说明这种模式，我们将构建一个简单的日志系统。它将由两个程序组成:第一个程序将发出日志消息，第二个程序是消费者。
其中我们会启动两个消费者，其中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把消息打印在屏幕上，事实上第一个程序发出的日志消息将广播给所有消费者。
4.2 Exchanges 4.2.1 交换机概念对于消息的发送，先发送给交换机，再由交换机将消息发送给具体的队列，最终由队列发送给消费者。
RabbitMQ 消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。
实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。
是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。
4.2.2 交换机的类型生产者将消息推送给交换机，再由交换机进行下一步的推送：
发送给单个队列发送给多个队列丢弃消息
默认
前面演示发送消息时，第一个参数为" "就是推送到默认的交换机
直接(direct)
主题（topic)
标题（headers)
扇出（fanout)
4.3 临时队列第一个参数是交换机的名称。空字符串表示默认或无名称交换机：消息能路由发送到队列中其实是由routingKey(bindingkey)绑定 key 指定的。
每当我们连接到 Rabbit 时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列**，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接，队列将被自动删除。**
创建临时队列的方式如下:
String queueName = channel.queueDeclare().getQueue();
4.4 绑定临时队列名称是随机的、没有进行队列的持久化，当消费者断开与队列的连接，临时队列就会自动删除。
什么是 bingding 呢，binding 其实是 exchange 和 queue 之间的桥梁，它告诉我们 exchange 和那个队列进行了绑定关系。
比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定：
4.5 Fanout 4.5.1 概念将交换机和具体的路由进行绑定，通过一个路由key来区别绑定。
Fanout 这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些exchange 类型。
MQ默认自带的一些交换机：
4.5.2 测试- 创建消息的接受者1
- 声明一个交换机并定义交换机类型创建一个临时队列，及时删除
- ……将接收到的消息存储到本地
- 声明一个交换机并定义交换机类型
public class Receiver01 {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
Channel channel = RabbitMQUtil.getChannel();
// 声明一个交换机和指定类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
// 把该临时队列绑定我们的 exchange 其中
// 最后一个参数routingKey(也称之为 binding key)为绑定区分
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println("等待接收消息,把接收到的消息打印在屏幕........... ");
// 消息成功接收的处理方式
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("控制台打印接收到的消息"+message);
};
// 消息处理
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}
2.创建消息的接受者2
public class Receiver02 {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
Channel channel = RabbitMQUtil.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
// 把该临时队列绑定我们的 exchange
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println("等待接收消息,把接收到的消息写到文件........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
File file = new File("C:\work\rabbitmq_info.txt");
FileUtils.writeStringToFile(file, message, "UTF-8");
System.out.println("数据写入文件成功");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
}
}
3.创建一个消息的发送者
public class EmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitMQUtil.getChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
Scanner sc = new Scanner(System.in);
System.out.println("请输入信息");
while (sc.hasNext()) {
String message = sc.nextLine();
// 发送消息，指定具体的交换机
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
System.out.println("生产者发出消息" + message);
}
}
}
}
5.路由模式 5.1 概述结果表明，发送方发送一条消息，另外两个消费者都能接收到这条消息。
在上一节中，我们构建了一个简单的日志记录系统。我们能够向许多接收者广播日志消息。在本节我们将向其中添加一些特别的功能-比方说我们只让某个消费者订阅发布的部分消息。例如我们只把严重错误消息定向存储到日志文件(以节省磁盘空间)，同时仍然能够在控制台上打印所有日志消息。
我们再次来回顾一下什么是 bindings，绑定是交换机和队列之间的桥梁关系。也可以这么理解：队列只对它绑定的交换机的消息感兴趣。
绑定用参数：routingKey 来表示也可称该参数为 binding key，创建绑定我们用代码: channel.queueBind(queueName, EXCHANGE_NAME, “routingKey”);
绑定之后的意义由其交换类型决定。
5.2 介绍路由模式和发布、订阅模式的区别是：交换机对具体的队列的绑定是不同的，交换发送消息时，交换机可以指定发送给的队列。
上一节中的我们的日志系统将所有消息广播给所有消费者，对此我们想做一些改变，例如我们希望将日志消息写入磁盘的程序仅接收严重错误(errros)，而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，在这里我们将使用 direct 这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的routingKey 队列中去。
在上面这张图中，我们可以看到 X 绑定了两个队列，绑定类型是 direct。队列Q1 绑定键为 orange，队列 Q2 绑定键有两个:一个绑定键为 black，另一个绑定键为 green。
在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列Q1。绑定键为 blackgreen 和的消息会被发布到队列 Q2，其他消息类型的消息将被丢弃。
5.3 多重绑定交换机将消息发送给指定的队列。
当然如果 exchange 的绑定类型是direct，但是它绑定的多个队列的 key 如果都相同，在这种情况下虽然绑定类型是 direct 但是它表现的就和 fanout 有点类似了，就跟广播差不多，如上图所示。
5.4 测试- 创建消息接收者1
- 声明交换机声明队列将队列和交换机绑定并指定具体的路由key
- ……队列和交换机绑定两次，两次指定不同的路由key
public class Receiver01 {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
Channel channel = RabbitMQUtil.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
String queueName = "console";
// 声明队列
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "info");
channel.queueBind(queueName, EXCHANGE_NAME, "warning");
System.out.println("等待接收消息........... ");
// 消息的处理
DeliverCallback deliverCallback = (consumerTag, delivery) ->
{String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" 接收绑定键 :"+delivery.getEnvelope().getRoutingKey()+", 消 息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
});
}
}
2.创建消息的接受者2
3.创建消息的生产者同理上面，修改绑定的key值。
public class EmitLog {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitMQUtil.getChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
//创建多个 bindingKey
Map bindingKeyMap = new HashMap<>();
bindingKeyMap.put("info","普通 info 信息");
bindingKeyMap.put("warning","警告 warning 信息");
bindingKeyMap.put("error","错误 error 信息");
//debug 没有消费这接收这个消息 所有就丢失了
bindingKeyMap.put("debug","调试 debug 信息");
for (Map.Entry bindingKeyEntry:bindingKeyMap.entrySet()){
String bindingKey = bindingKeyEntry.getKey();
String message = bindingKeyEntry.getValue();
channel.basicPublish(EXCHANGE_NAME,bindingKey, null, message.getBytes("UTF-8"));
System.out.println("生产者发出消息:" + message);
}
}
}
}
6.主题模式 6.1 概述结果，消息将由指定的消费者接收到。
在上一个小节中，我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机，而是使用了 direct 交换机，从而有能实现有选择性地接收日志。
尽管使用 direct 交换机改进了我们的系统，但是它仍然存在局限性-比方说我们想接收的日志类型有info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时候 direct 就办不到了。这个时候就只能使用 topic 类型。
6.2 Key值要求Topic，主题模式，通过在将队列绑定到交换机时，设置由多个单词组成的key值绑定。
在交换机发送消息到具体队列时可以通过模糊发送的特性将消息发送到多个队列中，集成了发布订阅和路由模式的优点。
发送到类型是 topic 交换机的消息的 routing_key 不能随意写，必须满足一定的要求，它必须是一个单词列表，以点号分隔开。这些单词可以是任意单词，比如说：“stock.usd.nyse”, “nyse.vmw”,“quick.orange.rabbit”.这种类型的。当然这个单词列表最多不能超过 255 个字节。
其他规则：
*(星号)可以代替一个单词#(井号)可以替代零个或多个单词
6.3 小结不一定是存在的单词，只要是由字母组成，并且符合条件的就行。
主题模式集成了发布订阅模式和理由模式的优点，在实现的时候，需要指明交换机的类型为"topic"，在将队列绑定的时的key值需要满足主题模式对key值的要求。
7.发布确认-高级在生产环境中由于一些不明原因，导致 rabbitmq 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。
于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？
特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢？
7.1 实例演示（Spring Boot版本）前面我们考虑的都是的消费者不能正常的接收消息，现在考虑MQ服务器如果意外宕机时，生产者发送的消息丢失的问题。
- 添加配置交换机配置类编写消息生产者
- 编写回调接口，实现消息未正常分发的回复
spring.rabbitmq.publisher-/confirm/i-type=correlated2.交换机配置类
public class /confirm/iConfig {
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i.exchange";
public static final String /confirm/i_QUEUE_NAME = "/confirm/i.queue";
// 声明业务 Exchange
@Bean("/confirm/iExchange")
public DirectExchange /confirm/iExchange(){
return new DirectExchange(/confirm/i_EXCHANGE_NAME);
}
// 声明确认队列
@Bean("/confirm/iQueue")
public Queue /confirm/iQueue(){
return QueueBuilder.durable(/confirm/i_QUEUE_NAME).build();
}
// 声明确认队列绑定关系
@Bean
public Binding queueBinding(@Qualifier("/confirm/iQueue") Queue queue,
@Qualifier("/confirm/iExchange") DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("key1");
}
}
3.编写消息生产者
3.1 消息生产者
因为调用回调接口是在模版类内部，因此需要将编写好的回调接口注入到模版类中，这样才能生效我们编写的回调接口。
@RestController
@RequestMapping("//confirm/i")
@Slf4j
public class Producer {
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i.exchange";
@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private MyCallBack myCallBack;
// 依赖注入 rabbitTemplate 之后再设置它的回调对象
@PostConstruct
public void init(){
rabbitTemplate.set/confirm/iCallback(myCallBack);
}
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
// 指定消息 id 为 1
CorrelationData correlationData1=new CorrelationData("1");
String routingKey="key1";
rabbitTemplate.convertAndSend(
/confirm/i_EXCHANGE_NAME,
routingKey,
message+routingKey,
correlationData1);
CorrelationData correlationData2=new CorrelationData("2");
routingKey="key2";
rabbitTemplate.convertAndSend(
/confirm/i_EXCHANGE_NAME,
routingKey,
message+routingKey,
correlationData2);
// 日志
log.info("发送消息内容:{}",message);
}
}
3.2 回调接口
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate./confirm/iCallback {
@Override
public void /confirm/i(CorrelationData correlationData, boolean ack, String cause)
{String id=correlationData!=null?correlationData.getId():"";
if(ack){
log.info("交换机已经收到 id 为:{}的消息",id);
}else{
log.info("交换机还未收到 id 为:{}消息,由于原因:{}",id,cause);
}
}
}
4.编写消息消费者
@Slf4j
public class /confirm/iConsumer {
public static final String /confirm/i_QUEUE_NAME = "/confirm/i.queue";
// 指定监听的对象
@RabbitListener(queues = /confirm/i_QUEUE_NAME)
public void receiveMsg(Message message) {
String msg = new String(message.getBody());
log.info("接受到队列 /confirm/i.queue 消息:{}", msg);
}
}
7.2 消息回退结果：当消息发送给交换机，交换机因意外宕机没有接收到，生产者会收到回调消息。
在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。
那么如何让无法被路由的消息帮我想办法处理一下？
通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。
7.3 交换机备份编写相关回调接口，注入到 Rabbit 模版类中。
有了 mandatory 参数和回退消息，我们获得了对无法投递消息的感知能力，有机会在生产者的消息无法被投递时发现并处理。但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。
而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。
而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。
如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？
前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。在 RabbitMQ 中，有一种备份交换机的机制存在，可以很好的应对这个问题。
什么是备份交换机呢？
备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
前面是将没有发送成功的消息返回给生产者，备份交换机的原理是：当消息没有发送成功，那么将没有发送成功的消息发送给备份交换机，再由备份交换机发送给绑定的队列，进行消息的再一次分发。
这里绑定了两个队列，一个是正常的分发，另一个是警告队列，通知开发者，有消息没有成功发送的作用。
备份交换机的优先级比消息回退高。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）