(四)、SpringBoot整合RabbitMQ
1.准备工作
(1).搭建一个SpringBoot项目
我们选择两个依赖: 第一个是:web 和 第二个是:RabbitMQ
(2).引入配置文件
# 服务端口 server: port: 8080 # RabbitMQ配置 spring: rabbitmq: host: 8.130.48.9 port: 5672 username: admin password: xxx virtual-host: /
2.订阅与发布 (Faout)
我们客户通过下订单,然后像四个队列中发送订单信息,服务接收者分别是: 短信、邮件、SMS短信、微信提示
(1).生产者
1. 交换机的类型为""。
package com.jsxs.service; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.UUID; /** * @Author Jsxs * @Date 2023/4/2 11:24 * @PackageName:com.jsxs.service * @ClassName: OrderService * @Description: TODO * @Version 1.0 */ @Service public class OrderService { @Resource // 获取rabbitMQ的服务 private RabbitTemplate rabbitTemplate; /** * * @param userId * @param productID * @param num */ public void makeOrder(String userId,String productID,int num){ //1. 生成订单 String orderID = UUID.randomUUID().toString().replace("-",""); System.out.println("订单号已经生产成功-"+orderID); //2. 设置交换机名字和路由 String exchangeName="fanout_order_producer"; String routineKey=""; //3. 发送消息 // 参数: (交换机、路由key或队列名、消息内容) rabbitTemplate.convertAndSend(exchangeName,routineKey,orderID); } }
(2).配置类
1. 声明交换机 2. 声明四个队列 3. 将交换机与队列进行绑定
package com.jsxs.config; 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 Jsxs * @Date 2023/4/2 11:52 * @PackageName:com.jsxs.config * @ClassName: RabbitMQConfig * @Description: TODO * @Version 1.0 */ @Configuration public class RabbitMQConfig { // 1. 声明注册fanout模式的交换机 @Bean public FanoutExchange fanoutExchange(){ // (交换机的名字、是否持久化。是否自动删除) return new FanoutExchange("fanout_order_producer",true,false); } // 2. 声明四个队列: 短信、SMS、WeChat、 @Bean public Queue SmsQueue(){ return new Queue("sms.fanout.queue",true); } @Bean public Queue MessageQueue(){ return new Queue("message.fanout.queue",true); } @Bean public Queue EmailQueue(){ return new Queue("email.fanout.queue",true); } @Bean public Queue WeChatQueue(){ return new Queue("wechat.fanout.queue",true); } // 3. 将队列与交换机进行绑定的操作 @Bean public Binding SmsBind(){ return BindingBuilder.bind(SmsQueue()).to(fanoutExchange()); } @Bean public Binding MessageBind(){ return BindingBuilder.bind(MessageQueue()).to(fanoutExchange()); } @Bean public Binding EmailBind(){ return BindingBuilder.bind(EmailQueue()).to(fanoutExchange()); } @Bean public Binding WechatBind(){ return BindingBuilder.bind(WeChatQueue()).to(fanoutExchange()); } }
(3).消费者
我们创建一个和proder同级的springboot目录
package com.jsxs.service.faout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:45 * @PackageName:com.jsxs.service.faout * @ClassName: EmailConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"email.fanout.queue"}) // 这个客户端的队列是哪个? public class EmailConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("email接收到的信息是:->"+message); } }
message
package com.jsxs.service.faout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: MessageConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"message.fanout.queue"}) // 这个客户端的队列是哪个? public class MessageConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("Message接收到的信息是:->"+message); } }
sms
package com.jsxs.service.faout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: SmsConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"sms.fanout.queue"}) // 这个客户端的队列是哪个? public class SmsConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("sms接收到的信息是:->"+message); } }
package com.jsxs.service.faout; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: WechatConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"wechat.fanout.queue"}) // 这个客户端的队列是哪个? public class WechatConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("wechat接收到的信息是:->"+message); } }
生产者为四个队列各发送1个信息
启动我们的消费者。进行接受消息。每一个客户端接受的消息都是一样的消息。
3.路由模式(Direct)
ctrl+r
: 在Idea中,我们可以选择一个变量名进行替换成另一个变量名。
(1).生产者
1. 指定我们的交换机的名字 2. 指定我们的路由Key
package com.jsxs.service; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.UUID; /** * @Author Jsxs * @Date 2023/4/2 11:24 * @PackageName:com.jsxs.service * @ClassName: OrderService * @Description: TODO : 分别给对应的路由Key发送消息. * @Version 1.0 */ @Service public class OrderService { @Resource // 获取rabbitMQ的服务 private RabbitTemplate rabbitTemplate; /** * * @param userId * @param productID * @param num */ public void makeOrder(String userId,String productID,int num){ //1. 生成订单 String orderID = UUID.randomUUID().toString().replace("-",""); System.out.println("订单号已经生产成功-"+orderID); //2. 设置交换机名字和路由 String exchangeName="direct_order_producer"; //3. 发送消息 // 参数: (交换机、路由key或队列名、消息内容) rabbitTemplate.convertAndSend(exchangeName,"sms","1"); rabbitTemplate.convertAndSend(exchangeName,"message","2"); rabbitTemplate.convertAndSend(exchangeName,"email","3"); rabbitTemplate.convertAndSend(exchangeName,"wechat","4"); } }
(2).配置文件
1. 一个SpringBoot中假如存在多个@Configuration那么我们可以 使用@Order()->进行指定我们的优先级顺序。 2. 声明交换机的类型和名字 (DirectExchange) 3. 声明四个队列 4. 将队列与交换机进行绑定并配置路由Key
package com.jsxs.config; 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; import org.springframework.core.annotation.Order; /** * @Author Jsxs * @Date 2023/4/2 14:50 * @PackageName:com.jsxs.config * @ClassName: DirectRabbitMQConfig * @Description: TODO : Direct 比 Foaunt 多了一个路由Key,类型在我们的SpringBoot中不用手动的进行设置,我们只需要在交换机声明的时候,进行寻找对应的类即可, * @Version 1.0 */ @Configuration @Order(1) // 假如说存在多个配置文件,我们可以用这个注解指定优先级顺序 public class DirectRabbitMQConfig { // 1. 声明注册direct模式的交换机 @Bean public DirectExchange directExchange(){ // (交换机的名字、是否持久化。是否自动删除) return new DirectExchange("direct_order_producer",true,false); } // 2. 声明四个队列: 短信、SMS、WeChat、 @Bean public Queue SmsQueue(){ return new Queue("sms.direct.queue",true); } @Bean public Queue MessageQueue(){ return new Queue("message.direct.queue",true); } @Bean public Queue EmailQueue(){ return new Queue("email.direct.queue",true); } @Bean public Queue WeChatQueue(){ return new Queue("wechat.direct.queue",true); } // 3. 将队列与交换机进行绑定的操作 @Bean public Binding SmsBind(){ return BindingBuilder.bind(SmsQueue()).to(directExchange()).with("sms"); } @Bean public Binding MessageBind(){ return BindingBuilder.bind(MessageQueue()).to(directExchange()).with("message"); } @Bean public Binding EmailBind(){ return BindingBuilder.bind(EmailQueue()).to(directExchange()).with("email"); } @Bean public Binding WechatBind(){ return BindingBuilder.bind(WeChatQueue()).to(directExchange()).with("wechat"); } }