RabbitMQ集群与消息可靠性

野性酷女 2022-09-16 05:55 260阅读 0赞

# 一、RabbitMQ集群架构 #

RabbitMQ集群架构有以下几种：主备模式、远程模式、镜像模式、多活模式。

## **主备模式** ##

就是一个主/备方案（主节点如果挂了，从节点提供服务而已）

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16][]

HaProxy配置：

listen rabbitmq_cluster
      bind 0.0.0.0:5672 
      mode tcp  #配置TCP模式
      balance roundrobin #简单的轮询
      server bhz76 192.168.11.12:5672 check inter 5000 rise 2 fall 3 #主节点
      server bhz77 192.168.11.13:5672 backup check inter 5000 rise 2 fall 3 #备用节点

备注：rabbitmq集群节点配置 \#inter 每隔5秒对mq集群做健康检查，2次正确证明服务器可用，3次失败证明服务器不可用，并且配置主备机制。

## 远程模式（不常用） ##

远距离通信和复制，就是我们可以把消息进行不同数据中心的复制工作，我们可以跨地域的让两个mq集群互联。

模型变成了近端同步确认，远端异步确认方式，大大提高了订单确认速度，并且还能保证可靠性。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

##  镜像模式（常用） ##

就是实现数据的同步，一般来讲是2-3个实现数据同步（对于100%数据可靠性解决方案一般是3个节点），保证100%数据不丢失，在实际工作中用的最多的。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

##  多活模式 ##

采用双中心模式（多中心），那么在两套（或多套）数据中心中各部署一套RabbitMQ集群，各中心之间还需要实现部分队列消息共享。实现异地数据复制的主流模式，因为远程模式配置比较复杂，所以一般来说实现异地集群都是使用双活或者多活模式来实现的。

这种模式需要依赖rabbitmq的federation插件，可以实现继续的可靠AMQP数据通信，多活模式在实际配置与应用非常的简单。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

备注：

主备模式和主从模式的区别：

*  主备模式：主节点提供读写，从节点不提供读写服务，只是负责提供备份服务,备份节点的主要功能是在主节点宕机时，完成自动切换 从-->主
 *  主从模式：主节点提供读写，从节点只读

Federation插件

*  Federation插件是一个不需要构建Cluster，而在Brokers之间传输消息的高性能插件，
 *  Federation插件可以在Brokers或者Cluster之间传输消息，连接双方可以使用不同的users和vistual hosts，
 *  双方也可以使用版本不同的RabbitMQ和Erlang。Federation插件使用AMQP协议通信，可以接收不连续的传输。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 4][]

Federation Exchanges,可以看成Downstream从Upstream主动拉取消息，但并不是拉取所有消息，必须是在Downstream上已经明确定义Bindings关系的Exchange，也就是有实际的物理Queue来接收消息，才会从Upstream拉取消息到Downstream。使用AMQP协议实施代理间通信，Downstream会将绑定关系组合在一起，绑定/解绑命令将会发送到Upstream交换机。因此，FederationExchange只接收具有订阅的消息。

# 二、RabbitMQ如何保证消息的可靠性 #

## 首先看一下RabbitMQ为什么不可靠 ##

RabbitMQ丢失的以下3种情况：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 5][]

1.  生产者方面：生产者发送消息至MQ的数据丢失
2.  RabbitMQ方面：MQ收到消息，暂存内存中，还没消费，自己挂掉，数据会都丢失
3.  消费者方面：消费者刚拿到消息，还没处理，挂掉了，MQ又以为消费者处理完

## 针对不可靠问题的解决方案 ##

我们针对这几方问题分别列出解决方案。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 6][]

### （1）生产者方面 ###

生产者方面：有两种方案，一是开启RabbitMQ事务（不推荐），二是开启confirm模式（异步，推荐）

**开启RabbitMQ事务**

AMQP协议提供了事务机制，在投递消息时开启事务支持，如果消息投递失败，则回滚事务。

// 1.自定义事务管理器
    @Configuration
    public class RabbitTranscation {
        
        @Bean
        public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory){
            return new RabbitTransactionManager(connectionFactory);
        }
    @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
            return new RabbitTemplate(connectionFactory);
        }
    }
     
    // 2.修改yml
    spring:
      rabbitmq:
        # 消息在未被队列收到的情况下返回
        publisher-returns: true
     
    // 3.开启事务支持
    rabbitTemplate.setChannelTransacted(true);
     
    // 4.消息未接收时调用ReturnCallback
    rabbitTemplate.setMandatory(true);
     
    // 5.生产者投递消息
    @Service
    public class ProviderTranscation implements RabbitTemplate.ReturnCallback {
    @Autowired
        RabbitTemplate rabbitTemplate;
    @PostConstruct
        public void init(){
            // 设置channel开启事务
            rabbitTemplate.setChannelTransacted(true);
            rabbitTemplate.setReturnCallback(this);
        }
        
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("这条消息发送失败了"+message+",请处理");
        }
        
        @Transactional(rollbackFor = Exception.class,transactionManager = "rabbitTransactionManager")
        public void publishMessage(String message) throws Exception {
            rabbitTemplate.setMandatory(true);
            rabbitTemplate.convertAndSend("javatrip",message);
        }
    }

但是，很少有人这么干，因为这是同步操作，一条消息发送之后会使发送端阻塞，以等待RabbitMQ-Server的回应，之后才能继续发送下一条消息，生产者生产消息的吞吐量和性能都会大大降低。

**开启confirm模式**

发送消息时将信道设置为confirm模式，消息进入该信道后，都会被指派给一个唯一ID，一旦消息被投递到所匹配的队列后，RabbitMQ就会发送给生产者一个确认。

// 1.开启消息确认机制
    spring:
      rabbitmq:
        # 消息在未被队列收到的情况下返回
        publisher-returns: true
        # 开启消息确认机制
        publisher-confirm-type: correlated
     
    // 2.消息未接收时调用ReturnCallback
    rabbitTemplate.setMandatory(true);
     
    // 3.生产者投递消息
    @Service
    public class ConfirmProvider implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    @Autowired
        RabbitTemplate rabbitTemplate;
    @PostConstruct
        public void init() {
            rabbitTemplate.setReturnCallback(this);
            rabbitTemplate.setConfirmCallback(this);
        }
    @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if(ack){
                System.out.println("确认了这条消息："+correlationData);
            }else{
                System.out.println("确认失败了："+correlationData+"；出现异常："+cause);
            }
        }
    @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("这条消息发送失败了"+message+",请处理");
        }
    public void publisMessage(String message){
            rabbitTemplate.setMandatory(true);
            rabbitTemplate.convertAndSend("javatrip",message);
        }
    }
     
    // 4.如果消息确认失败后，我们可以进行消息补偿，也就是消息的重试机制。当未收到确认信息时进行消息的重新投递。设置如下配置即可完成。
    spring:
      rabbitmq:
        # 支持消息发送失败后重返队列
        publisher-returns: true
        # 开启消息确认机制
        publisher-confirm-type: correlated
        listener:
          simple:
            retry:
              # 开启重试
              enabled: true
              # 最大重试次数
              max-attempts: 5
              # 重试时间间隔

### （2）RabbitMQ方面 ###

开启RabbitMQ持久化，将内存数据持久化到磁盘中。

创建队列的时候将持久化属性durable设置为true，同时要将autoDelete设置为false

@Queue(value = "javatrip",durable = "false",autoDelete = "false")

**持久化消息：**发送消息的时候将消息的deliveryMode设置为2，在Spring Boot中消息默认就是持久化的。

### （3）消费者方面 ###

关闭RabbitMQ自动ACK。消费者刚消费了消息，还没有处理业务，结果发生异常。这时候就需要关闭自动确认，改为手动确认消息。

// 1.修改yml为手动签收模式
    spring:
      rabbitmq:
        listener:
          simple:
            # 手动签收模式
            acknowledge-mode: manual
            # 每次签收一条消息
            prefetch: 1
     
     
    // 2.消费者手动签收
    @Component
    @RabbitListener(queuesToDeclare = @Queue(value = "javatrip", durable = "true"))
    public class Consumer {
    @RabbitHandler
        public void receive(String message, @Headers Map<String,Object> headers, Channel channel) throws Exception{
    System.out.println(message);
            // 唯一的消息ID
            Long deliverTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            // 确认该条消息
            if(...){
                channel.basicAck(deliverTag,false);
            }else{
                // 消费失败，消息重返队列
                channel.basicNack(deliverTag,false,true);
            }
          
        }

# 三、RabbitMQ幂等问题 #

幂等性问题通俗点讲就是保证数据不被重复消费，同时数据也不能少（就是上述的可靠性），也就是数据一致性问题。

数据重复的问题简单的多，就是在消费端判断数据是否已经被消费过

*  比如你拿个数据要写库，你先根据主键查一下，如果这数据都有了，你就别插入了，update 一下好吧。
 *  比如你是写 Redis，那没问题了，反正每次都是 set，天然幂等性。
 *  比如你不是上面两个场景，那做的稍微复杂一点，你需要让生产者发送每条数据的时候，里面加一个全局唯一的 id，类似订单 id 之类的东西，然后你这里消费到了之后，先根据这个 id 去比如 Redis 里查一下，之前消费过吗？如果没有消费过，你就处理，然后这个 id 写 Redis。如果消费过了，那你就别处理了，保证别重复处理相同的消息即可。
 *  比如基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了，重复数据插入只会报错，不会导致数据库中出现脏数据。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/a0454d6c733141b9b345ca59cda91932.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/ae8c7a233b324d22b77f4a4341169382.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/48f0689ddc5446749279e3405c44ad31.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/68e7ec4ba776407ebb1a2bd1595206f3.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/a14fabad62a0483681b339d377290606.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220828/4be6419c4d0d419fba3dada370677499.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ri4546L5a2Q_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220828/720bca3e3f904cd08c0a2f19839eec1f.png