码迷,mamicode.com
首页 > 其他好文 > 详细

Native RabbitMQ RejectMessage

时间:2020-09-17 14:11:23      阅读:17      评论:0      收藏:0      [点我收藏+]

标签:hang   world   final   交换   tag   walle   erro   int   ctc   

消费者在消费消息的时候,可以根据业务场景来选择拒绝消息,通过参数设置是否让消息再次入队。拒绝消息的指令有俩,一个是basicNack,一个是basicReject,这两个区别是后者不能批量拒绝,只能单条拒绝。代码演示如下:

一个生产者,两个正常消费者,一个拒绝消息的消费者,三个消费者共同订阅一个队列,这样生产者发布的消息会轮询被三个消费者消费,演示代码如下:

生产者

/**
 * 普通的生产者
 *
 * @author zhangjianbing
 * time 2020/09/04
 */
@SuppressWarnings("Duplicates")
public class NormalProducer {

    public static final String EXCHANGE_NAME = "reject-exchange";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.231.63.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wallet-sit");
        connectionFactory.setPassword("d6VrayFvcJyfY2Th");
        connectionFactory.setVirtualHost("wallet-sit");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        final Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false);

        String queue = "MESSAGE.CALLBACK.QUEUE";
        // 注意同时设置队列、消息持久化才显得有意义
        channel.queueDeclare(queue, false, false, false, null);

        String routeKey = "error";
        /** 将队列和交换器通过路由键绑定,表示此队列只接受error的消息 **/
        channel.queueBind(queue, EXCHANGE_NAME, routeKey);

        for (int i = 1; i < 11; i++) {
            String message = "hello_world_" + i;
            channel.basicPublish(EXCHANGE_NAME, routeKey, null, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }
        // 关闭信道和连接
        channel.close();
        connection.close();
    }

}

正常消费者A

/**
 * 普通消费者A
 *
 * @author zhangjianbing
 * time 2020/09/04
 */
@SuppressWarnings("Duplicates")
public class NormalConsumerA {

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.231.63.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wallet-sit");
        connectionFactory.setPassword("d6VrayFvcJyfY2Th");
        connectionFactory.setVirtualHost("wallet-sit");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        final Channel channel = connection.createChannel();

        String queue = "MESSAGE.CALLBACK.QUEUE";
        System.out.println("正在等待消息。。。。。。");

        /**声明一个消费者**/
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, StandardCharsets.UTF_8);
                System.out.println("NormalConsumerA收到的消息:" + message);
                // false 表示单条确认 true 表示确认tag之前所有的消息
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        /**手动确认模式**/
        channel.basicConsume(queue, false, consumer);
    }

}

正常消费者B

/**
 * 普通消费者B
 *
 * @author zhangjianbing
 * time 2020/09/04
 */
@SuppressWarnings("Duplicates")
public class NormalConsumerB {

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.231.63.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wallet-sit");
        connectionFactory.setPassword("d6VrayFvcJyfY2Th");
        connectionFactory.setVirtualHost("wallet-sit");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        final Channel channel = connection.createChannel();

        String queue = "MESSAGE.CALLBACK.QUEUE";
        System.out.println("正在等待消息。。。。。。");

        /**声明一个消费者**/
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, StandardCharsets.UTF_8);
                System.out.println("NormalConsumerB收到的消息:" + message);
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        /**手动确认模式**/
        channel.basicConsume(queue, false, consumer);
    }

}

拒绝消费者C

/**
 * 拒绝消费者
 *          当多个消费者共同消费同一个队列的时候
 *          此消费者会根据业务需求来拒绝消息,可以选择是否将消息重新入队
 *          重新入队的消息会被打上一个重新入队的标签
 *          重新入队的消息会放在队列的末尾,再次轮询投递给队列上的消费者
 *
 * 拒绝消息有两种方式:
 *          ① basicNack
 *          ② basicReject
 *          这俩唯一的不同就是basicNack有批量拒绝参数
 *
 * @author zhangjianbing
 * time 2020/09/04
 */
@SuppressWarnings("Duplicates")
public class RejectConsumer {

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.231.63.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wallet-sit");
        connectionFactory.setPassword("d6VrayFvcJyfY2Th");
        connectionFactory.setVirtualHost("wallet-sit");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        final Channel channel = connection.createChannel();

        String queue = "MESSAGE.CALLBACK.QUEUE";
        System.out.println("正在等待消息。。。。。。");

        /**声明一个消费者**/
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, StandardCharsets.UTF_8);
                System.out.println("RejectConsumer收到的消息:" + message);
                try {
                    throw new RuntimeException("RejectConsumer业务处理发生异常");
                } catch (Exception re) {
                    System.out.println(re.getMessage() + ",消息重新入队");
                    channel.basicReject(envelope.getDeliveryTag(), true);// 重新入队
                }
            }
        };

        /**手动确认模式**/
        channel.basicConsume(queue, false, consumer);
    }

}

测试结果

生产者发送10条消息,假设消息投递的顺序是A,B,C

消费者A:

正在等待消息。。。。。。
NormalConsumerA收到的消息:hello_world_1
NormalConsumerA收到的消息:hello_world_4
NormalConsumerA收到的消息:hello_world_7
NormalConsumerA收到的消息:hello_world_10
NormalConsumerA收到的消息:hello_world_9

消费者B:

正在等待消息。。。。。。
NormalConsumerB收到的消息:hello_world_2
NormalConsumerB收到的消息:hello_world_5
NormalConsumerB收到的消息:hello_world_8
NormalConsumerB收到的消息:hello_world_3
NormalConsumerB收到的消息:hello_world_6

消费者C:

正在等待消息。。。。。。
RejectConsumer收到的消息:hello_world_3
RejectConsumer业务处理发生异常,消息重新入队
RejectConsumer收到的消息:hello_world_6
RejectConsumer业务处理发生异常,消息重新入队
RejectConsumer收到的消息:hello_world_9
RejectConsumer业务处理发生异常,消息重新入队
RejectConsumer收到的消息:hello_world_6
RejectConsumer业务处理发生异常,消息重新入队

消费者C的结果中显示消息6被拒绝了两次,说明被拒绝的消息重新入队后还可以再次被C消费,只是再拒绝一次。

多个消费者共同消费一个队列,消息首先投递给谁,取决于哪个消费者先启动,这是轮询算法的问题。

Native RabbitMQ RejectMessage

标签:hang   world   final   交换   tag   walle   erro   int   ctc   

原文地址:https://www.cnblogs.com/zhangjianbing/p/13618231.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有
迷上了代码!