0320 rabbitmq 发布确认和死信队列

rabbitmq-demo/src/main/java/com/sf/deadletter/DeadLetter.java

@@ -0,0 +1,32 @@
+package com.sf.deadletter;
+
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Scanner;
+
+public class DeadLetter {
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare("exchange.dlx", "direct", true, false, null);
+        channel.queueDeclare("queue.dlx", true, false, false, null);
+        channel.queueBind("queue.dlx", "exchange.dlx", "dlx-routing-key");
+
+        Map<String, Object> param = new HashMap<String, Object>();
+        param.put("x-dead-letter-exchange", "exchange.dlx");
+        param.put("x-dead-letter-routing-key", "dlx-routing-key");
+        channel.exchangeDeclare("exchange.normal2", "direct");
+        channel.queueDeclare("queue.normal2", false, false, false, param);
+        channel.queueBind("queue.normal2", "exchange.normal2", "zhangsan");
+        Scanner scanner = new Scanner(System.in);
+        System.out.println("请输入消息：");
+        while (scanner.hasNext()) {
+            String message = scanner.next();
+            System.out.println(message);
+            channel.basicPublish("exchange.normal2", "dlx-routing-key", null, message.getBytes());
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/DeadLetter1.java

@@ -0,0 +1,36 @@
+package com.sf.deadletter;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.MessageProperties;
+import com.sf.util.MqUtils;
+
+
+import java.util.HashMap;
+
+public class DeadLetter1 {
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        // 定义 dlx
+        channel.exchangeDeclare("exchange.dlx", "direct", true, false, null);
+        // 定义 dlx queue
+        channel.queueDeclare("queue.dlx", true, false, false, null);
+        channel.queueBind("queue.dlx", "exchange.dlx", "dlx-routing-key");
+
+        // 定义正常的交换器
+        channel.exchangeDeclare("exchange.normal", "fanout", true, false, null);
+        HashMap<String, Object> arguments = new HashMap<>();
+        // 定义队列时，通过该属性给该队列设置 DLX
+        arguments.put("x-dead-letter-exchange", "exchange.dlx");
+        // 还可以通过该属性重新设置消息的路由键，否则使用原消息的路由键
+        arguments.put("x-dead-letter-routing-key", "dlx-routing-key");
+        // 设置该队列的 ttl
+        arguments.put("x-message-ttl", 10000);
+        channel.queueDeclare("queue.normal", true, false, false, arguments);
+        channel.queueBind("queue.normal", "exchange.normal", "routing");
+
+        String message = "hello world111";
+        channel.basicPublish("exchange.normal", "key", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
+
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/length/Consumer1.java

@@ -0,0 +1,43 @@
+package com.sf.deadletter.length;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.Delivery;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+
+public class Consumer1 {
+
+    private final static String NORMAL_EXCHANGE = "normal_exchange1";
+    private final static String NORMAL_QUEUE = "normal_queue1";
+    private final static String DEAD_EXCHANGE = "dead_exchange1";
+    private final static String DEAD_QUEUE = "dead_queue1";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
+        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
+
+        HashMap<String, Object> arguments = new HashMap<>();
+        // 定义队列时，通过该属性给该队列设置 DLX
+        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
+        // 还可以通过该属性重新设置消息的路由键，否则使用原消息的路由键
+        arguments.put("x-dead-letter-routing-key", "lisi");
+        arguments.put("x-max-length", 6);   // 设置正常队列长度为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("消费者1等待消息");
+        DeliverCallback deliverCallback = (String consumerTag, Delivery message) -> {
+            String body = new String(message.getBody());
+            System.out.println(body);
+        };
+        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> {
+        });
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/length/Producer.java

@@ -0,0 +1,24 @@
+package com.sf.deadletter.length;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+public class Producer {
+
+    private final static String EXCHANGE_NAME = "normal_exchange1";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
+
+        int cnt = 10;
+        while (cnt > 0) {
+            String message = cnt + "";
+            System.out.println(message);
+            channel.basicPublish(EXCHANGE_NAME, "zhangsan", null, message.getBytes());
+            cnt--;
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/reject/Consumer01.java

@@ -0,0 +1,54 @@
+package com.sf.deadletter.reject;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.CancelCallback;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.DeliverCallback;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class Consumer01 {
+    private static final String NORMAL_EXCHANGE = "normal_exchange2";    // 普通交换机名称
+    private static final String DEAD_EXCHANGE = "dead_exchange2";    // 死信交换机名称
+    private static final String NORMAL_QUEUE = "normal_queue2";  // 普通队列名称
+    private static final String DEAD_QUEUE = "dead_queue2";  // 死信队列名称
+
+    public static void main(String[] args) throws Exception {
+        // 创建channel
+        Channel channel = MqUtils.getChannel();
+        // 声明普通和死信交换机(类型都为DIRECT)
+        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
+        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
+        // 声明普通和死信队列(普通队列需要传递参数设置死信交换机及其对应的路由key)
+        Map<String, Object> arguments = new HashMap<>();
+        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE); // 设置死信交互机
+        arguments.put("x-dead-letter-routing-key", "lisi"); // 设置与死信交换机间的routing-key
+        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
+        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
+        // 绑定队列与交换机,设置其间的路由key
+        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
+        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
+
+        // 消费消息
+        System.out.println("Consumer01等待接收普通队列到消息......");
+        DeliverCallback deliverCallback = (consumerTag, message) -> {
+            String mes = new String(message.getBody(), "UTF-8");
+            if (mes.equals("5")) {
+                System.out.println("Consumer01接收到普通队列的消息" + mes + "并拒收该消息");
+                // 第二个参数requeue设置为false,代表拒绝重新入队,也就是该队列如果配置了死信交换机将发送到死信队列中
+                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
+            } else {
+                System.out.println("Consumer01接收到普通队列的消息:" + new String(message.getBody()));
+                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
+            }
+        };
+        CancelCallback cancelCallback = (consumerTag) -> {
+            System.out.println("消息消费被中断");
+        };
+        // 开启手动应答
+        boolean autoAck = false;
+        channel.basicConsume(NORMAL_QUEUE, autoAck, deliverCallback, cancelCallback);
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/reject/Producer.java

@@ -0,0 +1,23 @@
+package com.sf.deadletter.reject;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+public class Producer {
+
+    private final static String EXCHANGE_NAME = "normal_exchange2";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
+
+        int cnt = 10;
+        while (cnt > 0) {
+            String message = cnt + "";
+            System.out.println(message);
+            channel.basicPublish(EXCHANGE_NAME, "zhangsan", null, message.getBytes());
+            cnt--;
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/ttl/Consumer1.java

@@ -0,0 +1,42 @@
+package com.sf.deadletter.ttl;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.Delivery;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+
+public class Consumer1 {
+
+    private final static String NORMAL_EXCHANGE = "normal_exchange";
+    private final static String NORMAL_QUEUE = "normal_queue";
+    private final static String DEAD_EXCHANGE = "dead_exchange";
+    private final static String DEAD_QUEUE = "dead_queue";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
+        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
+
+        HashMap<String, Object> arguments = new HashMap<>();
+        // 定义队列时，通过该属性给该队列设置 DLX
+        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
+        // 还可以通过该属性重新设置消息的路由键，否则使用原消息的路由键
+        arguments.put("x-dead-letter-routing-key", "lisi");
+        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("消费者1等待消息");
+        DeliverCallback deliverCallback = (String consumerTag, Delivery message) -> {
+            String body = new String(message.getBody());
+            System.out.println(body);
+        };
+        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> {
+        });
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/ttl/Consumer2.java

@@ -0,0 +1,30 @@
+package com.sf.deadletter.ttl;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.Delivery;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+
+public class Consumer2 {
+
+    private final static String DEAD_EXCHANGE = "dead_exchange";
+    private final static String DEAD_QUEUE = "dead_queue";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
+        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
+        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
+
+        System.out.println("消费者2等待消息");
+        DeliverCallback deliverCallback = (String consumerTag, Delivery message) -> {
+            String body = new String(message.getBody());
+            System.out.println(body);
+        };
+        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> {
+        });
+    }
+}

rabbitmq-demo/src/main/java/com/sf/deadletter/ttl/Producer.java

@@ -0,0 +1,27 @@
+package com.sf.deadletter.ttl;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+import java.util.Scanner;
+
+public class Producer {
+
+    private final static String EXCHANGE_NAME = "normal_exchange";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
+
+        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
+        int cnt = 10;
+        while (cnt > 0) {
+            String message = cnt + "";
+            System.out.println(message);
+            channel.basicPublish(EXCHANGE_NAME, "zhangsan", properties, message.getBytes());
+            cnt--;
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/ProducerTTL.java

@@ -0,0 +1,31 @@
+package com.sf.handle;
+
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Scanner;
+
+public class ProducerTTL {
+
+    private final static String QUEUE_NAME = "hello1";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        // 队列的持久化 是在声明队列时
+        Map<String, Object> param = new HashMap<String, Object>();
+        param.put("x-message-ttl", 60000);
+        boolean durable = true;
+        channel.queueDeclare(QUEUE_NAME, durable, false, false, param);
+
+        Scanner scanner = new Scanner(System.in);
+        System.out.println("请输入消息：");
+        while (scanner.hasNext()) {
+            String message = scanner.next();
+            System.out.println(message);
+            // 如果只有队列是持久化 而消息不是持久化的  那消息也会丢失
+            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/ProducerTx.java

@@ -0,0 +1,35 @@
+package com.sf.handle;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.MessageProperties;
+import com.sf.util.MqUtils;
+
+public class ProducerTx {
+    private static final String EXCHANGE_NAME = "exchange_demo";
+    private static final String QUEUE_NAME = "queue_demo2";
+    private static final String ROUTING_KEY = "routing_key_demo";
+
+    public static void main(String[] args) throws Exception {
+        // 得到连接通道
+        Channel channel = MqUtils.getChannel();
+        // 创建一个type=direct 持久化 非自动删除的交换器
+        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
+        // 创建一个持久化 非排他 非自动删除的队列
+        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
+        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
+        String message = "hello world";
+        try {
+            channel.txSelect();
+            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
+                    MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
+            // 故意制造异常
+            int result = 1 / 0;
+            channel.txCommit();
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            // 回滚事务
+            channel.txRollback();
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/confirm/Producer.java

@@ -0,0 +1,38 @@
+package com.sf.handle.confirm;
+
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+import java.util.Scanner;
+
+/**
+ * 单个确认
+ */
+public class Producer {
+
+    private final static String QUEUE_NAME = "hello2";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        // 开启发布确认
+        channel.confirmSelect();
+        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+
+        Scanner scanner = new Scanner(System.in);
+        System.out.println("请输入消息：");
+        while (scanner.hasNext()) {
+            long start = System.currentTimeMillis();
+            String message = scanner.next();
+            System.out.println(message);
+            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
+            // 等待队列的接收结果
+            boolean flag = channel.waitForConfirms();
+            // 如果为true 代表成功
+            if(flag){
+                System.out.println("已经收到消息");
+            }
+            long end = System.currentTimeMillis();
+            System.out.println("发布消息耗时：" + (end - start) + "毫秒");
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/confirm/ProducerASync.java

@@ -0,0 +1,82 @@
+package com.sf.handle.confirm;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConfirmCallback;
+import com.rabbitmq.client.MessageProperties;
+import com.sf.util.MqUtils;
+
+
+import java.io.IOException;
+import java.util.Scanner;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+/**
+ * 异步发布
+ */
+public class ProducerASync {
+
+    private final static String QUEUE_NAME = "hello2";
+
+    public static void main(String[] args) throws Exception {
+        // 得到连接通道
+        Channel channel = MqUtils.getChannel();
+        // 声明队列
+        //   名称 是否持久化 exclusive排他锁 自动删除 其他参数(死信队列)
+        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+        // 开启发布确认 为了发送消息更安全
+        channel.confirmSelect();
+
+        // 通过跳表实现的线程安全map
+        // 将序号与消息进行关联  方便将小于等于某个消息序号的消息作为map取出来
+        ConcurrentSkipListMap<Long, String> outstandingConfirms =
+                new ConcurrentSkipListMap<>();
+        // 应答回调函数
+        // 消息序号 处理一个还是多个消息
+        ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {
+            System.out.println("已经收到消息");
+            if (multiple) {
+                // NavigableMap是SortedMap的子接口  navigable 可驾驶的 可航行的
+                // headMap() 返回小于第一个参数的键值对的视图
+                ConcurrentNavigableMap<Long, String> confirmed =
+                        outstandingConfirms.headMap(sequenceNumber, true);
+                confirmed.clear();
+            } else {
+                outstandingConfirms.remove(sequenceNumber);
+            }
+        };
+        // 未收到消息回调
+        ConfirmCallback confirmCallback = new ConfirmCallback() {
+            @Override
+            public void handle(long sequenceNumber, boolean multiple)
+                    throws IOException {
+                String message = outstandingConfirms.get(sequenceNumber);
+                System.out.println("序号为" + sequenceNumber + "的消息需要重新发送");
+            }
+        };
+        // 添加一个监听器 收到消息回调 Basic.Ack 或者未收到消息回调 Basic.Nack
+        channel.addConfirmListener(cleanOutstandingConfirms, confirmCallback);
+
+        System.out.println("输入消息：");
+        Scanner scanner = new Scanner(System.in);
+        while (scanner.hasNext()) {
+            long start = System.currentTimeMillis();
+            String message = scanner.next();
+            // 获取消息的下一个标号 从1开始
+
+            // 通过channel提供的方法getNextPublishSeqNo来获取 消息发布成功后的id
+            // 使用一个map来存储 待确认消息的id
+            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
+            channel.basicPublish("", "hello", null, message.getBytes());
+            System.out.println(message);
+            // 等待broker
+            // 阻塞等待
+//            boolean flag = channel.waitForConfirms();
+//            if (flag) {
+//                System.out.println("broker已经收到消息");
+//            }
+            long end = System.currentTimeMillis();
+            System.out.println("发布异步确认消息耗时：" + (end - start) + "毫秒");
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/confirm/ProducerBatch.java

@@ -0,0 +1,47 @@
+package com.sf.handle.confirm;
+
+import com.rabbitmq.client.Channel;
+import com.sf.util.MqUtils;
+
+import java.util.Scanner;
+
+/**
+ * 批量确认
+ */
+public class ProducerBatch {
+
+    private final static String QUEUE_NAME = "hello2";
+
+    public static void main(String[] args) throws Exception {
+        Channel channel = MqUtils.getChannel();
+        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+        // 开启发布确认
+        channel.confirmSelect();
+        // 批量的大小
+        int batchSize = 5;
+        // 当前发送的消息个数
+        int messageCnt = 0;
+
+        Scanner scanner = new Scanner(System.in);
+        long start = System.currentTimeMillis();
+        System.out.println("请输入消息：");
+        while (scanner.hasNext()) {
+            String message = scanner.next();
+            System.out.println(message);
+            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
+            messageCnt++;
+            if (messageCnt == batchSize) {
+                // 等待队列的接收结果
+                boolean flag = channel.waitForConfirms();
+                // 如果为true 代表成功
+                if (flag) {
+                    System.out.println("已经收到消息");
+                    long end = System.currentTimeMillis();
+                    System.out.println("发布消息耗时：" + (end - start) + "毫秒");
+                    messageCnt = 0;
+                    start = System.currentTimeMillis();
+                }
+            }
+        }
+    }
+}

rabbitmq-demo/src/main/java/com/sf/handle/confirm/TestMap.java

@@ -0,0 +1,25 @@
+package com.sf.handle.confirm;
+
+import java.util.NavigableMap;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+public class TestMap {
+
+    public static void main(String[] args) {
+        NavigableMap original = new TreeMap();
+        original.put("1", "1");
+        original.put("2", "2");
+        original.put("3", "3");
+        original.put("0", "0");
+        original.put("4", "4");
+
+        //this headmap1 will contain "0" "1" and "2"
+        SortedMap headmap1 = original.headMap("3");
+        System.out.println(headmap1);
+
+        //this headmap2 will contain "0" "1", "2", and "3" because "inclusive"=true
+        NavigableMap headmap2 = original.headMap("3", true);
+        System.out.println(headmap2);
+    }
+}