springboot2 整合 rabbitmq

1，依賴於配置

1，pom.xml 相關依賴

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

2，配置文件

spring:
  rabbitmq:
    addresses: 192.168.200.100:5672
    username: rabbit
    password: 123456
    virtual-host: /
    publisher-confirms: true
    publisher-returns: true
    template:
      mandatory: true
    listener:
      direct:
        acknowledge-mode: manual
      simple:
        concurrency: 3
        max-concurrency: 10

4，相關配置解析

基礎配置

spring.rabbitmq.host: 服務器地址
spring.rabbitmq.port: 服務器端口
spring.rabbitmq.addresses: 服務器連接，多個以逗號分隔，優先取 addresses，然后再取 host
spring.rabbitmq.username: 用戶名
spring.rabbitmq.password: 密碼
spring.rabbitmq.virtual-host: 虛擬主機
spring.rabbitmq.requested-heartbeat: 指定心跳超時，單位秒，0為不指定；默認60s
spring.rabbitmq.publisher-confirms: 是否啟用【發布確認】
spring.rabbitmq.publisher-returns: 是否啟用【發布返回】
spring.rabbitmq.connection-timeout: 連接超時，單位毫秒，0表示無窮大，不超時

https 訪問模式的 ssl 配置

spring.rabbitmq.ssl.enabled: 是否支持ssl
spring.rabbitmq.ssl.key-store: 指定持有SSL certificate的key store的路徑
spring.rabbitmq.ssl.key-store-password: 指定訪問key store的密碼
spring.rabbitmq.ssl.trust-store: 指定持有SSL certificates的Trust store
spring.rabbitmq.ssl.trust-store-password: 指定訪問trust store的密碼
spring.rabbitmq.ssl.algorithm: ssl使用的算法，例如，TLSv1.1

緩存配置

spring.rabbitmq.cache.channel.size: 緩存中保持的 channel 數量
spring.rabbitmq.cache.channel.checkout-timeout: 當緩存數量被設置時，從緩存中獲取一個channel的超時時間，單位毫秒；如果為0，則總是創建一個新channel
spring.rabbitmq.cache.connection.size: 緩存的連接數，只有是CONNECTION模式時生效
spring.rabbitmq.cache.connection.mode: 連接工廠緩存模式：CHANNEL 和 CONNECTION

消息監聽配置

spring.rabbitmq.listener.simple.auto-startup: 是否啟動時自動啟動容器
spring.rabbitmq.listener.simple.acknowledge-mode: 表示消息確認方式，其有三種配置方式，分別是none、manual（手動簽收） auto（自動簽收）
spring.rabbitmq.listener.simple.concurrency: 並發處理的消息數
spring.rabbitmq.listener.simple.max-concurrency: 並發處理的最大消息數
spring.rabbitmq.listener.simple.prefetch: 指定一個請求能處理多少個消息，如果有事務的話，必須大於等於transaction數量.
spring.rabbitmq.listener.simple.transaction-size: 指定一個事務處理的消息數量，最好是小於等於prefetch的數量.
spring.rabbitmq.listener.simple.default-requeue-rejected: 決定被拒絕的消息是否重新入隊；默認是true（與參數acknowledge-mode有關系）
spring.rabbitmq.listener.simple.idle-event-interval: 多少長時間發布空閑容器時間，單位毫秒

監聽重試

spring.rabbitmq.listener.simple.retry.enabled: 監聽重試是否可用
spring.rabbitmq.listener.simple.retry.max-attempts: 最大重試次數
spring.rabbitmq.listener.simple.retry.initial-interval: 第一次和第二次嘗試發布或傳遞消息之間的間隔
spring.rabbitmq.listener.simple.retry.multiplier: 應用於上一重試間隔的乘數
spring.rabbitmq.listener.simple.retry.max-interval: 最大重試時間間隔
spring.rabbitmq.listener.simple.retry.stateless: 重試是有狀態or無狀態

操作模板配置

spring.rabbitmq.template.mandatory: 啟用強制信息；默認false，這里必須設置為 true 才能是 return 模式生效
spring.rabbitmq.template.receive-timeout: receive() 操作的超時時間
spring.rabbitmq.template.reply-timeout: sendAndReceive() 操作的超時時間
spring.rabbitmq.template.retry.enabled: 發送重試是否可用
spring.rabbitmq.template.retry.max-attempts: 最大重試次數
spring.rabbitmq.template.retry.initial-interval: 第一次和第二次嘗試發布或傳遞消息之間的間隔
spring.rabbitmq.template.retry.multiplier: 應用於上一重試間隔的乘數
spring.rabbitmq.template.retry.max-interval: 最大重試時間間隔

2，發送消息 並 監聽處理消息

1, 作為消息發送的實體類，需要注意的是必須實現 Serializable 接口

package com.hwq.rabbitmq.entity;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

import java.io.Serializable;

@Getter
@Setter
@ToString
public class Order implements Serializable {

    private String id;
    private String name;

}

2，監聽器

package com.hwq.rabbitmq.listen;

import com.hwq.rabbitmq.entity.Order;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class TsetQueueListen {

    @RabbitListener(bindings = @QueueBinding(
            // 隊列        名稱                  持久化
            value = @Queue(value = "test.queue", durable = "true"),
            // 交換機            名稱                  持久化             交換機的模式     忽略異常
            exchange = @Exchange(value = "amq.direct", durable = "true", type = "direct", ignoreDeclarationExceptions = "true"),
            // 路由健 routerKey
            key = "test"
    ))
    @RabbitHandler
    public void onOrder(Message<Order> message, Channel channel) throws IOException {
        // 獲取消息標簽，用於手動簽收
        long tag = (long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        try {
            // 延遲一秒，模擬項目處理所需時間
            Thread.sleep(1000);
            // 獲取消息內容
            System.out.println(message.getPayload());
            // 手動簽收（正常）
            channel.basicAck(tag, false);
        } catch (InterruptedException ex) {
            // 手動簽收（異常）
            channel.basicNack(tag, false, false);
        }
    }

}

3，封裝發送消息的類，並對 確認模式和返回模式進行監聽

package com.hwq.rabbitmq.service;

import com.hwq.rabbitmq.entity.Order;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Service
public class RabbitSendService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 監聽 消息確認模式
    RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        System.out.println(correlationData);
        System.out.println("ack: " + ack);
        System.out.println(cause);
        if (!ack) {
            System.out.println("這里做一些異常處理");
        }
    };

    // 監聽 消息返回模式
    RabbitTemplate.ReturnCallback returnCallback = (message, replyCode, replyText, exchange, routingKey) -> {
        System.out.println("消息不可達預警");
    };

    /**
     * 發送消息
     * @param order 一個 java 類
     */
    public void sendOrder(Order order) {
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);

        CorrelationData cd = new CorrelationData();
        cd.setId(UUID.randomUUID().toString());

        // 發送纖細                    交換機        routerKey
        rabbitTemplate.convertAndSend("amq.direct", "test", order, cd);
    }

}

4，發送消息的控制器

package com.hwq.rabbitmq.controller;

import com.hwq.rabbitmq.entity.Order;
import com.hwq.rabbitmq.service.RabbitSendService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RequestMapping("queue")
@RestController
public class QueueController {

    @Autowired
    private RabbitSendService rabbitSendService;

    /**
     * 往消息隊列中發送數據
     */
    @RequestMapping("send")
    public String send() {
        Order order = new Order();
        order.setId("123456789123456798");
        order.setName("你的訂單");
        for (int i = 0; i < 20; i ++) {
            rabbitSendService.sendOrder(order);
        }
        return "ok";
    }

}

3，啟動之后訪問 http://ip:port/queue/send