个人学习系列|个人学习系列 - SpringBoot整合RabbitMQ 个人学习系列-SpringBoot整合Ra

研究一下消息队列，现在来简单搭建一下。

1. Docker搭建RabbitMQ
1.1 查询并下载RabbitMQ镜像

docker search rabbitmq

文章图片

// 选择可以访问web管理界面的tag docker pull rabbitmq:management

文章图片

1.2 运行RabbitMQ镜像

// 设置账号密码都为admin docker run -dit --name myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management

1.3 浏览器上访问服务器IP:15672 出现以下页面表示启动成功

文章图片

2. 搭建SpringBoot项目整合RabbitMQ
2.1 pom.xml 添加web和rabbitmq的依赖

org.springframework.boot spring-boot-starter-amqp org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.springframework.amqp spring-rabbit-test test

2.2 aplication.yml 将rabbitmq的地址用户名密码等配置上

spring: rabbitmq: host: localhost port: 5672 username: admin password: admin

2.3 新建交换机SenderConfig.java 有以下三种常用的交换机，我们这里使用第三种

DirectExchange
直连型交换机，根据消息携带的路由键，将消息转发给对应的队列
FanoutExchange
扇形交换机，接收到消息后会将消息转发到所有队列
TopicExchange
主题交换机，根据消息携带的路由键和交换机与队列绑定键的规则，将消息转发给对应的队列
规则：
*（星号）：表示一个字符必须出现
#（井号）：表示任意数量的字符

/** * 交换机 * @author zhouzhaodong */ @Configuration public class SenderConfig {/** * ----- 交换机 ----- * 参数意义: *name: 名称 *durable: 持久化 *autoDelete: 自动删除 */ @Bean public TopicExchange topicExchange() { return new TopicExchange("topicExchange", true, false); }/** * ----- 队列 ----- */ @Bean public Queue queueOne() { return new Queue("queueOne", true); }@Bean public Queue queueTwo() { return new Queue("queueTwo", true); }@Bean public Queue queueThree() { return new Queue("queueThree", true); }/** * ----- 绑定 ----- * routingKey就是路由规则，消息对应的队列，用来区分不同的消息队列 */ @Bean public Binding bindingFanoutOne() { return BindingBuilder.bind(queueOne()).to(topicExchange()).with("topic_one"); }@Bean public Binding bindingFanoutTwo() { return BindingBuilder.bind(queueTwo()).to(topicExchange()).with("topic_two"); }@Bean public Binding bindingFanoutThree() { return BindingBuilder.bind(queueThree()).to(topicExchange()).with("topic_one"); }}

2.4 发送者 SenderController.java

/** * 消息发送者 * * @author zhouzhaodong */ @RestController public class SenderController {@Resource AmqpTemplate amqpTemplate; Logger logger = LoggerFactory.getLogger(SenderController.class); @RequestMapping(value = "https://www.it610.com/send") public String sendMessage(String message) { logger.info("消息发送开始时间：" + new Date()); // 这里convertAndSend第一个参数是交换机的名称 // 第二个参数可以是routingKey // 最后一个参数就是要发送的消息 amqpTemplate.convertAndSend("topicExchange", "topic_one", message); return "发送成功"; }}

2.5 消费者 ReceiverController.java

/** * 消费者 * @author zhouzhaodong */ @Component public class ReceiverController {Logger logger = LoggerFactory.getLogger(ReceiverController.class); @RabbitHandler @RabbitListener(queues = "queueOne") public void processA(String message){ logger.info("queueOne接收消息时间为：" + new Date()); logger.info("queueOne接收消息为：" + message); }@RabbitHandler @RabbitListener(queues = "queueTwo") public void processB(String message){ logger.info("queueTwo接收消息时间为：" + new Date()); logger.info("queueTwo接收消息为：" + message); }@RabbitHandler @RabbitListener(queues = "queueThree") public void processC(String message){ logger.info("queueThree接收消息时间为：" + new Date()); logger.info("queueThree接收消息为：" + message); }}

3. 启动项目进行测试
3.1 调用生产者接口