RabbitMQ——交换机、死信队列_rabbitmq 交换机,队列

交换机

概念

RabbitMQ 消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。实际上，通常生产 者甚至都不知道这些消息传递传递到了哪些队列中。

相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这些由交换机的类型来决定。

交换机的类型

• 直接（direct）
• 主题（topic）
• 标题（headers） 几乎不用了
• 扇出（fanout）
• 无名交换机：我们在之前的学习过程中通过空字符串表示，就是使用的无名交换机

临时队列

之前的学习过程中，我们在创建队列的时候，都会给队列起一个别名。临时队列的作用就在于它的名称是随机的，而且一旦Rabbit断开连接，队列将会被自动删除。创建临时队列的方式如下

String queueName = channel.queueDeclare().getQueue();
1

fanout（扇出）

fanout交换机有点类似于广播的形式，之前我们发送的消息只会被一个消费者消费，而通过fanout交换机，可以做到所有消费者都获得这个消息。如下图所示

接下来我们通过实际案例来使用代码演示一下

编写两个工作线程

public class ReceiveLogs01 {
    public static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception{
        System.out.println("工作线程A正在等待消息......");
        //创建信道
        Channel channel = RabbitMqUtils.getChannel();
        //设置交换机类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //生成一个临时队列
        String queueName = channel.queueDeclare().getQueue();
        /**
         * 将交换机与队列绑定
         * 1.队列名称
         * 2.交换机名称
         * 3.routingkey设置为空字符串
         **/
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(),"UTF-8");
            System.out.println("接收到的消息是：" + message);
        };
        //接受消息
        channel.basicConsume(queueName,true,deliverCallback, consumerTag -> {});
    }
}

public class ReceiveLogs02 {
    public static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception{
        System.out.println("工作线程B正在等待消息......");
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(),"UTF-8");
            System.out.println("接收到的消息是：" + message);
        };
        channel.basicConsume(queueName,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

编写生产者

public class EmitLog {
    public static String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8"));
            System.out.println("已发送消息：" + message);
        }

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

测试

通过测试我们可以发现两个工作线程可以接收到同一个消息

direct（直接）

direct类型的交换机，可以根据routingkey来指定哪一个工作线程可以获取该消息，也就是说消息不再是一个一个分配或者不公平模式，而是可以进行指定。如下图所示

通过代码来演示一下

编写两个工作线程

public class ReceiveLogsDirect01 {
    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //绑定交换机，并设置类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //生成一个临时队列
        String queueName = channel.queueDeclare().getQueue();
        //多重绑定
        channel.queueBind(queueName,EXCHANGE_NAME,"black");
        channel.queueBind(queueName,EXCHANGE_NAME,"green");
        System.out.println("工作线程A正在等待消息......");
        //获取消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(),"UTF-8");
            System.out.println("接收到的消息是：" + message);
        };
        //接受消息
        channel.basicConsume(queueName,true,deliverCallback, consumerTag -> {});
    }
}

public class ReceiveLogsDirect02 {
    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //绑定交换机，并设置类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //生成一个临时队列
        String queueName = channel.queueDeclare().getQueue();
        //单个绑定
        channel.queueBind(queueName,EXCHANGE_NAME,"orange");
        System.out.println("工作线程B正在等待消息......");
        //获取消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(),"UTF-8");
            System.out.println("接收到的消息是：" + message);
        };
        //接受消息
        channel.basicConsume(queueName,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

编写生产者代码

public class EmitLogDirect {
    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        Scanner scanner = new Scanner(System.in);

        while (scanner.hasNext()){
            String message = scanner.next();
            //第二个参数指定哪个工作线程获取消息
            channel.basicPublish(EXCHANGE_NAME,"orange",null,message.getBytes("UTF-8"));
            System.out.println("已发送消息：" + message);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

测试

我们设置生成者发送消息给绑定为orange的队列，测试结果如下图，工作线程B会获得消息，而工作线程A没有获得消息

Topic（主题）

之前两种类型的交换机或多或少都有那么一点问题，比如说我现在需要在上述例子下，让工作线程A和工作线程B接收到同一个消息，这个没有办法实现，而Topic交换机功能最为强大。

发送到类型是 topic 交换机的消息的 routing_key 不能随意写，必须满足一定的要求，它必须是一个单 词列表，以点号分隔开。这些单词可以是任意单词，比如说：“stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”.这种类型的。当然这个单词列表最多不能超过 255 个字节。

在这个规则列表中，其中有两个替换符是大家需要注意的

*(星号)可以代替一个单词

#(井号)可以替代零个或多个单词

这个咱们就不进行代码演示了，简单来说一下，比如一个队列与交换机的绑定键为*.rabbit.*，另外一个为lazy.*，这时候生产者的key设置为lazy.rabbit那么两个线程都会获得这个消息

在实际开发过程中，建议使用此类型的交换机

死信队列

概念

死信，顾名思义就是无法被成功消费的消息，producer将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

成为死信队列的原因

• 消息被拒收
• 消息达到过期时间
• 队列已满

死信实战

先通过一张图来理解一下我们的案例，生产者发送一条消息，通过zhangsan这个绑定键来绑定普通队列，成功接受的消息被消费者C1消费，如果满足成为死信的情况，交给死信交换机，通过lisi来绑定死信队列，交给消费者2来处理。

消息过期情况

编写消费者C1

public class Consumer01 {
    //普通交换机名称
    private static String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static String DEAD_EXCHANGE = "dead_exchange";
    //普通队列名称
    private static String NORMAL_QUEUE = "normal_queue";
    //死信队列名称
    private static 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);
        //设置普通队列参数
        HashMap<String, Object> map = new HashMap<>();
        //key为固定值，设置转发到死信交换机的名称
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //key为固定值，设置死信交换机绑定死信队列的绑定键
        map.put("x-dead-letter-routing-key","lisi");
        //创建普通队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        //普通队列与普通交换机的绑定
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        //创建死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //死信队列绑定死信交换机
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        //获取消息
        System.out.println("消费者C1正在等待消息......");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消费者C1接收到消息：" + new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE,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

编写生产者

在生产者发送消息时，设置10秒后过期

public class Producer01 {
    private static String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //设置消息过期时间，单位为ms，设置10秒过期
        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().expiration("10000").build();
        //发送消息
        for (int i = 0; i < 10; i++) {
            String message = "info" + i;
            //第三个参数设置过期时间
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes("UTF-8"));
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

编写消费者C2

消费者C2较为简单，只需要接收死信队列的信息即可

public class Consumer02 {
    //死信队列名称
    private static String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception{
        //创建信道
        Channel channel = RabbitMqUtils.getChannel();
        //获取消息
        System.out.println("消费者C2正在等待消息......");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消费者C2接收到消息：" + new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(DEAD_QUEUE,true,deliverCallback , consumerTag -> {});
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

测试

我们先启动消费者C1，在关闭消费者C1不让他成功接收消息，再启动生产者。刚启动时，10条数据在normal_queue队列中，过了10秒后，转发到了dead_queue队列，启动消费者C2接收死信队列的消息

队列已满情况

我们只需要修改上面的一些条件即可，C2不用修改

修改生产者代码

生产者代码将设置的过期时间去掉即可

public class Producer01 {
    private static String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //设置消息过期时间，单位为ms，设置10秒过期
//        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().expiration("10000").build();
        //发送消息
        for (int i = 0; i < 10; i++) {
            String message = "info" + i;
            //第三个参数设置过期时间
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes("UTF-8"));
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

修改消费者C1代码

在原有基础上不变，在map中添加以下设置，作用是将此队列的长度设置为6

//key为固定值，设置普通队列的长度
map.put("x-max-length",6);
1
2

测试

还是先启动消费者C1，注意在此之前需要把原先的队列删除，否则会报错，启动完之后我们可以发现normal_queue队列中多了一个Lim的属性，这是说该队列的最大容积，也就是我们设置的6

这时候我们关闭消费者C1，再启动生产者代码，生产者一共发送十条信息，而普通队列最多容纳6条，4条消息会被转发到死信队列中

这时候我们启动消费者C1和C2，C1会消费普通队列的消息，C2会消费死信队列的消息

消息被拒

消息被拒，我们需要将之前设置的最大长度取消掉，再来修改消费者C1的代码

修改消费者C1

public class Consumer01 {
    //普通交换机名称
    private static String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static String DEAD_EXCHANGE = "dead_exchange";
    //普通队列名称
    private static String NORMAL_QUEUE = "normal_queue";
    //死信队列名称
    private static 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);
        //设置普通队列参数
        HashMap<String, Object> map = new HashMap<>();
        //key为固定值，设置转发到死信交换机的名称
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //key为固定值，设置死信交换机绑定死信队列的绑定键
        map.put("x-dead-letter-routing-key","lisi");
        //key为固定值，设置普通队列的长度
//        map.put("x-max-length",6);
        //创建普通队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        //普通队列与普通交换机的绑定
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        //创建死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        //死信队列绑定死信交换机
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        //获取消息
        System.out.println("消费者C1正在等待消息......");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            //假设我们拒收info5的消息
            String msg = new String(message.getBody(),"UTF-8");
            if ("info5".equals(msg)){
                System.out.println("消费者C1接收到消息：" + msg + ",但拒收了");
                //第二个参数requeue设置是否重新入队，选择false代表不重新入队，如果存在死信队列会转发到死信队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else {
                System.out.println("消费者C1接收到消息：" + msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }
        };
        //要想能实现拒收消息，需要修改为手动应答
        channel.basicConsume(NORMAL_QUEUE,false,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
51

测试

分别启动消费者C1、生产者、消费者C2，C1会拒收info5这条消息，C2可以接收到info5