RabbitMQ-死信队列

  • 1.概念:无法被消费的消息,若原队列没有后续操作,就成为死信

  • 2.触发条件
    消息被拒(basic.reject或basic.nack)
    消息TTL(Time To Live)过期
    队列达到最大长度

  • 3.代码(示例模拟消息TTL过期,详情都在代码注释里)

信道:

public class RabbitMqUtils {
    //得到一个连接的 channel
    public static Channel getChannel() throws Exception{
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setUsername("admin");
        factory.setPassword("admin");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

生产者:

public class Producer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) {
        try{
            Channel channel = RabbitMqUtils.getChannel();
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            // 设置过期时间
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
            for (int i = 0; i < 10; i++) {
                String message="info"+i;
                // param 1:交换机  param 2:路由键  param 4:消息体
                channel.basicPublish(NORMAL_EXCHANGE,"normal-routing-key",properties ,message.getBytes(StandardCharsets.UTF_8));
            }
        }catch (Exception e){
            System.out.println(e);
        }
    }
}

正常消费者:

public class Consumer01 {
    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通队列
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信队列
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // normal 队列
        // 在正常队列声明死信队列参数(死信交换机、路由键)
        HashMap<String, Object> hashMap = new HashMap<>(2);
        hashMap.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        hashMap.put("x-dead-letter-routing-key","dead-routing-key");

        channel.queueDeclare(NORMAL_QUEUE,false,false,false,hashMap);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"normal-routing-key");

        // 死信队列
        // param2:durable  param3:exclusive(排他)  param4:autoDelete   param5:属性参数
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"dead-routing-key");

        System.out.println("等待接收消息........... ");

        // deliverCallback: 当一个消息发送过来后的回调接口
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Consumer01 接收到消息"+message);
        };
        // cancelCallback:当一个消费者取消订阅时的回调接口;取消消费者订阅队列时除了使用{@link Channel#basicCancel}之外的所有方式都会调用该回调方法
        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback ->{});
    }
}

死信队列:

public class Consumer02 {
    // 死信队列
    public static final String DEAD_QUEUE = "dead-queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("等待消费私信队列....");
        DeliverCallback deliverCallback = (consumerTag,message) -> {
            System.out.println("Consumer02 接收死信队列的消息" + message);
        };
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,consumerTag ->{});
    }
}
  • 5.关于信道
    几乎所有操作都在channel中进行,channel是进行消息读写的通道。客户端可以建立多个channel,每个channel表示一个会话任务。
    Q:为什么使用channel,而不使用TCP连接?
    A:对于OS来说,建立和TCP连接是有代价的,频繁的建立关闭TCP连接对于系统的性能有很大的影响,而且TCP的连接数也有限制,这也限制了系统处理高并发的能力。但是,在TCP连接中建立Channel是没有上述代价的。对于Producer或者Consumer来说,可以并发的使用多个Channel进行Publish或者Receive。有实验表明,1s的数据可以Publish10K的数据包。当然对于不同的硬件环境,不同的数据包大小这个数据肯定不一样,但是我只想说明,对于普通的Consumer或者Producer来说,这已经足够了。

代码地址(click here)

上一篇:MySQL常用数据库小结


下一篇:不太好用canvas 版本