0


RabbitMQ

一、简介

    RabbitMQ 是一个消息代理:它接受并转发消息。     
  • 生产仅仅意味着发送,发送消息的程序是生产者。
  • 队列是 RabbitMQ 中邮箱的名称,尽管消息通过 RabbitMQ 和您的应用程序流动,但它们只能存储在队列中。队列仅受主机的内存和磁盘限制,它本质上是一个大型消息缓冲区。许多生产者可以发送消息到一个队列,许多消费者可以尝试从一个队列接收数据。
  • 消费具有与接收类似的含义。消费者是一个主要等待接收消息的程序

生产者、消费者和代理不必驻留在同一主机上;应用程序既可以是生产者,也可以是消费者。

1、大多应用中,可通过消息服务中间件来提升系统异步通信、扩展解耦能力。

2、当消息发送者发送消息以后,将由消息代理接管,消息代理保证消息传递到指定目的地。

3、消息队列主要有两种形式的目的地:

    1. 队列(queue):点对点消息通信(point-to-point)

    2. 主题(topic):发布(publish)/订阅(subscribe)消息通信

4、点对点式:

    消息发送者发送消息,消息代理将其放入一个队列中,消息接收者从队列中获取消息内容,消息读取后被移出队列。消息只有唯一的发送者和接受者,但并不是说只能有一个接收者。

5、发布订阅式:

    发送者(发布者)发送消息到主题,多个接收者(订阅者)监听(订阅)这个主题,那么就会在消息到达时同时收到消息。

二、核心概念

Message

    消息,消息是不具名的,它由消息头和消息体组成。消息体是不透明的,而消息头则由一系列的可选属性组成, 这些属性包括routing-key(路由键)、priority(相对于其他消息的优先权)、delivery-mode(指出该消息可能需要持久性存储)等。

Publisher

    消息的生产者,也是一个向交换器发布消息的客户端应用程序。

Exchange

    交换器,用来接收生产者发送的消息并将这些消息路由给服务器中的队列。 Exchange有4种类型:direct(默认),fanout, topic, 和headers,不同类型的Exchange转发消息的策略有所区别。

Queue

    消息队列,用来保存消息直到发送给消费者。它是消息的容器,也是消息的终点。一个消息可投入一个或多个队列。消息一直 在队列里面,等待消费者连接到这个队列将其取走。

Binding

    绑定,用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则,所以可以将交 换器理解成一个由绑定构成的路由表。 Exchange 和Queue的绑定可以是多对多的关系。

Connection

    网络连接,比如一个TCP连接。

Channel

    信道,多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接,AMQP 命令都是通过信道 发出去的,不管是发布消息、订阅队列还是接收消息,这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都 是非常昂贵的开销,所以引入了信道的概念,以复用一条 TCP 连接。

Consumer

    消息的消费者,表示一个从消息队列中取得消息的客户端应用程序。

Virtual Hos

    虚拟主机,表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器,拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础,必须在连接时指定,RabbitMQ 默认的 vhost 是 /

Broker

    表示消息队列服务器实体。

三、RabbitMQ运行机制

AMQP 中的消息路由

    AMQP 中消息的路由过程和 Java 开 发者熟悉的 JMS 存在一些差别, AMQP 中增加了 Exchange 和 Binding 的角色。生产者把消息发布 到 Exchange 上,消息最终到达队列并被消费者接收,而 Binding 决定交换机的消息应该发送到哪个队列。

Exchange的类型

    Exchange分发消息时根据类型的不同分发策略有区别,目前共四种类型:direct、 fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键, headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了。

四、引入RabbitMQ

1、引入依赖

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

2、配置

spring.rabbitmq.host=192.168.56.10
spring.rabbitmq.virtual-host=/

3、RabbitTemplate消息发送处理

    @RabbitListener 监听消息。

   
 @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMQ")
    public String sendTestMQ(){
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setOrderSn(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("order.release.order.queue", "order.create.order", orderEntity);
        return "ok";
    }
@RabbitListener(queues = "order.release.order.queue")
    public void listener(OrderEntity orderEntity, Channel channel, Message message){
        System.out.println(orderEntity.getOrderSn());
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

五、RabbitMQ消息确认机制

消费者获取到消息,成功处理,可以回复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延时队列

    场景: 比如未付款订单,超过一定时间后,系统自动取消订单并释放占有物品。

消息的TTL就是消息的存活时间

    RabbitMQ可以对队列和消息分别设置TTL

    对队列设置就是队列没有消费者连着的保留时间,也可以对每一个单独的消息做单独的设置。超过了这个时间,我们认为这个消息就死了,称之为死信。

    一个消息在满足如下条件下,会进死信路由。

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

    我们既可以控制消息在一段时间后变成死信,又可以控制变成死信的消息被路由到某一个指定的交换机。

手动ack&异常消息统一放在一个队列处理建议的两种方式:

    catch异常后,手动发送到指定队列,然后使用channel给rabbitmq确认消息已消费

    给Queue绑定死信队列,使用nack(requque为false)确认消息消费失败

**延时队列 **

@Bean
    public Queue stockDelayQueue(){
        /**
         * String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
         */
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "stock-event-exchange");
        arguments.put("x-dead-letter-routing-key", "stock.release");
        arguments.put("x-message-ttl", 120000);

        return new Queue("stock.delay.queue", true, false, false, arguments);
    }

七、消息可靠性

1、消息丢失

    **消息发送出去,由于网络问题没有抵达服务器**

            ①做好容错方法(try-catch),发送消息可能会网络失败,失败后要有重试机 制,可记录到数据库,采用定期扫描重发的方式

            ②做好日志记录,每个消息状态是否都被服务器收到都应该记录

            ③做好定期重发,如果消息没有发送成功,定期去数据库扫描未成功的消息进 行重发

    **消息抵达Broker,Broker要将消息写入磁盘(持久化)才算成功。此时Broker尚 未持久化完成,宕机。**

            ①publisher也必须加入确认回调机制,确认成功的消息,修改数据库消息状态。

    **自动ACK的状态下。消费者收到消息,但没来得及消息然后宕机 **,一定开启手动ACK,消费成功才移除,失败或者没来得及处理就noAck并重新入队。

2、消息重复

    消息消费成功,事务已经提交,ack时,机器宕机。导致没有ack成功,Broker的消息重新由unack变为ready,并发送给其他消费者,消息消费失败,由于重试机制,自动又将消息发送出去 ,成功消费,ack时宕机,消息由unack变为ready,Broker又重新发送。

    消费者的业务消费接口应该设计为幂等性的。

    使用防重表(redis/mysql),发送消息每一个都有业务的唯一标识,处理过就不用处理

    rabbitMQ的每一个消息都有redelivered字段,可以获取是否是被重新投递过来的,而不是第一次投递过来的。
标签: rabbitmq 分布式

本文转载自: https://blog.csdn.net/weixin_45079474/article/details/140421182
版权归原作者 尘埃不入你眼眸 所有, 如有侵权,请联系我们删除。

“RabbitMQ”的评论:

还没有评论