RabbitMQ从入门到实战(4)- 消息的交换（中）

「爱情、让人受尽委屈。」 2022-05-27 03:17 276阅读 0赞

转载于：http://www.cnblogs.com/4----/p/6590459.html

# 1.简介 #

本章节和官方教程相似度较高，英文好的可以移步[官方教程][Link 1]

在上一章的例子中，我们创建了一个消费者，生产日志消息，广播给两个消费者，对消息进行不同的处理。这一节，我们将对它进行扩展，实现一些更加高级的功能，例如：使消费者A只接受error级别的日志保存到硬盘，消费者B接收所有级别的消息进行打印。

本文中涉及到的所有概念（包括前面几章），都将摒弃个人经验，以官方文档为基础进行讲解，在书写本文的同时，也是我对RabbitMQ的重新学习。

# 2.绑定 #

回顾一下上一章的队列绑定代码

> // 把刚刚获取的队列绑定到logs这个交换中心上，  
> channel.queueBind(queueName, "logs", "");

这段代码在消费者中，为什么生产者没有？因为在RabbitMQ中消息是发送到交换中心（exchange）的，这在上一张已经重点强调过。

上述代码可以理解成，queueName这个队列对logs这个exchange中的消息感兴趣，routingKey是""

在发送消息的basicPublish方法中，也有一个参数叫做routingKey，没错，他们是有关联的，下面会介绍

在不同的exchange类型中，routingKey扮演的角色也相应的不同，比如上一章我们使用的fanout（扇出，多贴切的名字，想象一下WOW中盗贼的刀扇）将忽略routingKey，所有绑定在fanout类型的exchange上的队列，都将接收到该exchange上的所有消息。

# 3.Direct Exchange #

fanout类型的exchange没有给我们太多的灵活性，direct类型的echange非常简单，会匹配消息发布时的routingKey和queue的routingKey，完全相等则把消息放入该队列。

![image][]

如上图，Q1绑定了orange，Q2绑定了black和green，就可以实现不同级别的日志用不同的消费者进行处理

我们看到Q2绑定了两个routingKey，难道第二次绑定不会把第一次绑定覆盖掉吗？

实践出真正，我们来试一下

![image][image 1]

1.  声明一个名为logs的exchange，类型换为direct，让它通过routingKey的完全匹配去分发消息
2.  然后把消息发送到名为logs的exchange上，routingKey是外面传进来的

改造一下发送方法，轮流发送info和error信息

![image][image 2]

给Consummer队列绑定两个routingKey

![image][image 3]

激动人心的时刻到来了，跑一把

Duang，报错了

![image][image 4]

报错信息：

> inequivalent arg 'type' for exchange 'logs' in vhost '/': received 'direct' but current is 'fanout', class-id=40, method-id=10)

大意就是logs exchange已经被声明称fanout了，不能再声明成direct类型，RabbitMQ的队列声明方法和exchange声明方法都是幂等的，如果没有，就创建，如果有，参数相同，就不管，如果有了还用不同的参数重新声明，就报错

进入RabbitMQ控制台把logs删除，重新执行

![image][image 5]

逆袭成功，消费一下看看

![image][image 6]

成功了，一个队列可以绑定多个routingKey，这里注意先启动消费者，因为前面的代码里我们用的是临时队列，断开连接后，队列就删除了，如果先启动生产者，exchange接到消息后发现没有队列对它感兴趣，就任性的把消息给丢掉了。

一个队列可以绑定多个routingKey，反之，一个routingKey也可以绑定多个队列，如下图，感兴趣的朋友可以自己试一下

![image][image 7]

如果绑定在一个direct类型的exchange上的队列都使用同一个routingKey，那它就是一个fanout

# 4.实战 #

要实现本章的需求，即Q1只接收error级别的日志写到硬盘上，Q2接收error和info级别的日志打印出来

用direct类型的exchange来实现这个需求非常简单，Q1绑定error，Q2绑定error和info即可，缺点是Q2需要绑定N个routingKey，N=日志级别数量，我们可以用一些编程的技巧来规避它

Sender的代码上面已经改好了，把exchange换为direct，注意删除原exchange，不再赘述

Q2绑定所有日志级别，我们用一个Enum来规避手动绑定

定义一个Enum

1 public enum LogType{
    2     error,info;
    3 }

用foreach语法糖进行循环绑定

1 //绑定所有类型
    2 for(LogType logType: LogType.values()){
    3     channel.queueBind(queueName, "logs", logType.name());
    4 }

foreach是单线程的，这里也可以装个逼用一下JAVA8的lambda，由于lambda是并行处理，所以外围的try catch无效，需要在内层重新抓取异常，而且不能抛出，反而显得代码很不美好，装逼失败

![复制代码][copycode.gif]

1 IntStream.range(0, LogType.values().length).forEach(n->{
    2     try {
    3         channel.queueBind(queueName, "logs", LogType.values()[n].name());
    4     } catch (IOException e) {
    5         e.printStackTrace();
    6     }
    7 });

![复制代码][copycode.gif]

把另外一个Consumer改成只绑定error队列

1 channel.queueBind(queueName, "logs", LogType.error.name());

然后，改造一下发送消息的地方，一开始我们用了一个while还有一组if else，看起来比较挫，别人看你代码的时候，就不会觉得你很厉害，这样和你不写博客一样，对你的工作是没有好处的，我们把它改的高端一点

1 while(true){
    2     boolean info = ++i%2==0;
    3     String type = info?LogType.info.name():LogType.error.name();
    4     sender.sendMessage(type +" message: "+i, type);
    5     Thread.sleep(1000);
    6 }

对比一下

![image][image 2]

是不是觉得自己厉害了很多？这就是编码的艺术（得意脸）

好了，这一章没有太多内容，跑一下看看结果

![1120662-20170320185108736-1891329605.png][]

左边的Consumer1，消费了info和error级别的日志，右边的Consumer2，只消费了error级别的日志

# 5.结束语 #

这一章主要是介绍了RabbitMQ中direct类型的exchange，下一章将跟着官方教程的进度继续介绍topic类型的exchange，以及下下章介绍用RabbitMQ实现RPC调用。之后则会介绍RabbitMQ与Spring的集成等与真实开发环境更相关的技术。

[Link 1]: http://www.rabbitmq.com/tutorials/tutorial-four-java.html
[image]: /images/20220527/517db27c3cbc4ebb9f356d4ec5c42d75.png
[image 1]: /images/20220527/c40091c7b8a0490cba64950adc22693d.png
[image 2]: /images/20220527/29e8bea1498f4a62a037bbbc34c8fd9c.png
[image 3]: /images/20220527/2e5ecb39bd4e4c8abb2e790ac28e4ec4.png
[image 4]: /images/20220527/b7ff416fa5a6489f94e32b63b2e0c113.png
[image 5]: /images/20220527/13e29342396541d89afff4de3c7cdb13.png
[image 6]: /images/20220527/72dc92473f6f47128ad3610828a83c0b.png
[image 7]: /images/20220527/bcfe16870cb34e57a621e4e120b7b9c1.png
[copycode.gif]: /images/20220527/cad2416dcdae49e297fc6d300a552fa6.png
[1120662-20170320185108736-1891329605.png]: /images/20220527/2b9d721929134ea39d64456158cf64c6.png