消息中间件—SpringBoot下RabbitMQ实战

消息中间件简介

MQ全称（Message Queue）`又名消息队列，是一种异步通讯的中间件。可以将它理解成邮局，发送者将消息传递到邮局，然后由邮局帮我们发送给具体的消息接收者（消费者），具体发送过程与时间我们无需关心，它也不会干扰我进行其它事情。常见的MQ有kafka、activemq、rocketMQ、rabbitmq等等**

消息中间件的应用场景

跨系统数据传递、高并发流量削峰、数据异步处理。。。。

 

消息中间件对比

综上，各种对比之后，有如下建议：

一般的业务系统要引入 MQ，最早大家都用 ActiveMQ，但是现在确实大家用的不多了，没经过大规模吞吐量场景的验证，社区也不是很活跃，所以大家还是算了吧，我个人不推荐用这个了；

后来大家开始用 RabbitMQ，但是确实 erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它，对公司而言，几乎处于不可控的状态，但是确实人家是开源的，比较稳定的支持，活跃度也高；

不过现在确实越来越多的公司会去用 RocketMQ，确实很不错，毕竟是阿里出品，但社区可能有突然黄掉的风险（目前 RocketMQ 已捐给 Apache，但 GitHub 上的活跃度其实不算高）对自己公司技术实力有绝对自信的，推荐用 RocketMQ，否则回去老老实实用 RabbitMQ 吧，人家有活跃的开源社区，绝对不会黄。

所以中小型公司，技术实力较为一般，技术挑战不是特别高，用 RabbitMQ 是不错的选择；大型公司，基础架构研发实力较强，用 RocketMQ 是很好的选择。

如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。

 

安装使用

rabbitmq的windows版本安装，可以参考这个地址： https://blog.csdn.net/hzw19920329/article/details/53156015

 

Rabbitmq基础概念

Broker：简单来说就是消息队列服务器实体 Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列 Queue：消息队列载体，每个消息都会被投入到一个或多个队列 Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来 Routing Key：路由关键字，exchange根据这个关键字进行消息投递 vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离 producer：消息生产者，就是投递消息的程序 consumer：消息消费者，就是接受消息的程序 channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

 

rabbitmq运作流程

上图所示，在整个消息中间件的使用过程中，我们主要配置的是exchange,queue,rooting key三个，下面的代码主要也是根据着三个进行消息的生产和消费。

常用的交换机类型有：fanout、direct、topic、headers，他们特性如下：

fanout：绑定的都发送，忽略路由，不用传routekey

direct：全文匹配--常用

topic：模糊匹配--常用

headers:条件判断，性能很差，不建议使用，我们代码的demo，使用的是direct和topic,后续通过插件的形式，我们还添加了x-delayed-message这个延迟消息队列。

 

springboot代码实现：

步骤1：pom文件中引入相应的组件

<!--rabbitmq -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

步骤2：部署好mq之后，在application.properties文件中添加rabbitmq相应的配置

spring.rabbitmq.host=129.204.x.xxx
spring.rabbitmq.port=5672
spring.rabbitmq.username=xxx
spring.rabbitmq.password=xxx
spring.rabbitmq.virtual-host=xx

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.listener.simple.acknowledge-mode=manual

步骤3：进行交换机，队列，routingkey的定义和绑定，具体的代码在RabbitConfig文件中。

@Configuration
public class RabbitConfig {
​
    //绑定键
    public final static String man = "kms.topic.man";
    public final static String woman = "kms.topic.woman";
​
    /**
     * 队列 起名：TestDirectQueue
     * @return
     */
    @Bean
    public Queue DirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("kms.direct.queue",true);
    }
​
    /**
     * Direct交换机 起名：TestDirectExchange
     * @return
     */
    @Bean
    DirectExchange DirectExchange() {
        return new DirectExchange("kms.direct.exchange",true,false);
    }
​
    /**
     *  绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
     * @return
     */
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(DirectQueue()).to(DirectExchange()).with("kms.direct.routingKey");
    }
​
    @Bean
    public Queue firstTopicQueue() {
        return new Queue(RabbitConfig.man);
    }
​
    @Bean
    public Queue secondTopicQueue() {
        return new Queue(RabbitConfig.woman);
    }
​
    /**
     * 主题型交换机1
     * @return
     */
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange("kms.topic.exchange");
    }
​
    /**
     *  将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
     *  这样只要是消息携带的路由键是topic.man,才会分发到该队列
     * @return
     */
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstTopicQueue()).to(topicExchange()).with(man);
    }
​
    /**
     * 将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
     * 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
     * @return
     */
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondTopicQueue()).to(topicExchange()).with("kms.topic.#");
    }
​
    /**
     * TODO:RabbitMQ延迟队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return QueueBuilder.durable("kms.delay.queue").build();
    }
​
    @Bean
    public CustomExchange delayExchange(){
        Map<String,Object> map= Maps.newHashMap();
        map.put("x-delayed-type","direct");
        return new CustomExchange("kms.delay.exchange","x-delayed-message",true,false,map);
    }
​
    @Bean
    public Binding delayBinding(){
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("kms.delay.routingKey").noargs();
    }
​
​
}

步骤4：生产者创建消息的发布，代码在MqProviderService中,其中留意setup()方法，该方法的作用是为了判断消息是否成功发送到中间件， 消息发送完毕后，则回调此方法 ack代表发送是否成功。

/**
 * 发送数据到mq
 * @Author: yechongbai
 * @Date: 2020/5/11 16:04
 * @Copyright: www.zektech.cn
 * @since 1.0
 */
@Service
public class MqProviderService extends GlobalResponseHandler {
    private final Logger logger = LoggerFactory.getLogger(MqProviderService.class);
​
    @Autowired
    private RabbitTemplate rabbitTemplate;
​
    @PostConstruct
    public void setup() {
        // 消息发送完毕后，则回调此方法 ack代表发送是否成功
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("进入回调。。。");
                // ack为true，代表MQ已经准确收到消息
                if (!ack) {
                    System.out.println("发送消息到MQ失败");
                    System.out.println("ConfirmCallback: "+"原因："+cause);
                    return;
                }
                try {
                    System.out.println(correlationData.getId()+":成功发送消息到MQ");
                    // 修改本地消息表的状态为“已发送”。
                    //TODO:修改本地表
                } catch (Exception e) {
                    logger.error("警告：修改本地消息表的状态时出现异常", e);
                }
​
            }
        });
    }
​
    /**
     * 发送测试信息--直连交换机
     * @return
     */
    public String sendDirectMessage(){
        String messageId = String.valueOf(UUID.randomUUID());
        UserInfoVO userInfoVO=new UserInfoVO();
        userInfoVO.setPersonNo(messageId);
        userInfoVO.setPersonName("张三");
        userInfoVO.setPhoto("111");
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
        //步骤1.添加订单记录
        //前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
        // 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
        // CorrelationData 当收到消息回执时，会附带上这个参数
        rabbitTemplate.convertAndSend("kms.direct.exchange", "kms.direct.routingKey", userInfoVO,new CorrelationData(messageId));
        return buildSuccessResult();
    }
​
    /**
     * 发送测试信息--主题交换机1
     * @return
     */
    public String sendTopicMessage(){
        String messageId = String.valueOf(UUID.randomUUID());
        UserInfoVO userInfoVO=new UserInfoVO();
        userInfoVO.setPersonNo(messageId);
        userInfoVO.setPersonName("主题交换机1");
        userInfoVO.setPhoto("111");
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
        //步骤1.添加订单记录
        //前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
        // 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
        // CorrelationData 当收到消息回执时，会附带上这个参数
        rabbitTemplate.convertAndSend("kms.topic.exchange", "kms.topic.man", JSON.toJSONString(userInfoVO),new CorrelationData(messageId));
        return buildSuccessResult();
    }
​
    /**
     * 发送测试信息--主题交换机2
     * @return
     */
    public String sendTopicMessage2(){
        String messageId = String.valueOf(UUID.randomUUID());
        UserInfoVO userInfoVO=new UserInfoVO();
        userInfoVO.setPersonNo(messageId);
        userInfoVO.setPersonName("主题交换机2");
        userInfoVO.setPhoto("222");
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        System.out.println("推送消息："+ JSON.toJSONString(userInfoVO));
        //步骤1.添加订单记录
        //前提:消息不能丢失情况需要插入，如果类似微信推送这些不一定需要送到的可以不执行步骤2
        // 步骤2.插入推送消息到数据库 步骤1和步骤2需要事务，才能保证后续该消息是否被消费
        // CorrelationData 当收到消息回执时，会附带上这个参数
        rabbitTemplate.convertAndSend("kms.topic.exchange", "kms.topic.woman", JSON.toJSONString(userInfoVO),new CorrelationData(messageId));
        return buildSuccessResult();
    }
​
    /**
     * 发送测试信息--延迟队列
     * @param ttl
     * @return
     */
    public String sendDelayMessage(Long ttl){
        String messageId = String.valueOf(UUID.randomUUID());
        UserInfoVO userInfoVO=new UserInfoVO();
        userInfoVO.setPersonNo(messageId);
        userInfoVO.setPersonName("张三");
        userInfoVO.setPhoto("111");
        System.out.println(LocalDateTime.now()+":推送消息："+ JSON.toJSONString(userInfoVO));
        rabbitTemplate.convertAndSend("kms.delay.exchange", "kms.delay.routingKey", userInfoVO, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties mp=message.getMessageProperties();
                mp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                mp.setHeader("x-delay",ttl);
                System.out.println("延迟队列生产者-发出消息："+JSON.toJSONString(userInfoVO)+",TTL："+ttl);
                return message;
            }
        }, new CorrelationData(messageId));
        return buildSuccessResult();
    }
}

步骤5：消息接收者---消费消息，新建MqConsumerService，进行消息的监听并且消费。 默认情况下spring-boot-data-amqp是自动ACK机制，就意味着 MQ 会在消息消费完毕后自动帮我们去ACK，这样依赖就存在这样一个问题：如果报错了，消息不会丢失，会无限循环消费，很容易就吧磁盘空间耗完，虽然可以配置消费的次数但这种做法也有失优雅。目前比较推荐的就是我们手动ACK然后将消费错误的消息转移到其它的消息队列中，做补偿处理。由于我们需要手动控制ACK，因此下面监听完消息后需要调用basicAck通知rabbitmq消息已被正确消费，可以将远程队列中的消息删除 ，所以需要手动执行：channel.basicAck(tag,true);

@Component
public class MqConsumerService {
​
    /**
     * 监听的队列名称 TestDirectQueue
     * @param testMessage
     * @param channel
     * @param tag
     * @throws IOException
     */
    @RabbitListener(queues = "kms.direct.queue")
    public void processDirect(UserInfoVO testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try{
            System.out.println("DirectReceiver消费者收到消息  : "+ JSON.toJSONString(testMessage));
            //TODO：接收到消息进行业务操作，操作成功，告诉mq该消息已经消费
            //执行业务操作，同一个数据不能处理两次，根据业务情况去重，保证幂等性。 （拓展：redis记录处理情况）
            // 开启了手工确认机制，如果不加这个，项目重新启动，则改消息会被重新消费
            // 异常的话，可以选择让它重新入列，或者丢弃
            channel.basicAck(tag,true);
        }catch (Exception e){
            // 异常情况 :根据需要去： 重发/ 丢弃
            // 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
            channel.basicNack(tag, false, false);
            // 系统 关键数据，永远是有人工干预
        }
    }
​
    /**
     * 监听的队列名称 TestDirectQueue
     * @param testMessage
     * @param channel
     * @param tag
     * @throws IOException
     */
    @RabbitListener(queues = "kms.topic.man")
    public void processTopic(String testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try{
            JSONObject orderInfo = JSONObject.parseObject(testMessage);
            System.out.println("TopicReceiver消费者收到消息  : "+ testMessage);
            System.out.println("收到的personNO:"+orderInfo.getString("personNo"));
            //TODO：接收到消息进行业务操作
            channel.basicAck(tag,true);
        }catch (Exception e){
            // 异常情况 :根据需要去： 重发/ 丢弃
            // 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
            channel.basicNack(tag, false, false);
            // 系统 关键数据，永远是有人工干预
        }
    }
​
    /**
     * 监听的队列名称 TestDirectQueue
     * @param testMessage
     * @param channel
     * @param tag
     * @throws IOException
     */
    @RabbitListener(queues = "kms.topic.woman")
    public void processTopic2(String testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try{
            JSONObject orderInfo = JSONObject.parseObject(testMessage);
            System.out.println("TopicWomenReceiver消费者收到消息  : "+ testMessage);
            System.out.println("收到的personNO:"+orderInfo.getString("personNo"));
            channel.basicAck(tag,true);
        }catch (Exception e){
            // 异常情况 :根据需要去： 重发/ 丢弃
            // 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
            channel.basicNack(tag, false, false);
            // 系统 关键数据，永远是有人工干预
        }
    }
​
    /**
     * 监听的队列名称 DelayQueue
     * @param testMessage
     * @param channel
     * @param tag
     * @throws IOException
     */
    @RabbitListener(queues = "kms.delay.queue")
    public void processDelay(UserInfoVO testMessage,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try{
            System.out.println(LocalDateTime.now()+":DelayQueue消费者收到消息  : "+ JSON.toJSONString(testMessage));
            channel.basicAck(tag,true);
        }catch (Exception e){
            // 异常情况 :根据需要去： 重发/ 丢弃
            // 重发一定次数后， 丢弃， 日志告警,b1:true表示重新入列，false表示不入列
            channel.basicNack(tag, false, false);
            // 系统 关键数据，永远是有人工干预
        }
    }
​
}

步骤6：进行测试，新增controller，调用消息生产的方法，进行消息消费的查看。

@RestController
@RequestMapping("/remote/mq")
public class TestMqController {
    @Autowired
    private MqProviderService mqProviderService;
​
    @PostMapping("sendDirectMessage")
    public String sendDirectMessage(){
        mqProviderService.sendDirectMessage();
        return "ok";
    }
​
    @PostMapping("sendTopicMessage")
    public String sendTopicMessage(){
        mqProviderService.sendTopicMessage();
        return "ok";
    }
​
    @PostMapping("sendTopicMessage2")
    public String sendTopicMessage2(){
        mqProviderService.sendTopicMessage2();
        return "ok";
    }
​
    @PostMapping("sendDelayMessage")
    public String sendDelayMessage(){
        mqProviderService.sendDelayMessage(10*1000L);
        return "ok";
    }
}

执行sendDelayMessage输出结果如下：

2020-05-15T15:16:29.437:推送消息：{"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"}
延迟队列生产者-发出消息：{"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"},TTL：10000
进入回调。。。
5cd80fa8-42c0-4b70-932f-422ca4167a5a:成功发送消息到MQ
2020-05-15T15:16:39.933:DelayQueue消费者收到消息  : {"personName":"张三","personNo":"5cd80fa8-42c0-4b70-932f-422ca4167a5a","photo":"111"}