目录
一、引入依赖
二、配置rabbitmq的连接信息等
1、生产者配置
2、消费者配置
三、设置消息转换器
四、生产者代码示例
1、配置交换机和队列信息
2、生产消息代码
五、消费者代码示例
1、消费层代码
2、业务层代码
在分布式系统中，消息队列是一种重要的通信方式，它能够有效地将消息从一个应用程序传递到另一个应用程序。RabbitMQ是一款流行的开源消息队列系统，简单易用且功能强大。本文将介绍如何使用SpringBoot快速整合RabbitMQ，实现消息的发送和接收。
交换机： 主要负责接收生产者发送的消息，并根据特定的规则将这些消息路由到一个或多个队列中。交换机的类型有：
- Fanout Exchange（扇出交换机）
Fanout交换机会将接收到的所有消息广播到它知道的所有队列中。这种类型的交换机不考虑路由键，只是简单地将消息复制到所有绑定的队列中。适用于不需要选择性地发送消息给特定队列的情况，例如，广播系统通知或有多个服务需要消费同一份数据的场景。
- Direct Exchange（直连交换机）
Direct交换机根据消息的路由键将消息发送到与之匹配的队列中。只有当路由键与绑定关键字完全匹配时，消息才会被路由到相应的队列。适合于精确控制消息投递的场景，如特定的服务或功能模块只关心特定类型的消息。
- Topic Exchange（主题交换机）
Topic交换机允许更复杂的匹配规则，通过模式匹配的方式将消息路由到一个或多个队列。路由键和绑定键都使用点分隔的字符串，可以包含特殊字符如“#”和“*”来实现模糊匹配。"*"用于匹配一个单词，而“#”则用于匹配零个或多个单词。适合于需要按内容分类消息的系统，如日志处理系统，可以根据日志等级或来源将日志消息分发到不同的队列。
- Headers Exchange（头交换机）
Headers交换机使用消息头的一组键值对来决定消息应该被路由到哪个队列。这种交换机允许更细粒度的路由控制，但配置和使用较为复杂。适合需要基于消息多个属性来动态决定路由的场景，例如某些高级的路由策略或复杂的事件处理系统。
队列：主要用于存储消息，实现先进先出（FIFO）的特性。
一、引入依赖
这里引入了两个依赖。一个是rabbitmq的依赖，另一个是配置json转换器所需要的依赖。生产者和消费者服务都需要引入这两个依赖。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>
</dependency>
二、配置rabbitmq的连接信息等
1、生产者配置
rabbitmq:
host: 170.40.20.16
port: 5672
username: zhuoye
password: zy521
virtual-host: /
2、消费者配置
rabbitmq:
host: 170.40.20.16
port: 5672
username: zhuoye
password: zy521
virtual-host: /
listener:
simple:
prefetch: 1 #每次只能处理一个，处理完成才能获取下一个消息
三、设置消息转换器
默认情况下Spring采用的序列化方式是JDK序列化，而JDK的序列化存在可读性性差、占用内存大、存在安全漏洞等问题。所以，这里我们一般使用Jackson的序列化代替JDk的序列化。
在生产者和消费者的启动类上加上如下代码：
@SpringBootApplication
@EnableRabbit //开启rabbitmq的使用
public class ConsumerApp {
public static void main( String[] args ) {
SpringApplication.run(ConsumerApp.class, args);
}
//使用的是Jackson库中的Jackson2JsonMessageConverter类，代替使用jdk自带的序列化
@Bean
public MessageConverter jacksonMessageConvertor(){
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
jackson2JsonMessageConverter.setCreateMessageIds(true);//开启消息id的自动生成功能
return jackson2JsonMessageConverter;
}
}四、生产者代码示例
1、配置交换机和队列信息
@Configuration
public class RabbitMqConfig {
private static String EXCHANGE_NAME="amq.topic";
private static String QUEUE_NAME="alarm.data.topic.queue";
private static String CONFIRM_ALARM_QUEUE_NAME="alarm.confirm.data.topic.queue";
/**
* 声明交换机
*/
@Bean
public TopicExchange exchange(){
// durable:是否持久化,默认是false
// autoDelete:是否自动删除，当没有生产者或者消费者使用此交换机，该交换机会自动删除。
return new TopicExchange(EXCHANGE_NAME,true,false);
}
/**
* 声明告警队列
* @return
*/
@Bean("alarmQueue")
public Queue alarmQueue(){
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
return new Queue(QUEUE_NAME,true,false,false);
}
/**
* 声明确认告警队列
* @return
*/
@Bean("confirmAlarmQueue")
public Queue confirmAlarmQueue(){
return new Queue(CONFIRM_ALARM_QUEUE_NAME,true,false,false);
}
/**
* 声明告警队列绑定关系
* @param queue
* @param topicExchange
* @return
*/
@Bean
public Binding alarmBinding(@Qualifier("alarmQueue") Queue queue, TopicExchange topicExchange){
return BindingBuilder.bind(queue).to(topicExchange).with("server.event.#");
}
/**
* 声明确认告警队列绑定关系
* @param queue
* @param topicExchange
* @return
*/
@Bean
public Binding confirmAlarmBinding(@Qualifier("confirmAlarmQueue") Queue queue, TopicExchange topicExchange){
return BindingBuilder.bind(queue).to(topicExchange).with("server.event_confirm.#");
}2、生产消息代码
@Autowired
private RabbitTemplate rabbitTemplate;
private static String EXCHANGE_NAME="amq.topic";
private static String CONFIRM_ALARM_QUEUE_NAME="alarm.confirm.data.topic.queue";
@Test
void producerAlarmMsg() {
String msg = "发送一条告警消息";
rabbitTemplate.convertAndSend(EXCHANGE_NAME, "server.event.#",msg);
System.out.println("msg = " + msg);
}
@Test
void producerConfirmAlarmMsg() {
String msg = "发送一条确认告警消息";
rabbitTemplate.convertAndSend(CONFIRM_ALARM_QUEUE_NAME, "server.event_confirm.#",msg);
System.out.println("msg = " + msg);
}五、消费者代码示例
1、消费层代码
@Component
public class AlarmConsumer {
@Autowired
private IAlarmService alarmService;
@RabbitListener(queues ="alarm.data.topic.queue",concurrency = "5")
public void getAlarmInfo(String data){
alarmService.dealAlarmData(data);
}
@RabbitListener(queues ="alarm.confirm.data.topic.queue",concurrency = "5")
public void getConfirmAlarmInfo(String data){
alarmService.dealConfirmAlarmData(data);
}
}2、业务层代码
@Service
public class IAlarmServiceImpl implements IAlarmService {
@Override
public void dealAlarmData(String data) {
EquipAlarmResp equipAlarmResp= JSON.parseObject(result,EquipAlarmResp.class);
List<String> alarmIdsOld = dceEquipAlarmMapper.queryAllAlarmIds();
DceEquipAlarmDto dceEquipAlarmDto = CopyBeanUtils.copyProperties(equipAlarmResp, DceEquipAlarmDto.class);
dceEquipAlarmDto.setCreateTime(new Date());
dceEquipAlarmDto.setAlarmTime(dceEquipAlarmDto.getAlarmTime()/1000);
//查询出需要新增或者更新的数据
Boolean flag=alarmIdsOld.stream().filter(a->a.equals(dceEquipAlarmDto.getAlarmId())).findFirst().isPresent();
//开启事务，保证新增、更新、删除的原子性
TransactionStatus transaction = transactionManager.getTransaction(transactionDefinition);
List<DceEquipAlarmDto> list=new ArrayList<>();
list.add(dceEquipAlarmDto);
try {
//新增
if (!flag) {
dceEquipAlarmMapper.insertBatch(list);
}
//更新
if (flag) {
dceEquipAlarmMapper.updateBatch(list);
}
//提交事务
transactionManager.commit(transaction);
} catch (Exception e) {
//回滚
transactionManager.rollback(transaction);
log.error("DynamicEnvironmentServiceImpl.getAlarmInfoByRabbitMq 新华报业动环设备告警信息更新失败！", e);
}
}
@Override
public void dealConfirmAlarmData(String data) {
EquipConfirmAlarmResp alarmResp = JSON.parseObject(data,EquipConfirmAlarmResp.class);
Integer confirmTime = Integer.parseInt(String.valueOf(System.currentTimeMillis() / 1000));
alarmResp.setConfirmTime(confirmTime);
dceEquipAlarmMapper.updateConfirmAlarmBatch(alarmResp,alarmResp.getAlarmIds());
}
}
注：以上代码为对接告警信息和对接告警确认消息的示例。
