SpringBoot集成RocketMQ和RabbitMQ消息队列（可配置）

简介

在SpringBoot项目中集成消息是一个常见的需求，特别是在需要处理高并发、高性能的消息场景下。RocketMQ是一个分布式消息传递和流计算平台，具有高吞吐量、高可用性、可扩展性等特点。网上比较多的都是集成某一种单一消息队列，下面将介绍如何在SpringBoot项目中通过配置的方式集成RocketMQ和RabbitMQ。

步骤一：添加依赖

首先，你需要在你的SpringBoot项目的pom.xml文件中添加RocketMQ和RabbitMQ的依赖。如果你使用的是Maven，可以添加如下依赖：

		<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

步骤二：配置RocketMQ和RabbitMQ

在application.properties或application.yml文件中配置RocketMQ和RabbitMQ的相关属性。例如：

spring:
  # 开关，指定使用mq rocketmq rabbitmq no-mq(不使用mq)
  mq: rabbitmq
  rabbitmq:
    host: localhost
    username: admin
    password: password
    virtual-host: /
    publisher-confirms: true
    port: 5672
  autoconfigure:
    exclude: org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration

rocketmq:
  name-server: localhost:9876
  producer:
    #生产者组名，规定在一个应用里面必须时唯一的(这里直接引用项目名称,或者自定义唯一组名)。
    group: my-producer-group
    #消息发送的超时时间，毫米级别，默认为3S
    send-message-timeout: 3000
    #消息达到4096字节的时候，消息就会被压缩。默认就是4096，有利于网络传输，提升性能。
    compress-message-body-threshold: 4096
    #最大的消息限制 默认为128K
    max-message-size: 4194304
    #同步消息发送失败重试次数
    retry-times-when-send-failed: 3
    #在内部发送失败时是否重试其他代理。 源码：setRetryAnotherBrokerWhenNotStoreOK，就是指：发送到broker-a失败是否发送到broker-b。这个参数在有多个broker才生效。
    retry-next-server: true
    #异步消息发送失败重试的次数
    retry-times-when-send-async-failed: 3

配置RabbitMQ的Configuration,使用ConditionalOnProperty注解判断是否启用加载

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.annotation.Scope;

@Configuration
@ConditionalOnProperty(name="spring.mq",havingValue = "rabbitmq")
public class MQConfiguration {
    @Bean
    @Primary
    @ConfigurationProperties(prefix = "spring.rabbitmq")
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        //设置序列化规则
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
        //SimpleRabbitListenerContainerFactory发现消息中有content_type有text就会默认将其转换成string类型的
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(1);
        //针对每个监听器，设置最小的并发消费者的数量,一次性从队列取5条消息，开5个线程同时处理任务
        factory.setConcurrentConsumers(1);
        //设置最大的并发的消费者数量
        factory.setMaxConcurrentConsumers(1);
        //设置反序列化规则
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置确认模式手工确认, 开启手动 ack
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //投递失败时是否重新排队
        factory.setDefaultRequeueRejected(false);
        configurer.configure(factory, connectionFactory);
        return factory;
    }
}

import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConditionalOnProperty(name="spring.mq",havingValue = "rabbitmq")
public class MyRabbitAutoConfiguration extends RabbitAutoConfiguration {

}

步骤三：创建消息生产者

在Spring Boot应用中创建消息生产者用于消息发送，这里要注意的是需要创建一个消息发送工具类，消息发送接口类，一个RabbitMQ消息发送实现类，一个RocketMQ消息发送实现类和一个默认消息不发送实现类：
消息发送工具类：

 package com.seeingtv.vas.core.mq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @ClassName: SendMsgHandler
 * @Description: mq消息生产者
 * @author: 
 * @date: 2019年3月18日 下午7:13:14
 */
@Component
public class SendMsgHandler {
    private static final Logger log = LoggerFactory.getLogger(SendMsgHandler.class);
    @Autowired
    private IMqTemplate mqTemplate;
    /**
     * @return boolean
     * @Author 
     * @Description 发送mq消息
     * @Date 15:44 2019/7/19
     * @Param [queueName, datajson]
     **/
    public boolean sendMqMsg(String queueName, String datajson) {
        return mqTemplate.convertAndSend(queueName,datajson);
    }

    /**
     * @return boolean
     * @Author 
     * @Description 发送mq消息
     * @Date 15:44 2019/7/19
     * @Param [queueName, datajson]
     **/
    public boolean sendMqMsg(String queueName, String datajson, int priority) {
        return mqTemplate.convertAndSend(queueName,datajson,priority);
    }
}

消息发送接口类：

public interface IMqTemplate {

    boolean convertAndSend(String queueName, String datajson);

    boolean convertAndSend(String queueName, String datajson, int priority);
}

RabbitMQ消息发送实现类：

import com.alibaba.fastjson.JSONObject;
import com.seeingtv.vas.common.constant.MqConstants;
import com.seeingtv.vas.common.utils.Fastjson2Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

@Component
@ConditionalOnProperty(name="spring.mq",havingValue = "rabbitmq")
public class RabbitMqTemplateImpl implements IMqTemplate{
    private static final Logger log = LoggerFactory.getLogger(RabbitMqTemplateImpl.class);
    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public boolean convertAndSend(String queueName, String datajson) {
        try {
            JSONObject jsonObject = Fastjson2Utils.jsonToObject(datajson,JSONObject.class);
            amqpTemplate.convertAndSend(MqConstants.exchange,queueName+MqConstants.routingKey, jsonObject);
            return true;
        } catch (Exception e) {
            log.error("发送mq消息失败队列名称:{},mq消息:{},优先级:{}", queueName, datajson, e);
        }
        return false;
    }

    @Override
    public boolean convertAndSend(String queueName, String datajson, int priority) {
        try {
            JSONObject jsonObject = Fastjson2Utils.jsonToObject(datajson,JSONObject.class);
            amqpTemplate.convertAndSend(MqConstants.exchange,queueName+MqConstants.routingKey, jsonObject, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setPriority(priority);
                    return message;
                }
            });
            return true;
        } catch (AmqpException e) {
            log.error("发送mq消息失败队列名称:{},mq消息:{},优先级:{}", queueName, datajson, priority, e);
        }
        return false;
    }
}

RocketMQ消息发送实现类：

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

@Component
@ConditionalOnProperty(name = "spring.mq", havingValue = "rocketmq")
public class RocketMqTemplateImpl implements IMqTemplate {
    private static final Logger log = LoggerFactory.getLogger(RocketMqTemplateImpl.class);
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public boolean convertAndSend(String queueName, String dataJson) {
        boolean flag = false;
        try {
            SendResult sendResult = rocketMQTemplate.syncSend(queueName, dataJson);
            flag = sendResult.getSendStatus().equals(SendStatus.SEND_OK);
        } catch (Exception e) {
            log.error("mq发送消息失败", e);
        }

        return flag;
    }

    @Override
    public boolean convertAndSend(String queueName, String dataJson, int priority) {
        return convertAndSend(queueName, dataJson);
    }
}

默认消息不发送实现类：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;


@Component
@ConditionalOnProperty(name="spring.mq",havingValue = "no-mq", matchIfMissing = true)
public class NoMqTemplateImpl implements IMqTemplate{
    private static final Logger log = LoggerFactory.getLogger(NoMqTemplateImpl.class);

    @Override
    public boolean convertAndSend(String queueName, String datajson) {
        return false;
    }

    /**
     * @description:
     * @author:  
     * @date: 2022/10/28 14:18
     * @param: [queueName, datajson, priority]
     * @return: boolean
     **/
    @Override
    public boolean convertAndSend(String queueName, String datajson, int priority) {
        return false;
    }
}

至此已经可以通过配置文件中spring.mq来配置使用哪个mq来发送消息了

步骤四：创建消费者消费mq

RocketMQ消费者实现：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@ConditionalOnProperty(name="spring.mq",havingValue = "rocketmq")
@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "my-consumer-group")
public class MessageConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

RabbitMQ消费者实现：

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import java.io.IOException;

@ConditionalOnProperty(name = "spring.mq", havingValue = "rabbitmq")
@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue("your-topic"),
        exchange = @Exchange("your-exchange"),
        key = ("your-topic" + ".routingKey")))
public class MessageConsumer {
    private static final Logger log = LoggerFactory.getLogger(MessageConsumer.class);
    @RabbitHandler
    public void onMessage(@Payload Object o, Message msg, Channel channel) {
        try {
            System.out.println("Received message: " + o);
        } catch (Exception e) {
            log.error("MessageConsumer error！", e);
        } finally {
            try {
                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                log.error("MessageConsumer error", e);
            }
        }
    }
}

小结

实现的技术细节就是springboot的一个条件注解ConditionalOnProperty的使用。

ConditionalOnProperty注解‌是Spring Boot中的一个条件注解，主要用于根据配置文件中的属性值来决定是否创建或注入某个Bean。这个注解通过其属性name和havingValue来控制条件是否生效。‌

ConditionalOnProperty注解常用于以下场景：
‌条件创建Bean‌：在某些情况下，可能需要根据配置文件中的属性值来决定是否创建某个Bean。例如，如果配置文件中没有指定某个属性，则不创建相应的Bean。

‌条件注入Bean‌：在有多个实现类的情况下，可以通过ConditionalOnProperty注解配合配置文件来决定注入哪个实现类。例如，如果配置文件中指定了某个实现类，则注入该实现类；否则，注入另一个实现类。

‌基本用法‌：在@Configuration注解的类上使用@ConditionalOnProperty注解，可以控制该配置类是否生效。如果配置文件中的属性值与注解中的havingValue值相等，则该配置类生效；否则，不生效。

‌属性说明‌：
name：指定要检查的配置项的名称。
havingValue：指定配置项的期望值。如果配置项的值与havingValue相等，则条件成立；否则，条件不成立。
matchIfMissing：如果为true，即使配置项不存在，条件也会成立；否则，条件不成立。

以上案列只是使用RocketMQ和RabbitMQ作为消息队列配置化使用案例，小伙伴们可以根据自己需求引入kafka等其他消息队列

技术不易 蹒跚学步 众志成城 发扬光大

# java-rabbitmq # java-rocketmq # spring boot
Logo
乐奇 Rokid 开放社区

更多推荐

cover

TestStar | AIUI 数据库客户端测试Case执行实录

avatar 乐奇 Rokid 开放社区
cover

AR巡检哪家好

avatar 乐奇 Rokid 开放社区
cover

具身智能 | Ego数采设备与选择标准一文理清，避开无效采集

avatar 乐奇 Rokid 开放社区