SpringBoot整合RabbitMQ的快速使用教程_spring rabbitmq 使用

     

目录

一、引入依赖

二、配置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());
 
    }
 
}

注：以上代码为对接告警信息和对接告警确认消息的示例。