5、Spring Boot 3.x 集成 RabbitMQ

一、前言

本篇主要是围绕着 Spring Boot 3.x 与 RabbitMQ 的集成，这边文章比较简单，RabbitMQ 的集成没有太大的变化，这篇文章主要是为了后续的 RabbitMQ 的动态配置做铺垫。
1、Docker 安装 RabbitMQ
2、Spring Boot 3.x 集成 RabbitMQ

二、Docker 安装 RabbitMQ

1、创建docker-network

# 创建docker网络，方便后续连通多个容器
docker network create local-net

2、拉取镜像并启动容器

# 搜索 rabbitmq 相关镜像
docker search rabbitmq
# 指定版本拉取 rabbitmq 镜像
docker pull rabbitmq:3-management
# 查看本地镜像
docker images# 启动容器命令
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -v /Users/kenny/docker/rabbitmq/data:/var/lib/rabbitmq/mnesia --network=local-net rabbitmq:3-management# 查看启动容器
docker ps# 查看启动日志
docker logs -f rabbitmq# 浏览器进入rabbitmq控制台
# http://localhost:15672
# 默认账号： guest
# 默认密码： guest

三、Spring Boot 3.x 集成 RabbitMQ

1、pom.xml

		<!-- RabbitMQ --><!-- Spring Boot RabbitMQ Starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

2、application.yml

spring:# RabbitMQ配置rabbitmq:host: localhostport: 5672username: guestpassword: guestlistener:simple:# 手动确认模式acknowledge-mode: manualretry:# 开启重试enabled: true# 最大重试次数max-attempts: 5# 首次重试时间间隔initial-interval: 1000# 重试时间间隔递增max-interval: 10000

3、初始化Exchange、Queue

RabbitFountConfig.java

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @program: chain* @description: RabbitMQ Fount 配置类* @author: Kenny.Qiu* @create: 2024/10/12 08:59*/
@Configuration
public class RabbitFountConfig {// exchangepublic final static String DEFAULT_EXCHANGE = "exchange.fount";// queuepublic final static String DEFAULT_QUEUE = "queue.fount";// routing keypublic final static String DEFAULT_ROUTING_KEY = "routing.key.fount";/*** 声明注册 fanout 模式的交换机** @return 交换机*/@Beanpublic FanoutExchange defalutFanoutExchange() {// durable:是否持久化,默认是false// autoDelete:是否自动删除return new FanoutExchange(DEFAULT_EXCHANGE, true, false);}/*** 声明队列** @return Queue*/@Beanpublic Queue defaultQueue() {// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除return new Queue(DEFAULT_QUEUE, true);}/*** 声明绑定交换机与队列** @return Binding*/@Beanpublic Binding defaultBinding() {return BindingBuilder.bind(defaultQueue()).to(defalutFanoutExchange());}
}

4、接收者

DefaultDirectReceiveQueueService

import com.chain.air.rpp.exchange.config.rabbit.RabbitDirectConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** @program: chain* @description: 默认 direct 队列接收组件* @author: Kenny.Qiu* @create: 2024/10/12 10:04*/
@Slf4j
@Component
@RabbitListener(queues = RabbitDirectConfig.DEFAULT_QUEUE)
public class DefaultDirectReceiveQueueService {/*** 默认 direct 队列接收消息** @param message 消息内容*/@RabbitHandlerpublic void messageReceive(String message) {log.info("默认 direct 队列接收消息：{}", message);}
}

5、发送者

RabbitProducerService

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;/*** @className: RabbitProducerService* @program: chain* @description: RabbitMQ 生产者 Service 组件* @author: kenny* @create: 2024-10-04 01:11* @version: 1.0.0*/
@Slf4j
@Service
public class RabbitProducerService {@Resourceprivate RabbitTemplate rabbitTemplate;/*** 向动态创建的队列发送消息** @param queueName 队列名称* @param message   消息内容*/public void sendMessageToQueue(String queueName, String message) {log.info("向队列：{}，发送消息：{}", queueName, message);rabbitTemplate.convertAndSend(queueName, message);}/*** 向动态创建的交换机发送消息** @param exchangeName 交换机名称* @param routingKey   路由键* @param message      消息内容*/public void sendMessageToExchange(String exchangeName, String routingKey, String message) {log.info("向交换机：{}，路由键：{}，发送消息：{}", exchangeName, routingKey, message);rabbitTemplate.convertAndSend(exchangeName, routingKey, message);}
}

6、Controller

RabbitController.java

import com.chain.air.rpp.exchange.rabbit.RabbitProducerService;
import jakarta.annotation.Resource;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;/*** @className: RabbitController* @program: chain* @description: RabbitMQ Controller组件* @author: kenny* @create: 2024-10-03 22:02* @version: 1.0.0*/
@RestController
@RequestMapping("/rabbit")
public class RabbitController {/*** RabbitMQ 生产者 Service 组件*/@Resourceprivate RabbitProducerService rabbitProducerService;/*** 发送消息到指定的队列** @param queueName 队列名称* @param message   消息内容* @return 处理结果*/@RequestMapping("/send")public String sendMessage(@RequestParam String queueName, @RequestParam String message) {rabbitProducerService.sendMessageToQueue(queueName, message);return "向队列：" + queueName + "，发送消息：" + message;}/*** 发送消息到指定的交换机** @param exchangeName 交换机名称* @param routingKey   路由键* @param message      消息内容* @return 处理结果*/@RequestMapping("/send/exchange")public String sendMessageToExchange(@RequestParam String exchangeName, @RequestParam String routingKey, @RequestParam String message) {rabbitProducerService.sendMessageToExchange(exchangeName, routingKey, message);return "向交换机：" + exchangeName + "，发送消息：" + message;}
}

四、测试

1、启动服务

2、查看RabbitMQ控制台，Exchange、Queue是否完成创建

3、通过接口发送消息

4、通过RabbitMQ控制台，往Exchange、Queue分别发送消息，服务是否接收到

下一篇：6、Spring Boot 3.x集成RabbitMQ动态交换机、队列