文章目录
- 1、前情提要【RabbitMQ】
- 2、RabbitMQ-SpringBoot案例 -fanout模式
- 2.1 实现架构总览
- 2.2 具体实现
- 2.2.1生产者
- 2.2.1消费者
1、前情提要【RabbitMQ】
【RabbitMQ】消息队列-RabbitMQ篇章
RabbitMQ实现流程
2、RabbitMQ-SpringBoot案例 -fanout模式
2.1 实现架构总览
实现步骤:
1:创建生产者工程:sspringboot-rabbitmq-fanout-producer
2:创建消费者工程:springboot-rabbitmq-fanout-consumer
3:引入spring-boot-rabbitmq的依赖
4:进行消息的分发和测试
5:查看和观察web控制台的状况
2.2 具体实现
2.2.1生产者
在这之前提前开好服务器,并且启动mq的服务,可参考上面的链接
- 1、创建生产者springboot工程:sspringboot-rabbitmq-fanout-producer
- 2、导入启动(web、mq)依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>
- 3、在application.yml进行配置
# 服务端口
server:port: 8080
# 配置rabbitmq服务
spring:rabbitmq:username: adminpassword: adminvirtual-host: /# ip地址为开启mq服务的服务器地址host: 47.104.141.27port: 5672
- 4:定义订单的生产者
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutproducer.service;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.UUID;
/*** @author: 学相伴-飞哥* @description: OrderService* @Date : 2021/3/4*/
@Component
public class OrderService {@Autowiredprivate RabbitTemplate rabbitTemplate;// 1: 定义交换机private String exchangeName = "fanout_order_exchange";// 2: 路由keyprivate String routeKey = "";public void makeOrder(Long userId, Long productId, int num) {// 1: 模拟用户下单String orderNumer = UUID.randomUUID().toString();// 2: 根据商品id productId 去查询商品的库存// int numstore = productSerivce.getProductNum(productId);// 3:判断库存是否充足// if(num > numstore ){ return "商品库存不足..."; }// 4: 下单逻辑// orderService.saveOrder(order);// 5: 下单成功要扣减库存// 6: 下单完成以后System.out.println("用户 " + userId + ",订单编号是:" + orderNumer);// 发送订单信息给RabbitMQ fanoutrabbitTemplate.convertAndSend(exchangeName, routeKey, orderNumer);}
}
- 4、配置类绑定交换机和队列的关系
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutproducer.service;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*** @Author : JCccc* @CreateTime : 2019/9/3* @Description :**/
@Configuration
public class DirectRabbitConfig {//队列 起名:TestDirectQueue@Beanpublic Queue emailQueue() {// durable:是否持久化,默认是false,持久化队列:会被存储在磁盘上,当消息代理重启时仍然存在,暂存队列:当前连接有效// exclusive:默认也是false,只能被当前创建的连接使用,而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除,当没有生产者或者消费者使用此队列,该队列会自动删除。// return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认falsereturn new Queue("email.fanout.queue", true);}@Beanpublic Queue smsQueue() {// durable:是否持久化,默认是false,持久化队列:会被存储在磁盘上,当消息代理重启时仍然存在,暂存队列:当前连接有效// exclusive:默认也是false,只能被当前创建的连接使用,而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除,当没有生产者或者消费者使用此队列,该队列会自动删除。// return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认falsereturn new Queue("sms.fanout.queue", true);}@Beanpublic Queue weixinQueue() {// durable:是否持久化,默认是false,持久化队列:会被存储在磁盘上,当消息代理重启时仍然存在,暂存队列:当前连接有效// exclusive:默认也是false,只能被当前创建的连接使用,而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除,当没有生产者或者消费者使用此队列,该队列会自动删除。// return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认falsereturn new Queue("weixin.fanout.queue", true);}//Direct交换机 起名:TestDirectExchange@Beanpublic DirectExchange fanoutOrderExchange() {// return new DirectExchange("TestDirectExchange",true,true);return new DirectExchange("fanout_order_exchange", true, false);}//绑定 将队列和交换机绑定, 并设置用于匹配键:TestDirectRouting@Beanpublic Binding bindingDirect1() {return BindingBuilder.bind(weixinQueue()).to(fanoutOrderExchange()).with("");}@Beanpublic Binding bindingDirect2() {return BindingBuilder.bind(smsQueue()).to(fanoutOrderExchange()).with("");}@Beanpublic Binding bindingDirect3() {return BindingBuilder.bind(emailQueue()).to(fanoutOrderExchange()).with("");}
}
- 6.测试
向队列发送10条消息
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutproducer;
import com.xuexiangban.rabbitmq.springbootrabbitmqfanoutproducer.service.OrderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class SpringbootRabbitmqFanoutProducerApplicationTests {@AutowiredOrderService orderService;@Testpublic void contextLoads() throws Exception {for (int i = 0; i < 10; i++) {Thread.sleep(1000);Long userId = 100L + i;Long productId = 10001L + i;int num = 10;orderService.makeOrder(userId, productId, num);}}
}
启动测试,此时进入mq的web页面,查看交换机和队列是否绑定上了,查看是否有队列消息
向所有队列中都加入10条消息—说明交换机和队列绑定没问题
2.2.1消费者
参照生产者的创建方法,选择在平级目录下创建:
springboot-order-rabbitmq-consumber
2. 修改配置文件
# 服务器
server:
# 端口要改成不冲突的port: 8081/
# rabbitmq配置
spring:rabbitmq:username: adminpassword: adminvirtual-host: /host: 192.168.157.128 #127.0.0.1port: 5672
- 创建FanoutEmailConsumer、FanoutNoteConsumer、FanoutSMSConsumer消费者接收
4.运行测试