保险设计素材网站,什么网站是做家教的,天津建设人才网官网,万网董事长是谁文章目录 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、前情提要【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依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId
/dependency3、在application.yml进行配置
# 服务端口
server:port: 8080
# 配置rabbitmq服务
spring:rabbitmq:username: adminpassword: adminvirtual-host: /# ip地址为开启mq服务的服务器地址host: 47.104.141.27port: 56724定义订单的生产者
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 {//队列 起名TestDirectQueueBeanpublic 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交换机 起名TestDirectExchangeBeanpublic DirectExchange fanoutOrderExchange() {// return new DirectExchange(TestDirectExchange,true,true);return new DirectExchange(fanout_order_exchange, true, false);}//绑定 将队列和交换机绑定, 并设置用于匹配键TestDirectRoutingBeanpublic 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.运行测试
