上节我们讲了rabbitmq的概念，原理，5种模式已经分别与Java进行整合，这节我们讲rabbitmq与springboot进行整合。（要提前准备好一个spring项目的基本框架）
整合时有两种方式创建交换机，队列及他们的绑定关系
1.添加配置类（加上@Configuration注解和@Bean注解）(topic模式，direct模式使用，也需要用到@RabbitListener，只起监听队列的作用)
2.用注解创建（@RabbitListener，注解中的内容比之增加，用来创建交换机，队列及他们的绑定关系，以及监听队列）
fanout模式
1.引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring–boot–starter–amqp</artifactId>
</dependency>
然后在spring项目中创建两个模块，一个生产者模块，一个消费者模块
2.在yml/properties文件进行相应的配置
分别在两个模块中都进行创建
server:
port: 8080
spring:
application:
name: RabbitMQ–demo
#rabbitmq配置
rabbitmq:
username: guest
password: guest
virtual–host: /
host: 127.0.0.1
port: 5672
3.创建配置类（创建相应的交换机与队列，且绑定关系）
分别在两个模块中都进行创建（防止一方没有配置类启动时报错）
package com.lx.producer.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;
@Configuration
public class RabbitMqConfiguration {
//创建一个fanout类型的交换机
@Bean
public FanoutExchange fanoutExchange(){
//第一个参数是交换机的名字，第二个是是否持久化，第三个是是否自动删除
return new FanoutExchange("fanout_order_exchange",true,false);
}
//创建三个队列
@Bean
public Queue smsQueue(){
return new Queue("sms.fanout.queue",true);
}
@Bean
public Queue duanxinQueue(){
return new Queue("duanxin.fanout.queue",true);
}
@Bean
public Queue emailQueue(){
return new Queue("email.fanout.queue",true);
}
//三个队列分别进行绑定关系
@Bean
public Binding smsBinding(){
return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
}
@Bean
public Binding duanxinBinding(){
return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
}
@Bean
public Binding emailBinding(){
return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
}
}
4.编写业务代码
在生产者模块中的service层中创建相应的生产信息业务类
package com.lx.producer.service;
import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.sql.SQLOutput;
import java.util.UUID;
@Service
public class OrderService {
@Resource
RabbitTemplate rabbitTemplate;
public void makeOrder(String userId,String productId,Integer numbers){
String orderId= UUID.randomUUID().toString();
System.out.println("订单生产成功："+orderId);
//通过MQ完成消息的发送
String exchangeName="fanout_order_exchange";
String routingKey="";
rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);
}
}
在消费者模块中创建3个相应的消费信息业务类
（注意：必须要在类上写上@RabbitListener注解，表示监听某个队列）
package com.lx.consumer.service;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(queues = {"duanxin.fanout.queue"})
public class FanoutDuanxinConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("duanxin.fanout—接收到的信息是："+message);
}
}
package com.lx.consumer.service;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(queues = {"email.fanout.queue"})
public class FanoutEmailConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("email.fanout—接收到的信息是："+message);
}
}
package com.lx.consumer.service;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(queues = {"sms.fanout.queue"})
public class FanoutSmsConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("sms.fanout—接收到的信息是："+message);
}
}
最后先启动rabbitmq，在启动两个项目，我们就能得到再消费者中得到生产者发送的消息了
direct模式
direct模式与fanout模式的代码差不多，只是要把配置类和生产者的业务处理类修改一下
配置类修改如下（把刚刚fanout命名都改为direct了）
package com.lx.producer.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMqConfiguration {
//创建一个direct类型的交换机
@Bean
public DirectExchange directExchangeExchange(){
//第一个参数是交换机的名字，第二个是是否持久化，第三个是是否自动删除
return new DirectExchange("direct_order_exchange",true,false);
}
//创建三个队列
@Bean
public Queue smsQueue(){
return new Queue("sms.direct.queue",true);
}
@Bean
public Queue duanxinQueue(){
return new Queue("duanxin.direct.queue",true);
}
@Bean
public Queue emailQueue(){
return new Queue("email.direct.queue",true);
}
//三个队列分别进行绑定关系
@Bean
public Binding smsBinding(){
return BindingBuilder.bind(smsQueue()).to(directExchangeExchange()).with("sms");
}
@Bean
public Binding duanxinBinding(){
return BindingBuilder.bind(duanxinQueue()).to(directExchangeExchange()).with("duanxin");
}
@Bean
public Binding emailBinding(){
return BindingBuilder.bind(emailQueue()).to(directExchangeExchange()).with("email");
}
}
改成穿件direct类型的交换机以及路由与队列绑定时加了key
生产者业务类
package com.lx.producer.service;
import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.sql.SQLOutput;
import java.util.UUID;
@Service
public class OrderService {
@Resource
RabbitTemplate rabbitTemplate;
public void makeOrder(String userId,String productId,Integer numbers){
String orderId= UUID.randomUUID().toString();
System.out.println("订单生产成功："+orderId);
//通过MQ完成消息的发送
String exchangeName="direct_order_exchange";
String routingKey1="sms";
String routingKey2="duanxin";
rabbitTemplate.convertAndSend(exchangeName,routingKey1,orderId);
rabbitTemplate.convertAndSend(exchangeName,routingKey2,orderId);
}
}
其余的代码不要变，把所有名字从fanout改为direct就行
topic模式
topic模式中我们使用注解的方式来创建交换机，队列以及它们之间的绑定关系，只需要把消费者业务代码及生产者业务代码修改下即可
消费者业务代码(修改@RabbitListener注解)
package com.lx.consumer.service;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "duanxin.topic.queue",durable = "true",autoDelete = "false"),
exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
key = "#.duanxin.#"
))
public class TopicDuanxinConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("duanxin.topic—接收到的信息是："+message);
}
}
package com.lx.consumer.service;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "topic.email.queue",durable = "true",autoDelete = "false"),
exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
key = "*.email.#"
))
public class TopicEmailConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("email.direct—接收到的信息是："+message);
}
}
package com.lx.consumer.service;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Service;
@Service
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "sms.topic.queue",durable = "true",autoDelete = "false"),
exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
key = "com.#"
))
public class TopicSmsConsumer {
@RabbitHandler
public void receiveMessage(String message){
System.out.println("sms.direct—接收到的信息是："+message);
}
}
生产者业务代码(只修改routing keyj就行，通配符模式)
package com.lx.producer.service;
import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.sql.SQLOutput;
import java.util.UUID;
@Service
public class OrderService {
@Resource
RabbitTemplate rabbitTemplate;
public void makeOrder(String userId,String productId,Integer numbers){
String orderId= UUID.randomUUID().toString();
System.out.println("订单生产成功："+orderId);
//通过MQ完成消息的发送
String exchangeName="topic_order_exchange";
String routingKey1="com.email.duanxin";
// String routingKey2="duanxin";
rabbitTemplate.convertAndSend(exchangeName,routingKey1,orderId);
// rabbitTemplate.convertAndSend(exchangeName,routingKey2,orderId);
}
}
