簡介

本節主要演示交換機的另一種型別：主題型別topic，直連線型別direct必須是生產者釋出訊息指定的routingKey和消費者在佇列繫結時指定的routingKey完全相等時才能匹配到佇列上，與direct不同,topic可以進行模糊匹配，可以使用星號*和井號#這兩個萬用字元來進行模糊匹配，其中星號可以代替一個單詞；主題型別的轉發器的訊息不能隨意的設定選擇鍵（routing_key），必須是由點隔開的一系列的識別符號組成。識別符號可以是任何東西，但是一般都與訊息的某些特性相關。一些合法的選擇鍵的例子：”quick.orange.rabbit”,你可以定義任何數量的識別符號，上限為255個位元組。 #井號可以替代零個或更多的單詞，只要能模糊匹配上就能將訊息對映到佇列中。當一個佇列的繫結鍵為#的時候，這個佇列將會無視訊息的路由鍵，接收所有的訊息

生產者

public class Producer {
    @Test
    public void testBasicPublish() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday"
 
);
        factory.setPassword("mengday");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // Routing 的路由規則使用直連線
        String EXCHANGE_NAME = "exchange.topic.x";
        String[] routingKeys = {"quick.orange.rabbit", "lazy.orange.elephant"
 
, "mq.erlang.rabbit", "lazy.brown.fox", "lazy."};
        for (String routingKey : routingKeys){
            String message = "Hello RabbitMQ - " + routingKey;

            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
        }

        // 關閉資源
        channel.close();
        connection.close();
    }
}

消費者1

public class Consumer1 {
    @Test
    public void testBasicConsumer1() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday");
        factory.setPassword("mengday");

        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        String EXCHANGE_NAME = "exchange.topic.x";
        String QUEUE_NAME = "queue.topic.q1";
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

        String[] routingKeys = {"*.orange.*"};
        for (int i = 0; i < routingKeys.length; i++) {
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, routingKeys[i]);
        }

        System.out.println("Consumer Wating Receive Message");
        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");
                System.out.println(" [C] Received '" + message + "', 處理業務中...");
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);

        Thread.sleep(1000000);
    }

}

消費者2

public class Consumer2 {
    @Test
    public void testBasicConsumer2() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday");
        factory.setPassword("mengday");

        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        String EXCHANGE_NAME = "exchange.topic.x";
        String QUEUE_NAME = "queue.topic.q2";
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

        String[] routingKeys = {"*.*.rabbit", "lazy.#"};
        for (int i = 0; i < routingKeys.length; i++) {
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, routingKeys[i]);
        }

        System.out.println("Consumer Wating Receive Message");
        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");
                System.out.println(" [C] Received '" + message + "', 處理業務中...");
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);

        Thread.sleep(1000000);
    }
}

執行效果

先啟動消費者，再啟動生產者。

有的訊息可以滿足多個模糊的路由鍵，具體會路由到哪個佇列，關於同時滿足條件以後再研究。

下面是一張topic的圖，描述的比較準確

預定義交換器

系統預定義了一個日誌交換機，名稱為“amq.rabbitmq.log”，型別為topic， 可以通過宣告一個匿名佇列，然後通過路由鍵(error warning info)等分別與佇列進行繫結，從而來消費訊息。 可以針對不同級別的訊息進行處理，如error級別的可以發郵件給相關負責人。 消費者需要自己實現，往佇列中發訊息是RabbitMQ自己傳送的，當有日誌訊息時RabbitMQ會自動發到amq.rabbitmq.log交換機中。

我的微信公眾號: