14-RabbitMQ高级特性-Return返回消息

彼岸舞

发布于 2022-10-06 08:39:03

4080

发布于 2022-10-06 08:39:03

文章被收录于专栏：java开发的那点事

Return返回消息

Return消息机制

Return Listener 用于处理一些不可路由的消息
我们的消息生产者,通过指定一个Exchange和RouteKey, 把消息送达到某一个队列中去,然后我们的消费者监听队列,进行消费处理操作
但是在某些情况下, 我们在发送消息的时候, 当前的Exchange不存在, 或者指定的路由key, 路由不到, 这个时候我们就需要监听这些不可达的消息, 就需要使用Return Listener
在基础API中有一个关键的配置项
- Mandatory; 如果为true, 则监听器会接收到路由不可达的消息, 然后进行后续处理, 如果为false, 那么broker端会自动删除该消息

Return消息机制流程

Return消息机制代码实现

消费者

package com.dance.redis.mq.rabbit.returnlistener;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
 
public class Receiver4ReturnListener {
 
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        String exchangeName = "test_returnlistener_exchange";
        String queueName = "test_returnlistener_queue";
        String routingKey = "return.#";
        RabbitMQHelper.exchangeDeclare(channel,exchangeName,RabbitMQHelper.EXCHANGE_TYPE_TOPIC);
        RabbitMQHelper.queueDeclare(channel,queueName);
        channel.queueBind(queueName, exchangeName, routingKey);
 
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException {
                System.out.println("receive message:" + new String(body) + ", RoutingKey: " + envelope.getRoutingKey());
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(queueName, false, consumer);
        //等待回调函数执行完毕之后，关闭资源。
        TimeUnit.SECONDS.sleep(50);
        channel.close();
        RabbitMQHelper.closeConnection();
    }
}

生产者

package com.dance.redis.mq.rabbit.returnlistener;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ReturnListener;

import java.io.IOException;

public class Sender4ReturnListener {


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        String exchangeName = "test_returnlistener_exchange";
        String routingKey1 = "abcd.save";
        String routingKey2 = "return.save";
        String routingKey3 = "return.delete.abc";
        channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
            System.out.println("**************handleReturn**********");
            System.out.println("replyCode : " + replyCode);
            System.out.println("replyText : " + replyText);
            System.out.println("exchange  : " + exchange);
            System.out.println("routingKey: " + routingKey);
            System.out.println("body      : " + new String(body));
        });
        String msg = "Hello World RabbitMQ 4 Return Listener Message ...";
        boolean mandatory = true;
        channel.basicPublish(exchangeName, routingKey1, mandatory, null, msg.getBytes());
        channel.basicPublish(exchangeName, routingKey2, mandatory, null, msg.getBytes());
        channel.basicPublish(exchangeName, routingKey3, mandatory, null, msg.getBytes());
    }
}