0


springboot 整合 rabbitMQ(1)

一、MQ概述

  MQ

全称是

Message Queue(消息队列)

,是在消息传输过程中保存消息的容器,多用于分布式 系统之间进行通信。

二、MQ的优势和劣势

** 优势**:

  • 应用解耦 : 系统的耦合性越高,容错性就越低,可维护性就越低,使用MQ 使得应用间解耦, 提升容错性和可维护性。
  • 异步提速 : 提升用户体验和系统吞吐量(单位时间内处理请求的数目)。
  • 削锋填谷 : 使用了MQ之后,限制消费消息的速度为2000,这样一来,高峰期产生的数据势必 会被积压在 MQ中,高峰 就被“”掉了。但是因为消息积压,在高峰期过后的一段时间内,消费消息的速度还是会维持在1000,直到消费完积压的消息,这就叫做 “填谷”。使用MQ后,可以提高系统稳定性。

** 劣势**:

  • 系统的可用性降低 : 引入的外部依赖越多,系统稳定性越差,一旦MQ宕机,会对业务系统造 成影响。如何保证MQ的高可用?
  • 系统复杂度提高 : MQ的加入增加了系统的复杂度,以前是系统间同步的远程调用,现在是 通过MQ进行一部调用,怎么处理消息丢失情况?如何保证消息传递的顺序性。
  • 一致性问题 : A系统发给BCD系统,若其中一个系统处理数据失败,如何保证消息数据处理的 一致性问题。

** 三、常见的

MQ

产品**

RabbitMQ使用步骤

第一步:确保rabbitmq启动并且可以访问15672

第二步:导入依赖

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

第三步:配置

rabbitmq:
  host: ip
  port: 5672
  username: guest
  password: guest
  virtual-host: /
  listener:
    simple:
      prefetch: 1 # 默认每次取出一条消息消费, 消费完成取下一条
      acknowledge-mode: manual # 设置消费端手动ack确认  不写默认为自动确认(auto)
      retry:
        enabled: true # 是否支持重试
  publisher-confirm-type: correlated  #确认消息已发送到交换机(Exchange)
  publisher-returns: true  #确认消息已发送到队列(Queue)     

auto自动确认

  1. 消息成功被消费,没有抛出异常,则自动确认,回复ack。
  2. 当抛出ImmediateAcknowledgeAmqpException异常,则视为成功消费,确认该消息。
  3. 当抛出AmqpRejectAndDontRequeueException异常的时候,则消息会被拒绝丢弃,并且不会重新入队。
  4. 其他的异常,则消息会被拒绝,且requeue = true(默认就是true)时会重新入队

manual手工确认(推荐使用!可以防止消息丢失)

无论有没有异常,标准只有是否主动调用

basicAck()、basicNack()

等方法,没有调用则一直阻塞.

因此有异常时,(有重试就重试),抛出异常后,没有调用时,还是会一直阻塞。即使是

auto

模式的那两个特殊的异常,在

manual

中都是一样的,不会有特殊处理.

官网对应模式介绍:RabbitMQ Tutorials | RabbitMQ

第四步:编写生产者和消费者

   **这里写一个登录成功发送消息给登录人的业务(普通队列模式)** 
  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到

生产者

    ![](https://i-blog.csdnimg.cn/direct/dc36150199a44f54b3ec6e3a088ee92d.png)
  1. String exchange(交换机名):- 交换器的名称。交换器是一种逻辑实体,用于分发消息到相应的队列。你可以创建自定义的交换器类型,如 Direct、Fanout、Topic 或者 Headers 等。

  2. String routingKey(队列名):- 路由键。它是消息与队列之间的桥梁,决定了消息将被发送到哪个队列。路由键的值应匹配队列绑定到交换器时所使用的键。

  3. Object message(消息对象):- 要发送的消息对象。这个对象将被转换成字节流,然后发送到 RabbitMQ。Spring AMQP 使用 MessageConverter 将 Java 对象转换为 AMQP 格式的字节流。默认情况下,SimpleMessageConverter 可以处理简单的字符串、Java 序列化对象等。你可以通过配置 RabbitTemplatemessageConverter 属性来自定义消息转换器。

  4. MessagePostProcessor messagePostProcessor:- 这是一个接口,用于在消息发送前对其进行进一步的修改。你可以在这个处理器中设置消息头、属性或其他特定信息。MessagePostProcessor 接口有一个方法 postProcessMessage(Message message),你可以在这里修改消息实例。

  5. CorrelationData correlationData:- 相关数据,主要用于跟踪消息的发送结果。在异步发送消息的情况下,你可以使用此参数来关联发送请求和响应。CorrelationData 包含一个唯一标识符,用于识别消息。当收到响应时,可以通过这个标识符找到原始请求。

     发邮件是发给一个人,所以这里只需要传一个队列名和登录人对象。(发送给多个队列时需            要用到交换机)
    
     ![](https://i-blog.csdnimg.cn/direct/dbabc78f07844ef5b2c1fce44849704b.png)
    

消费者:

//@Header从消息头中提取消息的递送标签,标签是唯一的,对应唯一的消息
@RabbitListener(queuesToDeclare = @Queue("login_queue"))
public void test01(User getUser, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long         
                   deliveryTag)
{
    try {
             发送邮件的代码……
            
            //手动确认消息
            channel.basicAck(deliveryTag,true);
        } catch (MessagingException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

}

** 如果配置了手工消费,一定要消费完后手动确认消息,不然就会重复接收该消息,导致后面的消息接受不到!**


本文转载自: https://blog.csdn.net/WYH1233211234567/article/details/142743056
版权归原作者 除了代码啥也不会 所有, 如有侵权,请联系我们删除。

“springboot 整合 rabbitMQ(1)”的评论:

还没有评论