Springboot使用RabbitMQ 发送消息,监听接收消息

Posted lovoo

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Springboot使用RabbitMQ 发送消息,监听接收消息相关的知识,希望对你有一定的参考价值。

前言

在并发量大的商城项目中,如果有一百万的用户同时下单,那么我们在保证数据的一致性的同时,又如何保证数据库服务器不崩溃?
目前最好的方法是采用rabbitMQ进行处理,把所有的订单先暂存在rabbitMQ的消息队列中,然后监听rabbitMQ,获取对应的消息,然后根据数据库的处理能力将消息进行分批次处理,就可完美解决这些并发问题。

一、概述

  1. 大多应用中,可通过消息服务中间件来提升系统异步通信、扩展解耦能力
  2. 消息服务中两个重要概念: 消息代理(message broker)和目的地(destination) 当消息发送者发送消息以后,将由消息代理接管,消息代理保证消息传递到指定目的地。
  3. 消息队列主要有两种形式的目的地
    1. 队列(queue):点对点消息通信(point-to-point)
    2. 主题(topic):发布(publish)/订阅(subscribe)消息通信
  4. 点对点式:
    . • 消息发送者发送消息,消息代理将其放入一个队列中,消息接收者从队列中获 取消息内容,消息读取后被移出队列
    . • 消息只有唯一的发送者和接受者,但并不是说只能有一个接收者
  5. 发布订阅式:
    • 发送者(发布者)发送消息到主题,多个接收者(订阅者)监听(订阅)这个 主题,那么就会在消息到达时同时收到消息
  6. JMS(Java Message Service)JAVA消息服务:
    • 基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
  7. AMQP(Advanced Message Queuing Protocol)
    • 高级消息队列协议,也是一个消息代理的规范,兼容JMS
    • RabbitMQ是AMQP的实现

二、JMS和AMQP区别与联第

在这里插入图片描述

三、Docker安装RabbitMQ

docker pull rabbitmq:management
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management 
4369, 25672 (Erlang发现&集群端口) 
5672, 5671 (AMQP端口) 
15672 (web管理后台端口) 
61613, 61614 (STOMP协议端口) 
1883, 8883 (MQTT协议端口) 

官方网址:https://www.rabbitmq.com/networking.html

四、RabbitMQ运行机制

AMQP 中的消息路由

AMQP 中消息的路由过程和 Java 开 发者熟悉的 JMS 存在一些差别, AMQP 中增加了 Exchange 和 Binding 的角色。生产者把消息发布 到 Exchange 上,消息最终到达队列 并被消费者接收,而 Binding 决定交 换器的消息应该发送到那个队列。
在这里插入图片描述
Exchange 类型
• Exchange分发消息时根据类型的不同分发策略有区别,目前共四种类型:direct、 fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键, headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了,所以直接 看另外三种类型:
在这里插入图片描述
消息中的路由键(routing key)如果和 Binding 中的 binding key 一致, 交换器 就将消息发到对应的队列中。路由键与队 列名完全匹配,如果一个队列绑定到交换 机要求路由键为“dog”,则只转发 routing key 标记为“dog”的消息,不会转发 “dog.puppy”,也不会转发“dog.guard” 等等。它是完全匹配、单播的模式。
在这里插入图片描述
每个发到 fanout 类型交换器的消息都 会分到所有绑定的队列上去。fanout 交 换器不处理路由键,只是简单的将队列 绑定到交换器上,每个发送到交换器的 消息都会被转发到与该交换器绑定的所 有队列上。很像子网广播,每台子网内 的主机都获得了一份复制的消息。 fanout 类型转发消息是最快的。
在这里插入图片描述
topic 交换器通过模式匹配分配消息的 路由键属性,将路由键和某个模式进行 匹配,此时队列需要绑定到一个模式上。 它将路由键和绑定键的字符串切分成单 词,这些单词之间用点隔开。它同样也 会识别两个通配符:符号“#”和符号 “*”。#匹配0个或多个单词,*匹配一 个单词。

五、RabbitMQ消息确认机制-可靠抵达

保证消息不丢失,可靠抵达,可以使用事务消息,性能下降250倍,为此引入确认 机制
• publisher confirmCallback 确认模式
• publisher returnCallback 未投递到 queue 退回模式
• consumer ack机制
在这里插入图片描述
可靠抵达-ConfirmCallback

spring.rabbitmq.publisher-confirms=true

可靠抵达-ReturnCallback

•spring.rabbitmq.publisher-returns=true 
 spring.rabbitmq.template.mandatory=true

可靠抵达-Ack消息确认机制
消费者获取到消息,成功处理,可以回复Ack给Broker
•basic.ack用于肯定确认;broker将移除此消息
•basic.nack用于否定确认;可以指定broker是否丢弃此消息,可以批量
•basic.reject用于否定确认;同上,但不能批量 •默认自动ack,消息被消费者收到,就会从broker的queue中移除
•queue无消费者,消息依然会被存储,直到消费者消费
•消费者收到消息,默认会自动ack。但是如果无法确定此消息是否被处理完成, 或者成功处理。我们可以开启手动ack模式
•消息处理成功,ack(),接受下一个消息,此消息broker就会移除
•消息处理失败,nack()/reject(),重新发送给其他人进行处理,或者容错处理后ack
•消息一直没有调用ack/nack方法,broker认为此消息正在被处理,不会投递给别人,此时客户 端断开,消息不会被broker移除,会投递给别人

六、RabbitMQ延时队列(实现定时任务)

场景: 比如未付款订单,超过一定时间后,系统自动取消订单并释放占有物品。
常用解决方案: spring的 schedule 定时任务轮询数据库
缺点: 消耗系统内存、增加了数据库的压力、存在较大的时间误差
解决:rabbitmq的消息TTL和死信Exchange结合

消息的TTL(Time To Live)
消息的TTL就是消息的存活时间。
• RabbitMQ可以对队列和消息分别设置TTL。
• 对队列设置就是队列没有消费者连着的保留时间,也可以对每一个单独的消息做单独的 设置。超过了这个时间,我们认为这个消息就死了,称之为死信。
• 如果队列设置了,消息也设置了,那么会取小的。所以一个消息如果被路由到不同的队 列中,这个消息死亡的时间有可能不一样(不同的队列设置)。这里单讲单个消息的 TTL,因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x- message-ttl属性来设置时间,两者是一样的效果

Dead Letter Exchanges(DLX)
一个消息在满足如下条件下,会进死信路由,记住这里是路由而不是队列, 一个路由可以对应很多队列。(什么是死信)

Dead Letter Exchange其实就是一种普通的exchange,和创建其他 exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有 消息过期了,会自动触发消息的转发,发送到Dead Letter Exchange中去。

我们既可以控制消息在一段时间后变成死信,又可以控制变成死信的消息 被路由到某一个指定的交换机,结合二者,其实就可以实现一个延时队列

手动ack&异常消息统一放在一个队列处理建议的两种方式
• catch异常后,手动发送到指定队列,然后使用channel给rabbitmq确认消息已消费
• 给Queue绑定死信队列,使用nack(requque为false)确认消息消费失败

如图:
在这里插入图片描述

七、Springboot集成RabbitMQ

  1. 引入 spring-boot-starter-amqp
  2. application.yml配置
  3. 测试RabbitMQ
    1. AmqpAdmin:管理组件
    2. RabbitTemplate:消息发送处理组件
    3. @RabbitListener 监听消息的方法可以有三种参数(不分数量,顺序) • Object content, Message message, Channel channe

在application.yml配置

    rabbitmq:
    host: 192.168.16.149
    port: 5672
    virtual-host: /
    #开启发送端确认
    publisher-confirm-type: correlated
    #抵达队列确认
    publisher-returns: true
    template:
      #只要抵达队列,异步回调这个return confirm
      mandatory: true
    listener:
      simple:
        #默认是自动,当服务器宕机时,容易丢失消息,所以一般手动确认收到消息,保证数据不会丢失
        acknowledge-mode: manual

在MallOrderApplication.java中添加注解

@EnableRabbit
@EnableDiscoveryClient
@MapperScan("com.wys.mall.order.dao")
@SpringBootApplication
public class MallOrderApplication {

    public static void main(String[] args) {
        SpringApplication.run(MallOrderApplication.class, args);
    }

}

测试:

package com.wys.mall.order;

import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
class MallOrderApplicationTests {

    @Autowired
    private AmqpAdmin amqpAdmin;

    /**
     * 创建交换机
     * DirectExchange 点对点传递消息
     * FanoutExchange 一对多发送消息,发布-订阅模式
     * TopicExchange 一对多发送消息,但要分类
     */
    @Test
    void creatChange() {
        DirectExchange directExchange = new DirectExchange("exchange.first.direct", true,false);
        amqpAdmin.declareExchange(directExchange);

        FanoutExchange fanoutExchange = new FanoutExchange("exchange.first.fanout", true, false);
        amqpAdmin.declareExchange(fanoutExchange);

        TopicExchange topicExchange = new TopicExchange("exchange.first.topic", true, false);
        amqpAdmin.declareExchange(topicExchange);
    }

    /**
     * 创建队列
     */
    @Test
    void creatQueue() {
        Queue queue = new Queue("queue.first", true, false, false);
        amqpAdmin.declareQueue(queue);
    }

    @Test
    void binding() {
        Binding binding = new Binding("queue.first", Binding.DestinationType.QUEUE, "exchange.first.direct", "exchange.first.direct", null);
        amqpAdmin.declareBinding(binding);
    }
}
---------------------------------------------------------------------------------------
      /**
     * 发送消息  根据i值的奇偶,发送不同消息
     */
    @Test
    void sendMsg() {

        for (int i = 0; i < 10 ; i++) {
            if(i % 2== 0){
                OrderEntity orderEntity = new OrderEntity();
                orderEntity.setBillContent("aaaaaaaa--" + i);
                orderEntity.setAutoConfirmDay(i);
                orderEntity.setBillHeader("kkkk--" + i);
                rabbitTemplate.convertAndSend("exchange.first.direct", "queue.first", orderEntity);
            }else {
                OrderItemEntity itemEntity = new OrderItemEntity();
                itemEntity.setOrderSn(UuidUtils.generateUuid().toString());
                rabbitTemplate.convertAndSend("exchange.first.direct", "queue.first", itemEntity);
            }

        }
    }


接收消息

@RabbitListener(queues = {"queue.first"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {


    /**
     *
     * 同一个消息,只能有一个人接收
     * 消息按顺序处理,上一个处理完,下一个才开始
     * @Test
     *     void sendMsg() {
     *
     *         for (int i = 0; i < 10 ; i++) {
     *             OrderEntity orderEntity = new OrderEntity();
     *             orderEntity.setBillContent("aaaaaaaa--" + i);
     *             orderEntity.setAutoConfirmDay(i);
     *             orderEntity.setBillHeader("kkkk--" + i);
     *             rabbitTemplate.convertAndSend("exchange.first.direct", "queue.first", orderEntity);
     *         }
     *     }
     * @param message 封装的消息
     * @param orderEntity 发送消息时的类
     * @param channel 通信渠道
     */
//    @RabbitListener(queues = {"queue.first"})
    @RabbitHandler
    public void recieveOrderMessage(Message message, OrderEntity orderEntity, Channel channel) throws IOException {
        System.out.println("收到了消息了--->" + message + " ====》内容:" + orderEntity);
        System.out.println("渠道数量:" + channel.getChannelNumber());

        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("消息处理完成---------------------------------");
        //消息顺序,自增
        long deliveryTag = messageProperties.getDeliveryTag();
        System.out.println(deliveryTag);
        //回复,签收消息, fasle表示只签收当前消息,true签收所有
        channel.basicAck(deliveryTag, false);
    }
    
    @RabbitHandler
    public void recieveOrderItemMessage(Message message, OrderItemEntity orderItemEntity, Channel channel){
        System.out.println("收到了消息了--->" + message + " ====》内容:" + orderItemEntity);
        System.out.println("渠道数量:" + channel.getChannelNumber());
    }

}

基本配置:

  • 将保存为序列化的内容转化为json格式
  • 发送到broke时回调,确认消息是否到达
  • 发送到queue时回调,只有消息没有到达队列才回调
package com.wys.mall.order.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.codec.cbor.Jackson2CborDecoder;

import javax.annotation.PostConstruct;

/**
 * 消息队列配置
 * @Package com.wys.mall.order.config
 * @Author charlin
 * @Version: v1.0
 * @Date 2021-07-1010:03
 * @Copyright: 2021 www.wyscha.com Inc. All rights reserved.
 */
@Configuration
public class RabbitConfig {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 将保存为序列化的内容转化为json格式
     * @return
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 定制rabbitTemplate
     * RabbitConfig在创建完成后调用此方法
     */
    @PostConstruct
    public void initRabbitTemplate() {
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
            	发送到broke时回调,确认消息是否到达
             * @param correlationData 消息唯一关联数据
             * @param ack 是否成功还是失败
             * @param cause 失败原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("correlationData==" + correlationData);
                System.out.println("ack==" + ack);
                System.out.println("cause==" + cause);
            }
        });

        /**
         * 只有消息没有到达队列才回调
         */
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * @param message 失败消息
             * @param replyCode 失败码
             * <

以上是关于Springboot使用RabbitMQ 发送消息,监听接收消息的主要内容,如果未能解决你的问题,请参考以下文章

SpringBoot整合RabbitMQ之发送接收消息实战

springboot-rabbitmq的使用

SpringBoot Rabbitmq发送消息

SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理

SpringBoot整合RabbitMQ,实现消息发送和消费

SpringBoot实战系列RabbitMQ实现消息发送并实现邮箱发送异常监控报警实战