前言
本篇主要讲述Spring Boot与RabbitMQ的整合，内容非常简单，纯API的调用操作。 操作之间需要加入依赖Jar
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
ConnectionFactory配置
- spring boot 的RabbitAutoConfiguration 中使用RabbitConnectionFactoryCreator来创建connectionFactory , 相关属性配置在RabbitProperties, 以spring.rabbitmq 开头。
- 也可自定义 connectionFactory 如下
- 本例子使用 spring boot 内置的connectFactory
@Configuration
public class RabbitConfig {
public static final String EXCHANGE = "trmessageExchange";
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses("127.0.0.1:5672");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true); //如果需要confirm则设置为true
return connectionFactory;
}
}
消息生产者
/**
* rabbit 常量类
*/
public class RabbitConstans {
/**
* 交换机的名称
*/
public static final String EXCHANGE = "trmessageExchange";
/**
* 路由key 前缀
*/
public static final String PREROUTINGKEY = "trmessage";
/**
* 路由分割符号
*/
public static final String PREROUTINGSEG = ".";
}
TrmessageBean: 消息体
/**
* 消息体
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class TrmessageBean implements Serializable {
private static final long serialVersionUID = 3387041179388866333L;
/**
* 模块
*/
private String routingModule;
/**
* 子模块
*/
private String routingSub;
/**
* 消息数据
*/
private String context;
}
TrmessageSender:发送消息
/**
* 消息发送
*/
@Component
@Slf4j
public class TrmessageSender {
@Autowired
private AmqpTemplate rabbitTemplate;
/**
* 发送消息
*
* @param trmessageBean 消息实体
*/
public void send(TrmessageBean trmessageBean) {
String routingKey = RabbitConstans.PREROUTINGKEY + RabbitConstans.PREROUTINGSEG + trmessageBean.getRoutingModule() + RabbitConstans.PREROUTINGSEG + trmessageBean.getRoutingSub();
log.info(routingKey);
rabbitTemplate.convertAndSend(RabbitConstans.EXCHANGE, routingKey, trmessageBean);
}
}
- 通过使用RabbitTemplate来对开发者提供API操作
在发送消息时通过调用RabbitTemplate中的如下方法
public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData)
- exchange:交换机名称
- routingKey:路由关键字
- object:发送的消息内容
- correlationData:消息ID
消息消费者
消费者负责申明交换机(生产者也可以申明)、队列、两者的绑定操作。
交换机
/**
* 针对消费者配置
FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
HeadersExchange ：通过添加属性key-value匹配
DirectExchange:按照routingkey分发到指定队列
TopicExchange:多关键字匹配
*/
@Bean
TopicExchange exchange() {
return new TopicExchange(RabbitConstans.EXCHANGE);
}
在Spring Boot中交换机继承AbstractExchange类
队列
这里定义了两个队列。用于绑定不同的routing key
/**
* 查询相关的消息队列
*
* @return 消息队列
*/
@Bean
public Queue xfQueryQueue() {
return new Queue("xfQueryQueue");
}
/**
* 统计分析相关的queue
*
* @return 消息队列
*/
@Bean
public Queue xfAnalyQueue() {
return new Queue("xfAnalyQueue");
}
绑定
对以上定义的queue 进行绑定到不同的交换机上
@Bean
Binding bindingXfQueryQueue(@Qualifier("xfQueryQueue") Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("trmessage.xfQuery.*");
}
@Bean
Binding bindingXfAnalyQueue(@Qualifier("xfAnalyQueue") Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("trmessage.xfAnaly.*");
}
完成以上工作后，在spring boot中通过消息监听容器实现消息的监听，在消息到来时执行回调操作。
消息消费
@Component
@RabbitListener(queues = "xfAnalyQueue")
public class XfAnalyReceiver {
@RabbitHandler
public void process(TrmessageBean trmessageBean) {
System.out.println("xfAnaly receiver" + trmessageBean);
}
}
@Component
@RabbitListener(queues = "xfQueryQueue")
public class XfQueryReceiver {
@RabbitHandler
public void process(TrmessageBean trmessageBean) {
System.out.println("xfQuery receiver" + trmessageBean);
}
}
下面给出完整的配置文件：
@Configuration
public class RabbitConfig {
/**
* 查询相关的消息队列
*
* @return 消息队列
*/
@Bean
public Queue xfQueryQueue() {
return new Queue("xfQueryQueue");
}
/**
* 统计分析相关的queue
*
* @return 消息队列
*/
@Bean
public Queue xfAnalyQueue() {
return new Queue("xfAnalyQueue");
}
/**
* 创建交换机
*
* @return 交换机
*/
@Bean
TopicExchange exchange() {
return new TopicExchange(RabbitConstans.EXCHANGE);
}
/**
* 队列绑定并关联到RoutingKey
*
* @param queueMessages 队列名称
* @param exchange 交换机
* @return 绑定
*/
@Bean
Binding bindingXfQueryQueue(@Qualifier("xfQueryQueue") Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("trmessage.xfQuery.*");
}
@Bean
Binding bindingXfAnalyQueue(@Qualifier("xfAnalyQueue") Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("trmessage.xfAnaly.*");
}
}
application.yml
spring:
rabbitmq:
host: 205.0.3.94
port: 5672
username: admin
password: admin123
