转载请注明出处

0.目录

1.简介

上一章介绍了direct类型的exchange，并用它实现了一个伪广播（Queue绑定多个routingKey）日志系统，现在，考虑另一个问题，我们的日志系统不仅要分级别级别（error，info）记录日志，还需要通过发送日志的系统来匹配，比如说有一个“核心”系统，它发出的所有级别日志，都需要记录到硬盘，其他系统只需要把error级别的日志记录到硬盘。

如果用上一章的direct该怎么做呢？

给routingKey分层，变成类似这样的字符串：核心.info，核心.error，其他.info，其他.error
Q1绑定routingKey：核心.info，核心.error，其他.error，记录所有核心日志，记录其他error日志
Q2绑定routingKey：核心.info，其他.info，打印所有info日志

需求实现了，这时，项目经理说，两个日志级别太不好管理了，我们加个debug级别吧！

你的内心这样的

是时候学习一下Topic Exchange了

2.Topic Exchange

topic exchange对routingKey是有要求的，必须是一个关键字的列表才能发挥正常作用，用“.”分割每个关键字，你可以定义任意的层级，唯一的限制是最大长度为255字节。

上述需求，我们可以把routingKey的规则定义为 “<系统>.<日志级别>”，这个规则是抽象的，也就是说，是在你脑子里的，并没有地方去具体的把它绑定到exchange上，发送消息和绑定队列完全可以不按这个规则来，只会影响消息是否能分发到对应的队列上。

用“.”分割同样不是强制要求，routingKey里不包含这个字符也不会报错，“.”只会影响topic中对routingKey的分层解析，果不用它，那么topic的表现和direct一致

topic与direct的重要区别就是，它有两个关键字

“*”星号，代表一个词，比如上述规则：*.error 表示所有系统的error级别的日志
“#”井号，代表零个或多个词，比如上述规则： *.# 表示所有系统的所有消息，与单独一个#是等效的，core.# 表示核心系统的所有日志，它和 core.* 的区别是，即使以后规则改为 <系统>.<日志级别>.<其他条件>.<其他条件>.……，core.# 依然可以完成匹配，而 core.* 则无法匹配 core.info.xxx.xxx

第一条很好理解，第二条有点长，不会是骗人的吧？我们来实验一下

首先把把logs的type声明成topic，注意在控制台把上一章的direct类型的logs删除掉

channel.exchangeDeclare("logs","topic");

把Consumer的routingKey提取出来，方便后面测试

 /**

  * 获取一个临时队列，并绑定到相应的routingKey上，消费消息

  */

 public void consume(String routingKey) {

     try {

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

         //临时队列绑定的routingKey是外部传过来的

         channel.queueBind(queueName, "logs", routingKey);

         Consumer consumer = new DefaultConsumer(channel) {

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

                     byte[] body) throws IOException {

                 String message = new String(body, "UTF-8");

                 logger.debug(" [D] 打印日志："+message);

             }

         };

         //这里自动确认为true，接收到消息后该消息就销毁了

         channel.basicConsume(queueName, true, consumer);

     } catch (IOException e) {

         e.printStackTrace();

     }

 }

测试1

Consumer的临时队列绑定到logs上，routingKey设置为 a.#

 public static void main( String[] args )

 {

     LogConsumer consumer = new LogConsumer();

     consumer.consume("a.#");

 }

Sender的发送到logs上，routingKey设置为 a.b

 public static void main( String[] args ) throws InterruptedException{

     LogSender sender = new LogSender();

     while(true){

         String routingKey = "a.b";

         sender.sendMessage(" message: "+System.currentTimeMillis(), routingKey);

         Thread.sleep(1000);

     }

 }

运行，开始发送消息

成功消费

测试2

Consumer的临时队列绑定到logs上，routingKey设置为 a.*

1 public static void main( String[] args )

2 {

3     LogConsumer consumer = new LogConsumer();

4     consumer.consume("a.*");

5 }

Sender不变

运行，开始发送消息

消费成功

测试3

Consumer继续绑定到a.*

Sender的发送到logs上，routingKey设置为 a.b.c

1 public static void main( String[] args ) throws InterruptedException{

2     LogSender sender = new LogSender();

3     while(true){

4         String routingKey = "a.b.c";

5         sender.sendMessage(" message: "+System.currentTimeMillis(), routingKey);

6         Thread.sleep(1000);

7     }

8 }

运行，开始发送消息

a.*监听不到消息

测试4

Consumer改为监听 a.#

 public static void main( String[] args )

 {

     LogConsumer consumer = new LogConsumer();

     consumer.consume("a.#");

 }

继续往a.b.c发消息

a.# 消费成功

测试5

下面测点特殊的

Consumer绑定到 a.b

public static void main( String[] args )

{

    LogConsumer consumer = new LogConsumer();

    consumer.consume("a.b");

}

Sender发送到a.*

 public static void main( String[] args ) throws InterruptedException{

     LogSender sender = new LogSender();

     while(true){

         String routingKey = "a.*";

         sender.sendMessage(" message: "+System.currentTimeMillis(), routingKey);

         Thread.sleep(1000);

     }

 }

发送成功

消费不到

测试6

Sender发送到a.#，Consumer还是监听 a.b

1 public static void main( String[] args ) throws InterruptedException{

2     LogSender sender = new LogSender();

3     while(true){

4         String routingKey = "a.#";

5         sender.sendMessage(" message: "+System.currentTimeMillis(), routingKey);

6         Thread.sleep(1000);

7     }

8 }

发送成功

a.#也收不到消息

测试结果

通过以上六个测试，我们发现，topic中的通配符，只有在Queue绑定的时候才能起到通配符的作用，如果在发布消息的时候使用通配符，将作为普通的字符处理，发送的routingKey=a.* 并不能把消息发送到routingKey=a.b的Queue上，a.#同理，也不能把消息发送到routingKey=a.b.c的Queue上

3.实战

为了体现出#的作用，我们给一开始的需求增加一点难度，规则定位三层，<系统>.<级别>.<类型>，其中类型有两种，common和important

需求是，Q1监听core系统的所有日志和其他系统所有所有级别的important类型的日志以及error级的日志；Q2监听所有系统的info日志

首先改造一下Sender，让它可以通过多线程发送不同系统的日志消息

 package com.liyang.ticktock.rabbitmq;

 import java.util.HashMap;

 import java.util.Map;

 import java.util.Random;

 public class App {

     //声明一个类型MAP

     private static Map<Integer, String> typeMap;

     //声明一个Random

     private static Random random;

     //在静态代码块中初始化

     static {

         typeMap = new HashMap<>();

         typeMap.put(0, "important");

         typeMap.put(1, "common");

         random = new Random(System.currentTimeMillis());

     }

     /**

      * 获取一个随机类型的routingKey

      * @param system 系统名

      * @param level 日志级别

      * @return routingKey

      */

     public static String getRoutingKey(String system, String level) {

         return new StringBuilder()

                 .append(system).append(".")

                 .append(level).append(".")

                 .append(typeMap.get(random.nextInt(2)))

                 .toString();

     }

     /**

      * 新建一个线程，发送指定系统的消息

      * @param system

      */

     public static void createSender(final String system){

         new Thread(new Runnable() {

             //new一个Sender

             private  LogSender sender = new LogSender();

             @Override

             public void run() {

                 while(true){

                     long now = System.currentTimeMillis();

                     //通过当前时间生成错误级别

                     boolean info = now % 2 == 0;

                     //生成routingKey

                     String routingKey = getRoutingKey(system, info?"info":"error");

                     //发送消息

                     String msg = routingKey+"["+now+"]";

                     sender.sendMessage(msg, routingKey);

                     try {

                         //随机睡500-1000毫秒

                         int sleepTime = random.nextInt(1000);

                         Thread.sleep(sleepTime<500?500:sleepTime);

                     } catch (InterruptedException e) {

                         e.printStackTrace();

                     }

                 }

             }

         }).start();

     }

     public static void main(String[] args) throws InterruptedException {

         //开始发送core系统消息

         createSender("core");

         //开始发送biz系统消息

         createSender("biz");

     }

 }

上述代码可以用两个线程发送两个系统的随机日志消息

下面实现消费者

Q1监听core系统的所有日志和其他系统所有级别的important类型的日志以及error级的日志

把这句话拆分一下

core系统所有的日志，core.#
其他系统所有级别的important类型的日志，*.*.important
其他系统所有error级的日志，*.error.*

那么，我们只要给Q1绑定这三个routingKey就可以了，绑定多个routingKey我们在上一张已经验证过了

 package com.liyang.ticktock.rabbitmq;

 import java.io.IOException;

 import java.util.concurrent.TimeoutException;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import com.rabbitmq.client.AMQP;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.ConnectionFactory;

 import com.rabbitmq.client.Consumer;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 public class LogConsumer {

     private Logger logger = LoggerFactory.getLogger(LogConsumer.class);

     private ConnectionFactory factory;

     private Connection connection;

     private Channel channel;

     /**

      * 在构造函数中获取连接

      */

     public LogConsumer() {

         super();

         try {

             factory = new ConnectionFactory();

             factory.setHost("127.0.0.1");

             connection = factory.newConnection();

             channel = connection.createChannel();

             // 声明exchange，防止生产者没启动，exchange不存在

             channel.exchangeDeclare("logs","topic");

         } catch (Exception e) {

             logger.error(" [X] INIT ERROR!", e);

         }

     }

     /**

      * 提供个关闭方法，现在并没有什么卵用

      *

      * @return

      */

     public boolean closeAll() {

         try {

             this.channel.close();

             this.connection.close();

         } catch (IOException | TimeoutException e) {

             logger.error(" [X] CLOSE ERROR!", e);

             return false;

         }

         return true;

     }

     /**

      * 获取一个临时队列，并绑定到相应的routingKey上，消费消息

      */

     public void consume() {

         try {

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

             //core系统所有的日志

             channel.queueBind(queueName, "logs", "core.#");

             //其他系统所有级别的important类型的日志

             channel.queueBind(queueName, "logs", "*.*.important");

             //其他系统所有error级的日志

             channel.queueBind(queueName, "logs", "*.error.*");

             Consumer consumer = new DefaultConsumer(channel) {

                 @Override

                 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

                         byte[] body) throws IOException {

                     String message = new String(body, "UTF-8");

                     logger.debug(" [Q1] 打印日志："+message);

                 }

             };

             //这里自动确认为true，接收到消息后该消息就销毁了

             channel.basicConsume(queueName, true, consumer);

         } catch (IOException e) {

             e.printStackTrace();

         }

     }

 }

Q2监听所有系统的info日志

Q2应该绑定 *.info.*

 //所有系统的info日志

 channel.queueBind(queueName, "logs", "*.info.*");

把Q1 Q2打包成可执行jar，运行结果如下

可以看到需求已经正确实现了

4.结束语

到这里，RabbitMQ中四中exchagne类型：direct、topic、fanout、headers已经介绍了三种最常用的

headers不是很常用，放到Alternate Exchanges中一起介绍，后面还会介绍RabbitMQ的其他特性，如：TTL、Lazy Queue、Exchange To Exchange、Dead Lettering等，敬请期待

巴特西

RabbitMQ-从基础到实战（5）— 消息的交换（下）

转载请注明出处

0.目录

1.简介

2.Topic Exchange

测试1

测试2

测试3

测试4

测试5

测试6

测试结果

3.实战

4.结束语

最新文章

热门文章