Topic Exchange

topic模式按規則轉發是最靈活的一種匹配方式
# 匹配一個或者多個
* 匹配一個
rounting_key（路由鍵） : 訊息到交換機的時候，互動機會轉發到對應的佇列中，
那麼究竟轉發到哪個佇列，就要根據該路由鍵去進行匹配。路由鍵必須是一串字元用 . 隔開

如上圖所示:此類交換器使得來自不同的源頭的訊息可以到達一個對列，其實說的更明白一點就是模糊匹配的意思，例如：上圖中紅色對列的routekey為usa.#，#代表匹配任意字元，但是要想訊息能到達此對列，usa.必須匹配後面的#好可以隨意。圖中usa.news,usa.weather都能找到紅色佇列，符號“#”匹配一個或多個詞，符號“”匹配不多不少一個詞。因此“usa.#”能夠匹配到“usa.news.XXX”，但是“usa.” 只會匹配到“usa.XXX”。 注：交換器說到底是一個名稱與佇列繫結的列表。當訊息釋出到交換器時，實際上是由你所連線的通道，將訊息路由鍵同交換器上繫結的列表進行比較，最後路由訊息

生產者專案topic的bean配置

pom檔案匯入，和application.properties的配置和上篇文章相同，在這裡就不贅述了。

package com.rabbit.produce.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitConfig {

    /**
     * 定義佇列名稱
     */
    final static String message = "topic.message";
    final static String messages = "topic.messages";

    @Bean
    public Queue queueMessage() {
        return new Queue(TopicRabbitConfig.message);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(TopicRabbitConfig.messages);
    }

    /**
     * 這裡的exchange是交換機的名稱字串和傳送訊息時的名稱必須相同
     * this.rabbitTemplate.convertAndSend("exchange", "topic.1", context);
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("exchange");
    }

    /**
     * @param queueMessage 佇列
     * @param exchange     交換機
     *                     bindings 繫結交換機佇列資訊
     */
    @Bean
    Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }
}

 

生產者訊息傳送

package com.rabbit.produce.controller;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class TopicSender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send() {
        String context = "hi, i am message all";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("exchange", "topic.1", context);
    }

    public void send1() {
        String context = "hi, i am message 1";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
    }

    public void send2() {
        String context = "hi, i am messages 2";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);

    }
}
 

消費者專案（或本專案）接受訊息

佇列topic.messages監聽方法

• @RabbitListener 可以標註在類上面，需配合 @RabbitHandler 註解一起使用
• @RabbitListener 標註在類上面表示當有監聽到佇列topic.messages訊息的時候，就交給 @RabbitHandler 的方法處理，具體使用哪個方法處理，根據 MessageConverter 轉換後的引數型別

package com.rabbit.produce.controller;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author : lqf
 * @Description :
 * @date : Create in 18:37 2018/9/26
 */
@Component
@RabbitListener(queues = "topic.messages")
public class TopicReceiver2 {
    @RabbitHandler
    public void process(String message) {
        System.out.println("TopicReceiver2  : " + message);
    }
}

佇列topic.message監聽方法

package com.rabbit.produce.controller;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author : lqf
 * @Description :
 * @date : Create in 18:37 2018/9/26
 */
@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {
    @RabbitHandler
    public void process(String message) {
        System.out.println("TopicReceiver1  : " + message);
    }
}

測試

package com.rabbit.produce;

import com.rabbit.produce.controller.HelloSender;
import com.rabbit.produce.controller.HelloSender2;
import com.rabbit.produce.controller.TopicSender;
import com.rabbit.produce.entrty.User;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class BootRabbitmqProduceApplicationTests {

    @Autowired
    private TopicSender topicSender;

    @Test
    public void topicExchange() throws Exception {
        topicSender.send();
        topicSender.send1();
        topicSender.send2();
    }

}

測試結果

TopicReceiver1  : hi, i am message 1
TopicReceiver2  : hi, i am message all
TopicReceiver2  : hi, i am message 1
TopicReceiver2  : hi, i am messages 2

通過上方的測試結果可以看出topic.messages佇列也就是TopicReceiver2列印的內容監聽了所有routing_key為topic.#的訊息。看下面的配置資訊

@Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
    }

通過上方程式碼可以看到我們將佇列topic.messages繫結的routing_key為topic.#所以監聽器在監聽佇列topic.messages的時候匹配的是所有以topic開頭的rounting_key。這樣就符合了上方的列印結果