• RibbitMQ学习笔记之死信队列



    总目录如下:
    1. RibbitMQ学习笔记之MQ 的相关概念
    2. RibbitMQ学习笔记之MQ练习
    3. RibbitMQ学习笔记之MQ练习
    4. RibbitMQ学习笔记之MQ发布确认
    5. RibbitMQ学习笔记之交换机实战
    6. RibbitMQ学习笔记之死信队列
    7. RibbitMQ学习笔记延迟队列
    8. RibbitMQ学习笔记之发布确认高级
    9. RibbitMQ学习笔记之 RabbitMQ 其他知识点扩展
    10. RibbitMQ学习笔记之RabbitMQ 集群


    6. 死信队列

    6.1. 死信的概念

    先从概念解释上搞清楚这个定义,死信,顾名思义就是无法被消费的消息,字面意思可以这样理解,一般来说,producer 将消息投递到 broker 或者直接到queue 里了,consumer 从 queue 取出消息进行消费,但某些时候由于特定的原因导致 queue 中的某些消息无法被消费,这样的消息如果没有后续的处理,就变成了死信,有死信自然就有了死信队列。

    应用场景:为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

    6.2. 死信的来源

    消息 TTL 过期 消息存活时间过期

    队列达到最大长度(队列满了,无法再添加数据到 mq 中)

    消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false. 拒绝应答 不放在队列中 也会在死信中

    6.3. 死信实战

    6.3.1. 代码架构图

    在这里插入图片描述

    6.3.2. 消息 TTL 过期

    生产者代码

    public class Producer {
     
    private static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] argv) throws Exception {
     
    try (Channel channel = RabbitMqUtils.getChannel()) { channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     
    
    
    //设置消息的 TTL 时间 newAMQP.BasicProperties().builder().expiration("10000").build(); //该信息是用作演示队列个数限制  for (int i = 1; i <11 ; i++) {  String message="info"+i;  channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());  
     
    System.out.println("生产者发送消息:"+message);
     
    }
     
    }
    }
    }
     
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 死信队列  生产者
     */
    public class Producer {
    
        //普通交换机 名称
        public static final String NORMAL_EXCHANGE="normal_exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Channel channel = RabbitMqutils.getChannel();
           //死信消息 设置TTl  单位是ms 100000ms=10s
            AMQP.BasicProperties basicProperties =
                    new AMQP.BasicProperties()
                            .builder().expiration("1000").build();
    
            for (int i = 0; i < 11; i++) {
                String message="info"+i;
    
                channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
    
            }
    
    
    
    
    
        }
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38

    消费者 C1 代码(启动之后关闭该消费者 模拟其接收不到消息)

    public class Consumer01 {
    //普通交换机名称
     
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
     
    private static final String DEAD_EXCHANGE = "dead_exchange"; public static void main(String[] argv) throws Exception {
    Channel channel = RabbitUtils.getChannel();
    
    
    //声明死信和普通交换机 类型为 direct channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     
    //声明死信队列
     
    String deadQueue = "dead-queue"; 
    channel.queueDeclare(deadQueue, false, false, false, null);
     
    //死信队列绑定死信交换机与 routingkey channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
     
    //正常队列绑定死信队列信息
     
    Map<String, Object> params = new HashMap<>();
     
    //正常队列设置死信交换机 参数 key 是固定值 params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
     
    //正常队列设置死信 routing-key 参数 key 是固定值 params.put("x-dead-letter-routing-key", "lisi");
     
    String normalQueue = "normal-queue";
    channel.queueDeclare(normalQueue, false, false, false, params); 
    channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
     
    System.out.println("等待接收消息 ........... ");
     
    DeliverCallback deliverCallback = (consumerTag, delivery) -> 
    { String message = new String(delivery.getBody(), "UTF-8");
     
    System.out.println("Consumer01 接收到消息"+message);
     
    };
     
    channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> { });
     
    }
     
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
     package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DeliverCallback;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 死信队列
     * 消费者 1
     * 大小写切换 ctrl+shirft+u
     */
    public class Consumer01 {
        //普通交换机的名称
        public static final String NORMAL_EXCHANGE = "normal_exchange";
        //死信交换机的名称
        public 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();
            //声明死信交换机 类型为direct  1.交换机的名字  2.交换机的类型
            channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
            //声明普通交换机 类型为direct
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            /**
             *          * 生成一个队列
             *          * 1.队列名称
             *          * 2.队列中的消息是否持久化  默认消息存储在内存中 否不持久化 (持久化放在磁盘中)
             *          *3.该队列是否只供一个消费者进行消费 是否共享  true 可以多个消费者消费  默认false,只能一个消费者消费
             *          4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true自动删除 false不删除
             *          5.其他参数
             *
             */
    
            //声明死信队列
            channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    
            //声明普通队列
            Map<String, Object> arguments = new HashMap<>();
            //过期时间   10s=100000ms
            // arguments.put("x-message-ttl",100000);
            //正常队列设置死信交换机        1.值  2.死信交换机名字
            arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
            //设置死信RoutingKey
            arguments.put("x-dead-letter-routing-key", "lisi");
            //声明普通队列
            channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
    
            //绑定普通的交换机与偶有普通的队列
            channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
            //绑定死信的交换机与偶有死信的队列
            channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
            System.out.println("等待接收消息.......");
    
            DeliverCallback deliverCallback = (var1, var2) -> {
                System.out.println("Consumer01" + new String(var2.getBody(), "utf-8"));
            };
            channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, var1 -> {
            });
    
        }
    
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72

    消费者 C2 代码(以上步骤完成后 启动 C2 消费者 它消费死信队列里面的消息)

    public class Consumer02 {
     
    private static final String DEAD_EXCHANGE = "dead_exchange"; 
    public static void main(String[] argv) throws Exception {
    
    
    
    Channel channel = RabbitUtils.getChannel(); 
    channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); 
    String deadQueue = "dead-queue";
     
    channel.queueDeclare(deadQueue, false, false, false, null); 
    channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
     
    System.out.println("等待接收死信队列消息 ........... "); 
    DeliverCallback deliverCallback = (consumerTag, delivery) ->
     
    { String message = new String(delivery.getBody(), "UTF-8"); 
    System.out.println("Consumer02 接收死信队列的消息" + message);
    };
     
    channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> { });
    }
    }
     
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
     package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DeliverCallback;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 死信队列
     * 消费者 1
     * 大小写切换 ctrl+shirft+u
     */
    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 = (var1, var2) -> {
                System.out.println("Consumer02" + new String(var2.getBody(), "utf-8"));
            };
            channel.basicConsume(DEAD_QUEUE, true, deliverCallback, var1 -> {
            });
    
        }
    
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    6.3.3. 队列达到最大长度
    1. 消息生产者代码去掉 TTL 属性

    在这里插入图片描述

    public class Producer {
     
    private static final String NORMAL_EXCHANGE = "normal_exchange"; 
    public static void main(String[] argv) throws Exception {
     
    try (Channel channel = RabbitMqUtils.getChannel())
     
    { 
    channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     
    //该信息是用作演示队列个数限制
     
    for (int i = 1; i <11 ; i++) {
     String message="info"+i;
     
    channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
     System.out.println("生产者发送消息:"+message);
     
    }
    }
    }
     
    }
     
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    2. C1 消费者修改以下代码(启动之后关闭该消费者 模拟其接收不到消息

    在这里插入图片描述

    注意此时需要把原先队列删除 因为参数改变了

    1. C2 消费者代码不变(启动 C2 消费者)

    在这里插入图片描述

    6.3.4. 消息被拒

    1. 消息生产者代码同上生产者一致

    2. C1 消费者代码(启动之后关闭该消费者 模拟其接收不到消息)

    public class Consumer01 {
    //普通交换机名称
     
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
     
    private static final String DEAD_EXCHANGE = "dead_exchange"; public static void main(String[] argv) throws Exception {
     
    Channel channel = RabbitUtils.getChannel();
     
    //声明死信和普通交换机 类型为 direct channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     
    //声明死信队列
     
    String deadQueue = "dead-queue"; channel.queueDeclare(deadQueue, false, false, false, null);
     
    //死信队列绑定死信交换机与 routingkey channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
    
    
    
    //正常队列绑定死信队列信息
     
    Map<String, Object> params = new HashMap<>();
     
    //正常队列设置死信交换机 参数 key 是固定值 params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
     
    //正常队列设置死信 routing-key 参数 key 是固定值 params.put("x-dead-letter-routing-key", "lisi"); String normalQueue = "normal-queue";
     
    channel.queueDeclare(normalQueue, false, false, false, params); channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
     
    System.out.println("等待接收消息 ........... ");
     
    DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); if(message.equals("info5")){
     
    System.out.println("Consumer01 接收到消息"   +                      message  +                                                                                       "并拒绝签收该消息");
     
    //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
     
    }else {
     
    System.out.println("Consumer01 接收到消息"+message); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
    };
     
    boolean autoAck = false;
     
    channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> { });
    }
     
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    1. C2 消费者代码不变

    启动消费者 1 然后再启动消费者 2

    在这里插入图片描述
    在这里插入图片描述

    在这里插入图片描述

    上面3个部分内容代码整理 如下 :

    C1

    package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DeliverCallback;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 死信队列
     * 消费者 1
     * 大小写切换 ctrl+shirft+u
     */
    public class Consumer01 {
        //普通交换机的名称
        public static final String NORMAL_EXCHANGE = "normal_exchange";
        //死信交换机的名称
        public 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();
            //声明死信交换机 类型为direct  1.交换机的名字  2.交换机的类型
            channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
            //声明普通交换机 类型为direct
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            /**
             *          * 生成一个队列
             *          * 1.队列名称
             *          * 2.队列中的消息是否持久化  默认消息存储在内存中 否不持久化 (持久化放在磁盘中)
             *          *3.该队列是否只供一个消费者进行消费 是否共享  true 可以多个消费者消费  默认false,只能一个消费者消费
             *          4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true自动删除 false不删除
             *          5.其他参数
             *
             */
    
    
            //声明普通队列
            Map<String, Object> arguments = new HashMap<>();
            //过期时间   10s=100000ms
           //  arguments.put("x-message-ttl",100000);
            //正常队列设置死信交换机        1.值  2.死信交换 机名字
            arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
            //设置死信RoutingKey
            arguments.put("x-dead-letter-routing-key", "lisi");
            //设置正常队列长度的限制
            //arguments.put("x-max-length",6);
    
            //声明普通队列
            channel.queueDeclare(NORMAL_QUEUE, false, false, false,arguments);
    
            //声明死信队列
            channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
    
            //绑定普通的交换机与偶有普通的队列
            channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
            //绑定死信的交换机与偶有死信的队列
            channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
            System.out.println("等待接收消息.......");
    
            DeliverCallback deliverCallback = (var1, var2) -> {
               String msg= new String(var2.getBody(), "utf-8");
               if (msg.equals("info5")){
                   System.out.println("Consumer01接受的消息是"+msg+"消息是被拒绝的");
                   // 1.获取到值 2.不放回普通队列
                   channel.basicReject(var2.getEnvelope().getDeliveryTag(),false );
               }else {
                   System.out.println("Consumer01接受的消息是" +msg);
                   channel.basicReject(var2.getEnvelope().getDeliveryTag(),false );
               }
    
    
            };
         //   channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, var1 -> {});
            channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, var1 -> {});
        }
    
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85

    C2

     package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.DeliverCallback;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 死信队列
     * 消费者 1
     * 大小写切换 ctrl+shirft+u
     */
    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 = (var1, var2) -> {
                System.out.println("Consumer02" + new String(var2.getBody(), "utf-8"));
            };
            channel.basicConsume(DEAD_QUEUE, true, deliverCallback, var1 -> {
            });
    
        }
    
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    生产者

     package com.atguigu.rabbitmq.eight;
    
    import com.atguigu.rabbitmq.utils.RabbitMqutils;
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 死信队列  生产者
     */
    public class Producer {
    
        //普通交换机 名称
        public static final String NORMAL_EXCHANGE="normal_exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Channel channel = RabbitMqutils.getChannel();
           //死信消息 设置TTl  单位是ms 100000ms=10s  过期时间
    //        AMQP.BasicProperties basicProperties =
    //                new AMQP.BasicProperties()
    //                        .builder().expiration("1000").build();
    
            for (int i = 1; i < 11; i++) {
                String message="info"+i;
                channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
              //  channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
                System.out.println("生产者发送消息"+message);
            }
    
    
    
    
    
        }
    
    }
     
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
  • 相关阅读:
    Go常见的语法题目
    CRMEB商城源码开源标准版v5.2.0+后端+前端uni-app开源包安装教程
    2024免费的硬盘数据恢复软件有哪些?
    编译原理:词法分析
    后端配置跨域怎么配置
    记将一个大型客户端应用项目迁移到 dotnet 6 的经验和决策
    RabbitMQ(四)交换机Exchanges
    如何使用chatGPT写好Prompt提示词: Few-shots
    行业追踪,2023-09-13
    Citrix XenDesktop云桌面单点登录XenApp虚拟应用小技巧
  • 原文地址:https://blog.csdn.net/qq_42055933/article/details/127399615