RabbitMQ-從基礎到實戰（5）— 消息的交換（下）

轉載請注明出處

0.目錄

RabbitMQ-從基礎到實戰（1）— Hello RabbitMQ

RabbitMQ-從基礎到實戰（2）— 防止消息丟失

RabbitMQ-從基礎到實戰（3）— 消息的交換（上）

RabbitMQ-從基礎到實戰（4）— 消息的交換（中）

RabbitMQ-從基礎到實戰（6）— 與Spring集成

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的重要區別就是，它有兩個關鍵字

1. “*”星號，代表一個詞，比如上述規則：*.error 表示所有系統的error級別的日志
2. “#”井號，代表零個或多個詞，比如上述規則： *.# 表示所有系統的所有消息，與單獨一個#是等效的，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);  9         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.#"); 5 }

Sender的發送到logs上，routingKey設置為 a.b

public static void main( String[] args ) throws InterruptedException{
    LogSender sender = new LogSender();
    while(true){
        String routingKey = "a.b"; 5         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.*"; 5         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.#"); 65             //其他系統所有級別的important類型的日志
            channel.queueBind(queueName, "logs", "*.*.important"); 67             //其他系統所有error級的日志
            channel.queueBind(queueName, "logs", "*.error.*"); 69             
            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等，敬請期待