SpringBoot應用操作Rabbitmq（direct高級操作）

一、首先聲明完成任務架構，通過direct訂閱/發布的模式進行生產消費。

 

 a、消息生產指定交換器和路由key

 b、消費者綁定交換器，路由key和隊列的關系（集群監控收到的消息不重復）

二、實戰演練

1、首先第一步是引入消息隊列的依賴包

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>

2、添加配置application.yml文件或者properties文件

spring:
  application:
    #指定應用的名字
    name: rabbit-add
  #配置rabbitmq
  rabbitmq:
  #鏈接主機
    host: 127.0.0.1
  #端口
    port: 5672
  #已經授權的用戶賬號密碼
    username: user
    password: user
  #指定的虛擬主機，默認/,
    virtual-host: my_vhost


# 自定義配置應用於direct交換器
mq:
  config:
   #自定義交換器名稱
     exchange: log.direct
     queue:
        #自定義error和info隊列名稱
        errorName: log.error
        infoName: log.info
        #自定義error和info路由鍵的名稱
        routingInfoKey: info.routing.key
        routingErrorKey: error.routing.key

3、創建隊列的消費者，

注意：1、當前代碼autoDelete屬性為false，創建的是臨時隊列

　　   2、RabbitHandler的isDefault屬性的使用，默認false，可能會出現監聽這未找到方法的循環異常

         3、集群模式將某個消費者復制多份即可

　a、error隊列的消費者

package com.niu.direct;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
 * @author niunafei
 * @function
 * @email [e-mail]
 * @date 2020/4/28  7:20 PM
 * @RabbitListener 自定義監聽事件
 * @QueueBinding 綁定交換器與隊列的關系value 指定隊列exchange指定交換器
 * value= @Queue 指定配置隊列的信息 value隊列名稱 autoDelete是否是臨時隊列
 * exchange= @Exchange 指定交換器 value指定交換器名稱 type交換器類型
 * key  指定路由鍵
 */
@Component
@Slf4j
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${mq.config.queue.errorName}", autoDelete = "true"),
                exchange = @Exchange(value = "${mq.config.exchange}", type = ExchangeTypes.DIRECT),
                key = "${mq.config.queue.routingErrorKey}")
)
public class ErrorReceiver {

    /**
     * 設置監聽方法
     *  @RabbitHandler 聲明監聽方法是下面的 isDefault屬性是默認false接受的完整對象，true接受body體
     *
     * @param msg
     */
    @RabbitHandler(isDefault = true)
    public void process(String msg) {
        log.info("接受到消息：error {}", msg);
    }
}

　　b、info隊列的消費者

package com.niu.direct;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
 * @author niunafei
 * @function
 * @email [e-mail]
 * @date 2020/4/28  7:20 PM
 * @RabbitListener 自定義監聽事件
 * @QueueBinding 綁定交換器與隊列的關系value 指定隊列exchange指定交換器
 * value= @Queue 指定配置隊列的信息 value隊列名稱 autoDelete是否是臨時隊列
 * exchange= @Exchange 指定交換器 value指定交換器名稱 type交換器類型
 * key  指定路由鍵
 */
@Component
@Slf4j
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(
                        value = "${mq.config.queue.infoName}", autoDelete = "true"
                ),
                exchange = @Exchange(
                        value = "${mq.config.exchange}", type = ExchangeTypes.DIRECT),
                key = "${mq.config.queue.routingInfoKey}")
)
public class InfoReceiver {

    /**
     * 設置監聽方法
     *
     * @param msg
     * @RabbitHandler 聲明監聽方法是下面的 isDefault屬性是默認false接受的完整對象，true接受body體
     */
    @RabbitHandler(isDefault = true)
    public void process(String msg) {
        log.info("接受到消息：info {}", msg);
    }
}

4、創建消息的生產者,這里為了方便就同一個方法生產

package com.niu.direct;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @author niunafei
 * @function
 * @email [e-mail]
 * @date 2020/4/29  9:44 AM
 */
@Component
public class Sender {
    /**
     * spring整合的操作類
     * Message 發送的消息對象
     * void send(Message var1) throws AmqpException;
     * <p>
     * var1 路由鍵 Message 發送的消息對象
     * void send(String var1, Message var2) throws AmqpException;
     * <p>
     * var1 指定交換器名稱 var2 路由鍵 Message 發送的消息對象
     * void send(String var1, String var2, Message var3) throws AmqpException;
     *
     * convertAndSend() 方法不需要指定MessageProperties屬性即可發布
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${mq.config.queue.routingInfoKey}")
    private String routingInfoKey;
    @Value("${mq.config.queue.routingErrorKey}")
    private String routingErrorKey;
    @Value("${mq.config.exchange}")
    private String exchange;

    public void send(String msg) {
        Message message = new Message(msg.getBytes(), new MessageProperties());
        //需要指定交換器和路由鍵就可以轉發
        rabbitTemplate.send(exchange, routingInfoKey, message);
        rabbitTemplate.send(exchange, routingErrorKey, message);
    }

}

 

5測試結果：

 

注意：ack確認機制，容易產生數據丟失，和產生內存泄漏，消費者進行死循環，配置這兩個屬性進行確認。

1、autoDelete屬性設置為false

@Queue(value = "${mq.config.queue.orderName}", autoDelete = "false"

2、消費者進行死循環問題

 

docker安裝rabbitmq：rabbitMQ安裝docker版 /權限管理命令

簡單應用來這里吧： SpringBoot應用操作Rabbitmq

簡單應用來這里吧： SpringBoot應用操作Rabbitmq（direct高級操作）

簡單應用來這里吧：SpringBoot應用操作Rabbitmq（topic交換器高級操作）   

簡單應用來這里吧：SpringBoot應用操作Rabbitmq（fanout廣播高級操作）