消息中間件——RabbitMQ（十）RabbitMQ整合SpringBoot實戰！（全）

時間  2019-11-09
標籤 消息 中間件 rabbitmq 整合 springboot 實戰 欄目 RabbitMQ 简体版
原文   原文鏈接

求關注
RabbitMQ整合SpringBoot實戰！（全）

前言

1. SpringBoot整合配置詳解

  • publisher-confirms,實現一個監聽器用於監聽Broker端給咱們返回的確認請求：RabbitTemplate.ConfirmCallbackjava

  • publisher-returns,保證消息對Broker端是可達的，若是出現路由鍵不可達的狀況，則使用監聽器對不可達的消息進行後續的處理，保證消息的路由成功：RabbitTemplate.ReturnCallbackgit

注意一點，在發送消息的時候對template進行配置mandatory=true保證監聽有效
生產端還能夠配置其餘屬性，好比發送重試，超時時間，次數，間隔等github

2. 代碼演示

2.1 生產端

2.1.1 新建項目springboot-producer

pom.xml面試

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.cp</groupId>
    <artifactId>springboot-producer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>springboot-producer</name>
    <description>springboot-producer</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

RabbitSender.java 消息生產者spring

@Component
public class RabbitSender {

    //自動注入RabbitTemplate模板類
    @Autowired
    private RabbitTemplate rabbitTemplate;  
    
    //回調函數: confirm確認
    final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.err.println("correlationData: " + correlationData);
            System.err.println("ack: " + ack);
            if(!ack){
                //能夠進行日誌記錄、異常處理、補償處理等
                System.err.println("異常處理....");
            }else {
                //更新數據庫，可靠性投遞機制
            }
        }
    };
    
    //回調函數: return返回
    final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
        @Override
        public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
                String exchange, String routingKey) {
            System.err.println("return exchange: " + exchange + ", routingKey: " 
                + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
        }
    };
    
    //發送消息方法調用: 構建Message消息
    public void send(Object message, Map<String, Object> properties) throws Exception {
        MessageHeaders mhs = new MessageHeaders(properties);
        Message msg = MessageBuilder.createMessage(message, mhs);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        //id + 時間戳 全局惟一  用於ack保證惟一一條消息,這邊作測試寫死一個。可是在作補償策略的時候，必須保證這是全局惟一的消息
        CorrelationData correlationData = new CorrelationData("1234567890");
        rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
    }
    
}

application.properties數據庫

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

2.1.2 操做管控臺

添加Exchange
Exchange-1添加成功apache

添加Queue
Queue-1編程

Exchange綁定Queue
綁定綁定成功springboot

修改routingKey，springboot改成spring，則進入的是returnCallback方法微信

returnCallback打印日誌-錯誤

這時候咱們發現報錯了

correlationData: CorrelationData [id=1234567890]
ack: false
異常處理....

2.1.3 解決ack爲false問題

這是因爲咱們在測試方法中進行測試，當測試方法結束，rabbitmq相關的資源也就關閉了，雖然咱們的消息發送出去，但異步的ConfirmCallback卻因爲資源關閉而出現了上面的問題。
加入Thread.sleep()便可解決。

@Test
public void testSender1() throws Exception {
     Map<String, Object> properties = new HashMap<>();
     properties.put("number", "12345");
     properties.put("send_time", simpleDateFormat.format(new Date()));
     rabbitSender.send("Hello RabbitMQ For Spring Boot!", properties);
     Thread.sleep(2000);
}

成功解決~
returnCallback打印日誌-正確

2.2 消費端

消費端核心配置：

簽收模式-手工簽收

spring.rabbitmq.listener.simple.acknowledge-mode=manual

設置監聽限制：最大10，默認5

spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

  • 首先配置手工確認模式，用於ACK的手工處理，這樣咱們能夠保證消息的可靠性送達，或者再消費端消費失敗的時候能夠作到重回隊列（不建議）、根據業務記錄日誌等處理。

  • 能夠設置消費端的監聽個數和最大個數，用於監控消費端的併發狀況

@RabbitListener註解使用

  • 消費端監聽@RabbitListener註解,這個對於在實際工做中很是的好用
  • @RabbitListener是一個組合註解，裏面能夠註解配置
  • @QueueBinding、@Queue、@Exchange直接經過這個組合註解一次性搞定消費端交換機、隊列、綁定、路由、而且配置監聽功能等。

@RabbitListener註解

好比在方法onMessage上加@RabbitListener註解，同時須要加另一個註解@RabbitHandler,代碼被消費者監聽。

創建綁定，在Value上寫上隊列，設置Exchange，是否持久化，設置Exchange的類型、表達式設置爲true以及路由key。經過這種簡單的方式，就能夠完成以前很複雜的代碼邏輯。同時建議將配置放入到配置文件中，動態獲取。若是mq中沒有相應的隊列、Exchange等，註解聲明也能夠建立它們，你們能夠自行測試！

2.2.1 新建項目springboot-consumer

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.cp</groupId>
    <artifactId>springboot-consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>springboot-consumer</name>
    <description>springboot-consumer</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>


</project>

RabbitReceiver.java 消息生產者

@Component
public class RabbitReceiver {

    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue-1", 
            durable="true"),
            exchange = @Exchange(value = "exchange-1", 
            durable="true", 
            type= "topic", 
            ignoreDeclarationExceptions = "true"),
            key = "springboot.*"
            )
    )
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {
        System.err.println("--------------------------------------");
        System.err.println("消費端Payload: " + message.getPayload());
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        //手工ACK,獲取deliveryTag
        channel.basicAck(deliveryTag, false);
    }
}

application.properties

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=user_cp
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/vhost_cp
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

運行Application，查看以前在生產端發送的消息，是否能被消費。

打印結果
打印結果
這裏以前因爲我測試的時候多發了消息，因此消費的時候會有這麼多。

3. 優化代碼

  • 自定義Java對象消息
  • @RabbitListener註解中的配置改成動態配置

@Payload：指定具體的消息體Body。
@Headers: 獲取Headers。

3.1 消費端優化

一、先定義一個Order對象

public class Order implements Serializable {

    private String id;
    private String name;
    
    public Order() {
    }
    public Order(String id, String name) {
        super();
        this.id = id;
        this.name = name;
    }
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
}

注意：咱們在傳輸對象的時候，必須序列化。不然會傳輸失敗。

二、RabbitReceiver添加監聽

/**
     * 
     *  spring.rabbitmq.listener.order.queue.name=queue-2
        spring.rabbitmq.listener.order.queue.durable=true
        spring.rabbitmq.listener.order.exchange.name=exchange-2
        spring.rabbitmq.listener.order.exchange.durable=true
        spring.rabbitmq.listener.order.exchange.type=topic
        spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
        spring.rabbitmq.listener.order.key=springboot.*
     * @param order
     * @param channel
     * @param headers
     * @throws Exception
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", 
            durable="${spring.rabbitmq.listener.order.queue.durable}"),
            exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", 
            durable="${spring.rabbitmq.listener.order.exchange.durable}", 
            type= "${spring.rabbitmq.listener.order.exchange.type}", 
            ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
            key = "${spring.rabbitmq.listener.order.key}"
            )
    )
    @RabbitHandler
    public void onOrderMessage(@Payload com.cp.springboot.entity.Order order, 
            Channel channel, 
            @Headers Map<String, Object> headers) throws Exception {
        System.err.println("--------------------------------------");
        System.err.println("消費端order: " + order.getId());
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        //手工ACK
        channel.basicAck(deliveryTag, false);
    }

已經將配置寫入到了application.properties中，進行動態獲取。也能夠像咱們公司同樣放入到配置中心當中。例如：攜程開源配置中心Apollo

三、application.properties

spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

3.2 生產端優化

一、一樣是一個Order對象，必須跟消費端的保持一致。

二、RabbitSender添加發送消息

//發送消息方法調用: 構建自定義對象消息
public void sendOrder(Order order) throws Exception {
    rabbitTemplate.setConfirmCallback(confirmCallback);
    rabbitTemplate.setReturnCallback(returnCallback);
    //id + 時間戳 全局惟一 
    CorrelationData correlationData = new CorrelationData("0987654321");
    rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
}

三、添加測試方法

@Test
public void testSender2() throws Exception {
     Order order = new Order("001", "第一個訂單");
     rabbitSender.sendOrder(order);
     //防止資源提早關閉，ConfirmCallback異步回調失敗
     Thread.sleep(2000);
}

4.測試

運行testSender2()方法。

生產端打印消息
生產端打印消息

消費端打印消息
消費端打印消息

至此，RabbitMQ整合SpringBoot完畢，在實際工做中，使用場景也是差很少的。

文末

歡迎關注我的微信公衆號：Coder編程
獲取最新原創技術文章和免費學習資料，更有大量精品思惟導圖、面試資料、PMP備考資料等你來領，方便你隨時隨地學習技術知識！
新建了一個qq羣：315211365，歡迎你們進羣交流一塊兒學習。謝謝了！也能夠介紹給身邊有須要的朋友。

文章收錄至
Github: https://github.com/CoderMerlin/coder-programming
Gitee: https://gitee.com/573059382/coder-programming
歡迎關注並star~
微信公衆號

參考文章：

《RabbitMQ消息中間件精講》

推薦文章：

消息中間件——RabbitMQ（七）高級特性全在這裏!（上）

消息中間件——RabbitMQ（八）高級特性全在這裏!（下）

消息中間件——RabbitMQ（九）RabbitMQ整合Spring AMQP實戰！（全）

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程