SpringBoot amqp MQ RabbitMQ

参考文章

Springboot 整合RabbitMq ，用心看完这一篇就够了==>https://blog.csdn.net/qq_35387940/article/details/100514134

MQ应用场景

应用场景

注册完成时, 发送邮件和短信通知使用MQ.
支付完成时, 回调通知使用MQ
下单时,订单系统使用MQ存入, 库存系统使用MQ取出
流量削峰/抢单时, 前100用户请求存入MQ, 超过100个直接返回"error", 秒杀业务从MQ中取那100个.

RabbitMQ各组件的关系

我的processon图片地址：https://www.processon.com/diagraming/5ce6114fe4b022becb418ee7

virtual host虚拟主机

每个virtual host本质上都是一个RabbitMQ Server，拥有它自己的queue，exchagne，和bings rule等等。这保证了你可以在多个不同的application中使用RabbitMQ。

Exchange交换器

exchange交换器用于将生产者生产的数据根据绑定规则转发到相应的队列中, 一共有4大类型，direct,fanout,topic,headers(不用)

Exchange direct类型

消息中的路由键（routing key)如果和Binding中的binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全一致才匹配。

Exchange Fanout类型

不处理路由键，一次性直接转发到所有绑定的队列上。就像广播一样，转发消息最快。

Exchange Topic类型

通过匹配模式来处理路由键，用于发布订阅模式。

"#"匹配多个词, "*"匹配一个词, 特别注意, 这里的一个词的概念不是一个英文单词, 而是以"."分隔后的词, 比如

词语1.词语2.词语3 , 具体样例为 abc1,def2,ghi3 那么abc1是一个词, def2是一个词, ghi3是一个词.

binding 绑定器

绑定各种转发规则

exchange-binding-queue关系图如下

我的processon地址: https://www.processon.com/diagraming/5ce79afde4b07b4302212db8

再次强调: exchange.topic中的绑定规则为 "#"匹配多个词, "*"匹配一个词, 特别注意, 这里的一个词的概念不是一个英文单词, 而是以"."分隔后的词, 比如

词语1.词语2.词语3 , 具体样例为 abc1,def2,ghi3 那么abc1是一个词, def2是一个词, ghi3是一个词.

queue 队列

队列queue就是等待生产者生产的数据写入, 及消费者把数据取出的一个数据队列.

docker安装rabbit

安装简易说明

web端管理界面

5672是rabbitmq服务端口, 15672是rabbitmq 网页管理端口, 使用默认帐号/密码 guest/guest 访问 http://IP:15672/ 即可.

添加exchanger

添加Queues

添加绑定binding

注意: 下图经过PS,将三图合一

获取之后删除

Queues 界面 Get Message(s) 选择 ack mode 响应模式 , 测试使用一般选第1种, 正常使用一般选第2种.

Nack message requere true: 直译: 不响应消息 , 再从队列中取数,可得 ( 即获取数据后, 再次获取仍可以得到相同值)
Ack message requeue false : 直译: 响应消息, 再从队列中取数, 不可得 (即获取数据后, 再次获取将得到空值)

RabbitMQ基本原理

自动配置
1、RabbitAutoConfiguration
2、有自动配置了连接工厂ConnectionFactory；
3、RabbitProperties 封装了 RabbitMQ的配置
4、 RabbitTemplate ：给RabbitMQ发送和接受消息；
5、 AmqpAdmin ： RabbitMQ系统管理功能组件; AmqpAdmin：创建和删除 Queue，Exchange，Binding
6、@EnableRabbit + @RabbitListener 监听消息队列的内容

SpringBoot中使用RabbitMQ

pom.xml导入RabbitMQ依赖

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

application.properties配置

spring.rabbitmq.addresses=centos
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 以下默认
# spring.rabbitmq.port=5672
# spring.rabbitmq.virtual-host=/

RabbitMQ自动装配类RabbitAutoConfiguration

@Configuration
@ConditionalOnClass({RabbitTemplate.class, Channel.class})
@EnableConfigurationProperties({RabbitProperties.class})
@Import({RabbitAnnotationDrivenConfiguration.class})
public class RabbitAutoConfiguration {
......
    @Configuration
    @Import({RabbitAutoConfiguration.RabbitConnectionFactoryCreator.class})
    protected static class RabbitTemplateConfiguration {
        private final ObjectProvider<MessageConverter> messageConverter;
        private final RabbitProperties properties;
        //RabbitProperties封装了RabbitMQ的配置
        public RabbitTemplateConfiguration(ObjectProvider<MessageConverter> messageConverter, RabbitProperties properties) {
            this.messageConverter = messageConverter;
            this.properties = properties;
        }

        //自动装配RabbitTemplate ,用于给RabbitMQ发送和接收消息
        @Bean
        @ConditionalOnSingleCandidate(ConnectionFactory.class)
        @ConditionalOnMissingBean({RabbitTemplate.class})
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            MessageConverter messageConverter = (MessageConverter)this.messageConverter.getIfUnique();
            //如果需要以json字符串的形式存入,则要自定义一个MessageConverter Bean 来替换默认bean
            if (messageConverter != null) {
                rabbitTemplate.setMessageConverter(messageConverter);
            }

            rabbitTemplate.setMandatory(this.determineMandatoryFlag());
            Template templateProperties = this.properties.getTemplate();
            Retry retryProperties = templateProperties.getRetry();
            if (retryProperties.isEnabled()) {
                rabbitTemplate.setRetryTemplate(this.createRetryTemplate(retryProperties));
            }

            if (templateProperties.getReceiveTimeout() != null) {
                rabbitTemplate.setReceiveTimeout(templateProperties.getReceiveTimeout());
            }

            if (templateProperties.getReplyTimeout() != null) {
                rabbitTemplate.setReplyTimeout(templateProperties.getReplyTimeout());
            }

            return rabbitTemplate;
        }
        
        ......
        //-  AmqpAdmin ： RabbitMQ系统管理功能组件; 用于创建和删除 Queue，Exchange，Binding
        @Bean
        @ConditionalOnSingleCandidate(ConnectionFactory.class)
        @ConditionalOnProperty(
            prefix = "spring.rabbitmq",
            name = {"dynamic"},
            matchIfMissing = true
        )
        @ConditionalOnMissingBean({AmqpAdmin.class})
        public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
            return new RabbitAdmin(connectionFactory);
        }
......
}

RabbitTemplate 和 AmqpAdmin 都是自动装配的Bean,可以直接@AutoWired使用它们

测试用例

package com.rabbitmq;

import com.rabbitmq.bean.Student;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
public class Springboot03RabbitmqApplicationTests {
    Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    AmqpAdmin amqpAdmin;

    @Test//声明交换器 , 并未使用
    public void declareExchange() {
        amqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange"));
        logger.info("declareExchange");
    }

    @Test//声明队列 , 并未使用
    public void declareQueue() {
        amqpAdmin.declareQueue(new Queue("amqpadmin.queue", true));
        logger.info("declareQueue");
    }

    @Test//声明绑定规则 , 并未使用
    public void declareBinding() {
        amqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE, "amqpadmin.exchange", "amqp.student", null));
        logger.info("declareBinding");
    }


    @Test
    public void contextLoads(){
        //rabbitTemplate.send()方法,在使用Message时,需要自己构造一个,用于定义消息体内容和消息头
        //rabbitTemplate.send(exchage,routeKey,message);

        //rabbitTemplate.convertAndSend()方法是send()方法的简化版, 把object默认当成消息体，只需要传入要发送的对象，自动序列化发送给rabbitmq；内部还是用了Message对象, 只是把Message的Header默认了,只对body部分做操作转换
        //rabbitTemplate.convertAndSend(exchage,routeKey,object);
    }


    /**
     * 生产者
     * 默认被序列化发送,因为rabbitTemplate默认使用的messageConverter是序列化后发送的,所以存入rabbitMQ中查看是乱的,
     * 如果需要以json字符串的形式存入,则要自定义一个MessageConverter Bean 来替换默认bean , 本样例参见MyRabbitConfig.java
     * direct单播（点对点）
     */
    @Test
    public void direct() {
        Student stu = new Student(4, "bobo", 18, true, new Date(), "fenfen");
        //对象被默认序列化以后发送出去
        rabbitTemplate.convertAndSend("exchange.direct", "student.bobo", stu);
    }

    /**
     * 生产者
     * fanout广播模式
     */
    @Test
    public void fanout() {
        Student stu = new Student(2, "bobo", 18, true, new Date(), "fenfen");
        //对象被默认序列化以后发送出去
        rabbitTemplate.convertAndSend("exchange.fanout", null, stu);//fanout广播模式,不需要routingKey, 就算指定了也会被无视
    }

    /**
     * 生产者
     * topic 主题模式
     */
    @Test
    public void topic() {
        Student stu = new Student(3, "bobo", 18, true, new Date(), "fenfen");
        //对象被默认序列化以后发送出去,由于student.bobo符合student.#绑定规则,该规则默认分发给student,student.bobo,student.sisi这三个队列
        rabbitTemplate.convertAndSend("exchange.topic", "student.bobo", stu);
    }

    /**
     * 消费者
     * 默认是Ack message requeue false模式(响应消息,再从队列取数时将无法取到)
     */
    @Test
    public void receive() {
        Object o = rabbitTemplate.receiveAndConvert("student.bobo");
        logger.info(o.getClass()+"");
        logger.info(o.toString());
    }

    /**
     *
     */
    public void listener(){
        //参见具体MyRabbitListener
    }

}

在使用Spring RabbitMQ做消息监听时，如果监听程序处理异常了，且未对异常进行捕获，会一直重复接收消息，然后一直抛异常。

RabbitMQ消息监听程序异常时，消费者会向rabbitmq server发送Basic.Reject，表示消息拒绝接受，由于Spring默认requeue-rejected配置为true，消息会重新入队，然后rabbitmq server重新投递，造成了程序一直异常的情况。

所以说了这么多，我们通过rabbitmq监听消息的时候，程序一定要添加try…catch语句!!!当然你也可以根据实际情况，选择设置requeue-rejected为false来丢弃消息。
本小段摘自: RabbitMQ消息监听异常问题探究==>https://blog.csdn.net/u014513883/article/details/77907898

如何转成json存储呢

默认转换器存储的都是序列化对象, 添加自定义转换器

源码分析总结：

1.MessageConverter可以把java对象转换成Message对象，也可以把Message对象转换成java对象

2.MessageListenerAdapter内部通过MessageConverter把Message转换成java对象，然后找到相应的处理方法，参数为转换成的java对象。

3.SimpleMessageConverter处理逻辑：如果content_type是以text开头，则把消息转换成String类型如果content_type的值是application/x-java-serialized-object则把消息序列化为java对象，否则，把消息转换成字节数组。

所以我们需要自定义一个MessageConverter Bean来替换SimpleMessageConverter

package com.rabbitmq.config;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MyRabbitConfig {
    //替换默认转换器,把对象转成json字符串,存取rabbitMQ
    @Bean
    public MessageConverter messageConverter(){
        System.err.println("MyRabbitConfig created");
        return new Jackson2JsonMessageConverter();
    }
}

正常消费者消费的时候需要监听模式

在Bean上添加@RabbitListener如下

package com.rabbitmq.service;

import com.rabbitmq.bean.Student;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class RabbitService {
    Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 直接获取body并把它转成Student类型, queues参数可以指定多个队列,适合读取符合Student规范的json数据,所以也可以写成 Object 手动强转.
     */
    @RabbitListener(queues = {"student","student.bobo"})
    public void receiveStudent(Student stu) {
        logger.info("receiveStudent:" + stu.toString());
    }

    /**
     * 直接获取body并把它转成Student类型, queues参数可以指定多个队列
     */
    @RabbitListener(queues = {"student","student.sisi"})
    public void receiveMessage(Message message) {
        logger.info(message.getMessageProperties().toString());//获取header
        logger.info(message.getBody().toString());//获取body, 返回的是字节数组,适合读取非字符串文件
    }
}

RabbitMQ：@RabbitListener 与 @RabbitHandler 及消息序列化==>https://www.jianshu.com/p/911d987b5f11

延时队列

SpringBoot整合RabbitMQ实现延时队列==>https://www.cnblogs.com/jockming/p/13180669.html

我的项目git地址

https://gitee.com/KingBoBo/springboot-03-rabbitmq

待了解

rabbitmq的备份与还原

【RabbitMQ】一文带你搞定RabbitMQ延迟队列==>https://www.cnblogs.com/mfrank/p/11260355.html

SpringBoot高级篇Redis之ZSet数据结构使用姿势==>https://blog.csdn.net/qq_17312239/article/details/104020031

遇见异常

o.s.a.rabbit.connection.CachingConnectionFactory - Attempting to connect to: [localhost:5672]

明明有配置文件, 却提示连接localhost:5672失败, 摆明了配置文件无效, 仔细核实配置即可

使用异步后的烦恼

烦恼一：数据丢失的风险

解决方式：先写日志或数据库，后放入异步队列.

烦恼二：对其他系统的压力变大

解决方式：使用一定的限流和熔断，对其他系统进行保护。

烦恼三：数据保存后异步任务未执行

解决方式：使用异步任务补偿的方式，定期从数据库中获取数据，放到队列中进行执行，执行后更新数据状态位。

烦恼四：怎样队列长设置和消费者数量

解决方式：使用实际的压力测试来获得队列长度。或者使用排队论的数学公式得到初步的值,然后进行实际压测。

最后介绍一下项目中的经验：

1.量力而行：根据业务特点进行技术选型，业务量小尽量避免使用异步。有所为，有所不为

2.数据说话：异步时一定要进行必要的压力测试

3.先找出系统的关键点：优化单体系统内的性能，再通过整体系统分解来全局优化

4.根据团队和项目的特点选择框架。

一个可供参考的Java高并发异步应用案例==>http://www.uml.org.cn/zjjs/2016060310.asp

参考

RabbitMQ三种Exchange模式(fanout,direct,topic)的性能比较(转)==>https://www.cnblogs.com/shenyixin/p/9084249.html

Springboot Rabbitmq 使用Jackson2JsonMessageConverter 消息传递后转对象==>https://www.cnblogs.com/timseng/p/11688019.html

rabbitMQ实现推迟队列==>https://www.cnblogs.com/zhshlimi/p/10913586.html

rabbitmq重试机制==>https://blog.csdn.net/xixingzhe2/article/details/84345054