rabbitMQ 基礎之Exchange Type

首先先介紹下rabbitmq的一些基礎概念

一、隊列、生產者、消費者

      隊列是RabbitMQ的內部對象，用於存儲消息。P（生產者）生產消息並投遞到隊列中，C（消費者）能夠從隊列中獲取消息並消費。

      

      多個消費者能夠訂閱同一個隊列，這時隊列中的消息會被平均分攤給多個消費者進行處理，而不是每一個消費者都收到全部的消息並處理。

      

二、Exchange、Binding

      剛纔咱們看到生產者將消息投遞到隊列中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的狀況是，生產者將消息發送到Exchange（交換器，下圖中的X），再經過Binding將Exchange與Queue關聯起來。

      

三、Exchange Type、Bingding key、routing key

      在綁定（Binding）Exchange與Queue的同時，通常會指定一個binding key。在綁定多個Queue到同一個Exchange的時候，這些Binding容許使用相同的binding key。

      生產者在將消息發送給Exchange的時候，通常會指定一個routing key，來指定這個消息的路由規則，生產者就能夠在發送消息給Exchange時，經過指定routing key來決定消息流向哪裏。

      RabbitMQ經常使用的Exchange Type有三種：fanout、direct、topic。

      fanout:把全部發送到該Exchange的消息投遞到全部與它綁定的隊列中。

      direct:把消息投遞到那些binding key與routing key徹底匹配的隊列中。

      topic:將消息路由到binding key與routing key模式匹配的隊列中。

示例代碼 git springboot-rabbitmq-exchange

四、direct模式實例

  4.一、添加pom文件

<!-- rabbitmq依賴 -->
  <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

4.二、添加application.yml配置

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: xxx
    password: xxxx
    publisher-confirms: true   #消息發送到交換機確認機制,是否確認回調

4.3 ExchangeConfig配置

@Configuration
public class ExchangeConfig {
    /**
     *   1.定義direct exchange，綁定queueTest
     *   2.durable="true" rabbitmq重啓的時候不須要建立新的交換機
     *   3.direct交換器相對來講比較簡單，匹配規則爲：若是路由鍵匹配，消息就被投送到相關的隊列
     *     fanout交換器中沒有路由鍵的概念，他會把消息發送到全部綁定在此交換器上面的隊列中。
     *     topic交換器你採用模糊匹配路由鍵的原則進行轉發消息到隊列中
     *   key: queue在該direct-exchange中的key值，當消息發送給direct-exchange中指定key爲設置值時，
     *   消息將會轉發給queue參數指定的消息隊列
     */
    @Bean
    public DirectExchange directExchange(){
        DirectExchange directExchange = new DirectExchange(RabbitMqConfig.EXCHANGE,true,false);
        return directExchange;
    }

    @Bean
    public TopicExchange topicExchange(){
        TopicExchange topicExchange = new TopicExchange(RabbitMqConfig.EXCHANGE_TOPIC,true,false);
        return topicExchange;
    }

    @Bean
    public FanoutExchange fanoutExchange (){
        FanoutExchange fanoutExchange = new FanoutExchange(RabbitMqConfig.EXCHANGE_FANOUT,true,false);
        return fanoutExchange;
    }

}

4.4 QueueConfig配置

@Configuration
public class QueueConfig {
    @Bean
    public Queue firstQueue() {
        /**
         durable="true" 持久化 rabbitmq重啓的時候不須要建立新的隊列
         auto-delete 表示消息隊列沒有在使用時將被自動刪除 默認是false
         exclusive  表示該消息隊列是否只在當前connection生效,默認是false
         */
        return new Queue("first-queue",true,false,false);
    }

    @Bean
    public Queue secondQueue() {
        return new Queue("second-queue",true,false,false);
    }

    @Bean
    public Queue topicQueue() {
        return new Queue("topic-queue",true,false,false);
    }

    @Bean
    public Queue topicQueue1() {
        return new Queue("topic-queue1",true,false,false);
    }

    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout-queue1",true,false,false);
    }

    @Bean
    public Queue fanoutQueue() {
        return new Queue("fanout-queue",true,false,false);
    }


}

4.五、RabbitMqConfig配置

@Configuration
public class RabbitMqConfig {
    /** 消息交換機的名字*/
    public static final String EXCHANGE = "exchangeTest";
    /** 消息交換機的名字*/
    public static final String EXCHANGE_TOPIC = "exchangeTopic";
    public static final String EXCHANGE_FANOUT = "exchangeFanout";

    /** 隊列key1*/
    public static final String ROUTINGKEY1 = "queue_one_key1";
    /** 隊列key2*/
    public static final String ROUTINGKEY2 = "queue_one_key2";
    public static final String ROUTINGKEY3 = "*.topic.*";
    public static final String ROUTINGKEY_TOPIC = "aaa.topic.*";


    @Autowired
    private QueueConfig queueConfig;
    @Autowired
    private ExchangeConfig exchangeConfig;

    /**
     * 鏈接工廠
     */
    @Autowired
    private ConnectionFactory connectionFactory;

    /**
     將消息隊列1和交換機進行綁定
     */
    @Bean
    public Binding binding_one() {
        return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY1);
    }

    /**
     * 將消息隊列2和交換機進行綁定
     */
    @Bean
    public Binding binding_two() {
        return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY2);
    }

    @Bean
    public Binding binding_topic() {
        return BindingBuilder.bind(queueConfig.topicQueue()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY3);
    }

    @Bean
    public Binding binding_topic1() {
        return BindingBuilder.bind(queueConfig.topicQueue1()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY_TOPIC);
    }

    @Bean
    public Binding binding_fanout() {
        return BindingBuilder.bind(queueConfig.fanoutQueue()).to(exchangeConfig.fanoutExchange());
    }

    @Bean
    public Binding binding_fanout_for_third() {
        return BindingBuilder.bind(queueConfig.fanoutQueue1()).to(exchangeConfig.fanoutExchange());
    }

    /**
     * queue listener  觀察 監聽模式
     * 當有消息到達時會通知監聽在對應的隊列上的監聽對象
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer_one(){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.addQueues(queueConfig.firstQueue());
        simpleMessageListenerContainer.setExposeListenerChannel(true);
        simpleMessageListenerContainer.setMaxConcurrentConsumers(5);
        simpleMessageListenerContainer.setConcurrentConsumers(1);
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設置確認模式手工確認
        return simpleMessageListenerContainer;
    }

   

    /**
     * 定義rabbit template用於數據的接收和發送
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        /**若使用confirm-callback或return-callback，
         * 必需要配置publisherConfirms或publisherReturns爲true
         * 每一個rabbitTemplate只能有一個confirm-callback和return-callback
         */
        template.setConfirmCallback(msgSendConfirmCallBack());
        //template.setReturnCallback(msgSendReturnCallback());
        /**
         * 使用return-callback時必須設置mandatory爲true，或者在配置中設置mandatory-expression的值爲true，
         * 可針對每次請求的消息去肯定’mandatory’的boolean值，
         * 只能在提供’return -callback’時使用，與mandatory互斥
         */
        //  template.setMandatory(true);
        return template;
    }

    /**
     * 消息確認機制
     * Confirms給客戶端一種輕量級的方式，可以跟蹤哪些消息被broker處理，
     * 哪些可能由於broker宕掉或者網絡失敗的狀況而從新發布。
     * 確認而且保證消息被送達，提供了兩種方式：發佈確認和事務。(二者不可同時使用)
     * 在channel爲事務時，不可引入確認模式；一樣channel爲確認模式下，不可以使用事務。
     * @return
     */
    @Bean
    public MsgSendConfirmCallBack msgSendConfirmCallBack(){
        return new MsgSendConfirmCallBack();
    }
}

4.6 DirectExchange 模式

綁定關係以下

/**
     將消息隊列1和交換機進行綁定
     */
    @Bean
    public Binding binding_one() {
        return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY1);
    }

    /**
     * 將消息隊列2和交換機進行綁定
     */
    @Bean
    public Binding binding_two() {
        return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY2);
    }

生產者發送對應消息

/**
     * DirectExchange 生產者 發送消息
     * @param uuid
     * @param message  消息
     */
    public void send(String uuid,Object message) {
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE, RabbitMqConfig.ROUTINGKEY2,
                message, correlationId);
    }

 消費者分別消費對應隊列信息 

@Component
public class SecondConsumer {

    @RabbitListener(queues = {"second-queue"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        // 處理消息
        System.out.println("Second Consumer {} handleMessage :"+message);
    }

}

消費結果

4.8 TopicExchange模式

因爲在綁定隊列時，綁定關係以下

public static final String ROUTINGKEY3 = "*.topic.*";
 public static final String ROUTINGKEY_TOPIC = "aaa.topic.*";
 @Bean
    public Binding binding_topic() {
        return BindingBuilder.bind(queueConfig.topicQueue()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY3);
    }

    @Bean
    public Binding binding_topic1() {
        return BindingBuilder.bind(queueConfig.topicQueue1()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY_TOPIC);
    }

 那麼此時生產者，發送aaa.topic.bbb的routing_key時。topicConsumer和topicConsumer1都能消費信息

/**
     * TopicExchange 生產者
     * @param uuid
     * @param message
     */
    public void sendTopic(String uuid,Object message) {
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_TOPIC, "aaa.topic.bbb",
                message, correlationId);
    }

方法的第一個參數是交換機名稱,第二個參數是發送的key,第三個參數是內容,RabbitMQ將會根據第二個參數去尋找有沒有匹配此規則的隊列,若是有,則把消息給它,若是有不止一個,則把消息分發給匹配的隊列(每一個隊列都有消息!),顯然在咱們的測試中,參數2匹配了兩個隊列,所以消息將會被髮放到這兩個隊列中,而監聽這兩個隊列的監聽器都將收到消息!那麼若是把參數2改成bbb.topic.bbb呢?顯然只會匹配到一個隊列,那麼TopicConsumer方法對應的監聽器收到消息!

 消費分別消費對應隊列信息

@Component
public class TopicConsumer {
    @RabbitListener(queues = {"topic-queue"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        // 處理消息
        System.out.println("TopicConsumer {} handleMessage :"+message);
    }
}

 

@Component
public class TopicConsumer1 {
    @RabbitListener(queues = {"topic-queue1"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        // 處理消息
        System.out.println("TopicConsumer1 {} handleMessage :"+message);
    }
}

消費者消費結果

4.9 fanoutExchange 模式

綁定關係以下

@Bean
    public Binding binding_fanout() {
        return BindingBuilder.bind(queueConfig.fanoutQueue()).to(exchangeConfig.fanoutExchange());
    }

    @Bean
    public Binding binding_fanout_for_third() {
        return BindingBuilder.bind(queueConfig.fanoutQueue1()).to(exchangeConfig.fanoutExchange());
    }

 消費者配置

@Component
public class FanoutConsumer {
    @RabbitListener(queues = {"fanout-queue"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        // 處理消息
        System.out.println("FanoutConsumer {} handleMessage :"+message);
    }
}

 

@Component
public class FanoutConsumer1 {

    @RabbitListener(queues = {"fanout-queue1"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        // 處理消息
        System.out.println("FanoutConsumer1 {} handleMessage :"+message);
    }

}

生產者

public void sendFanout(String uuid,Object message) {
        CorrelationData correlationId = new CorrelationData(uuid);
        //中間是設置路由規則，因爲是廣播模式，這個規則會被拋棄
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_FANOUT, "", message);
    }

Fanout Exchange形式又叫廣播形式,所以咱們發送到路由器的消息會使得綁定到該路由器的每個Queue接收到消息,這個時候就算指定了Key,或者規則(即上文中convertAndSend方法的參數2),也會被忽略!

消費結果

5.0 消息確認回調

public class MsgSendConfirmCallBack implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        System.out.println("MsgSendConfirmCallBack  , 回調id:" + correlationData);
        if (b) {
            System.out.println("消息發送成功");
        } else {
            System.out.println("消息發送失敗:" + s+"\n從新發送");
        }
    }
}