- 整体核心
- 目标
- 实现步骤
- 生产者
- 创建生产者工程：springboot-rabbitmq-fanout-producer
- 在pom.xml中引入依赖
- 在application.yml进行配置
- 定义订单的生产者
- 绑定关系
- 进行测试
- 消费者
- 创建消费者工程：springboot-rabbitmq-fanout-consumer
- 引入依赖pom.xml
- 在application.yml进行配置
- 消费者 – 邮件服务
- 消费者 – 短信服务
- 消费者 – 微信服务
- 启动服务SpringbootRabbitmqFanoutConsumerApplication，查看效果
整体核心
目标
:::info
使用springboot完成rabbitmq的消费模式-Fanout
:::
实现步骤
:::info
1：创建生产者工程：springboot-rabbitmq-fanout-producer
2：创建消费者工程：springboot-rabbitmq-fanout-consumer
3：引入spring-boot-rabbitmq的依赖
4：进行消息的分发和测试
5：查看和观察web控制台的状况
:::
生产者
创建生产者工程：springboot-rabbitmq-fanout-producer
在pom.xml中引入依赖
org.springframework.boot
spring-boot-starter-amqp
org.springframework.boot
spring-boot-starter-web
在application.yml进行配置
服务端口
server:
port: 8080
配置rabbitmq服务
spring:
rabbitmq:
username: admin
password: admin
virtual-host: /
host: 47.104.141.27
port: 5672
定义订单的生产者
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 {
@Autowired
private RabbitTemplate rabbitTemplate;
// 1: 定义交换机
private String exchangeName = "fanout_order_exchange";
// 2: 路由key
private 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 fanout
rabbitTemplate.convertAndSend(exchangeName, routeKey, orderNumer);
}
}
绑定关系
:::info
绑定关系方式有两种：一种是配置类，一种是注解@RabbitListener(bindings =@QueueBinding
推荐使用配置类
生产者和消费者中只需要一边添加配置类就可以了，建议消费者中添加，因为通常消费者先启动
:::
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutproducer.service;
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;
/**
* @Author : JCccc
* @CreateTime : 2019/9/3
* @Description :
**/
@Configuration
public class FanoutRabbitConfig {
// 声明注册fanout模式的交换机
@Bean
public FanoutExchange fanoutOrderExchange() {
return new FanoutExchange("fanout_order_exchange", true, false);
}
// 声明队列
@Bean
public Queue emailQueue() {
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
//一般设置一下队列的持久化就好,其余两个就是默认false
return new Queue("email.fanout.queue", true);
}
@Bean
public Queue smsQueue() {
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
//一般设置一下队列的持久化就好,其余两个就是默认false
return new Queue("sms.fanout.queue", true);
}
@Bean
public Queue weixinQueue() {
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
//一般设置一下队列的持久化就好,其余两个就是默认false
return new Queue("weixin.fanout.queue", true);
}
//绑定 将队列和交换机绑定
@Bean
public Binding emailBinding() {
return BindingBuilder.bind(emailQueue()).to(fanoutOrderExchange());
}
@Bean
public Binding smsBinding() {
return BindingBuilder.bind(smsQueue()).to(fanoutOrderExchange());
}
@Bean
public Binding weixinBinding() {
return BindingBuilder.bind(weixinQueue()).to(fanoutOrderExchange());
}
}
进行测试
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 {
@Autowired
OrderService orderService;
@Test
public 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);
}
}
}
消费者
创建消费者工程：springboot-rabbitmq-fanout-consumer
引入依赖pom.xml
org.springframework.boot
spring-boot-starter-amqp
org.springframework.boot
spring-boot-starter-web
在application.yml进行配置
服务端口
server:
port: 8081
配置rabbitmq服务
spring:
rabbitmq:
username: admin
password: admin
virtual-host: /
host: 47.104.141.27
port: 5672
消费者 – 邮件服务
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutconsumer.consumer;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;
// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
// email.fanout.queue 是队列名字，这个名字你可以自定随便定义。
value = @Queue(value = "email.fanout.queue",autoDelete = "false"),
// order.fanout 交换机的名字 必须和生产者保持一致
exchange = @Exchange(value = "fanout_order_exchange",
// 这里是确定的rabbitmq模式是：fanout 是以广播模式 、 发布订阅模式
type = ExchangeTypes.FANOUT)
))
@Component
public class EmailService {
// @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值
@RabbitHandler
public void messagerevice(String message){
// 此处省略发邮件的逻辑
System.out.println("email-------------->" + message);
}
}
消费者 – 短信服务
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutconsumer.consumer;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;
// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
// email.fanout.queue 是队列名字，这个名字你可以自定随便定义。
value = @Queue(value = "sms.fanout.queue",autoDelete = "false"),
// order.fanout 交换机的名字 必须和生产者保持一致
exchange = @Exchange(value = "fanout_order_exchange",
// 这里是确定的rabbitmq模式是：fanout 是以广播模式 、 发布订阅模式
type = ExchangeTypes.FANOUT)
))
@Component
public class SMSService {
// @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值
@RabbitHandler
public void messagerevice(String message){
// 此处省略发邮件的逻辑
System.out.println("sms-------------->" + message);
}
}
消费者 – 微信服务
package com.xuexiangban.rabbitmq.springbootrabbitmqfanoutconsumer.consumer;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;
// bindings其实就是用来确定队列和交换机绑定关系
@RabbitListener(bindings =@QueueBinding(
// email.fanout.queue 是队列名字，这个名字你可以自定随便定义。
value = @Queue(value = "weixin.fanout.queue",autoDelete = "false"),
// order.fanout 交换机的名字 必须和生产者保持一致
exchange = @Exchange(value = "fanout_order_exchange",
// 这里是确定的rabbitmq模式是：fanout 是以广播模式 、 发布订阅模式
type = ExchangeTypes.FANOUT)
))
@Component
public class WeixinService {
// @RabbitHandler 代表此方法是一个消息接收的方法。该不要有返回值
@RabbitHandler
public void messagerevice(String message){
// 此处省略发邮件的逻辑
System.out.println("weixin-------------->" + message);
}
}
启动服务SpringbootRabbitmqFanoutConsumerApplication，查看效果
Original: https://www.cnblogs.com/wl3pb/p/16577897.html
Author: 清风（学习-踏实）
Title: 140_SpringBoot案例-fanout发布订阅模式
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/581235/
转载文章受原作者版权保护。转载请注明原作者出处！