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