一、消息队列如何解决消息不会丢失问题
消息从生产到消费可以经历三个阶段:生产阶段、存储阶段和消费阶段。
生产阶段:在这个阶段,从消息在Producer创建出来,经过网络传输发送到Broker端。
存储阶段: 消息在Broker端存储,如果是集群,消息会在这个阶段被复制到其他的副本上。
消费阶段:Consumer从Broker上拉取消息,经过网络 传输发送在Consumer上。
在这三个阶段都存在消息可能丢失的情况。
生产阶段:消息队列通常使用确认机制,来保证消息可靠传递:当你代码调用发送消息的方法,消息队列的客户端会把消息发送到Broker,Broker接受到消息会返回客户端一个确认。只要Producer收到了Broker的确认响应,就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送的确认响应后,会自动重试,如果重试再失败,就会一返回值或者异常方式返回给客户端。所以在编写发送消息的代码,需要正确处理消息发送返回值或者异常,保证这个阶段消息不丢失。
存储阶段:如果对消息可靠性要求非常高,可以通过配置Broker参数来避免因为宕机丢消息。对于单个节点Broker,需要配置Broker参数,在收到消息后,将消息写入磁盘再给Producer返回确认响应。如果是Broker集群,需要将Broker集群配置成:至少两个以上节点收到消息,再给客户端发送确认响应。
消费阶段:消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递。Consumer收到消息后,需在执行消费逻辑后在发送确认消息。
总结:
生产阶段,需要捕获消息发送错误,并重发消息
存储阶段,通过配置刷盘和复制参数,让消息写入多个副本的磁盘上,来确保消息不会因为某个Broker宕机或者磁盘损坏而丢失。
消费阶段:需要在处理完全部消费业务逻辑后,再发送确认消息。
二、解决消息不会重复消费
幂等性:一个请求,不管重复来多少次,结果是不会改变的。
RabbitMQ、RocketMQ、Kafka等任何队列不保证消息不重复,如果业务需要消息不重复消费,则需要消费端处理业务消息要保持幂等性
方式一:Redis的setNX) , 做消息id去重 java版本目前不支持设置过期时间
//Redis中操作,判断是否已经操作过 TODO boolean flag = jedis.setNXkey); ifflag){ //消费 }else{ //忽略,重复消费 }
方式二:redis的 Incr 原子操作:key自增,大于0 返回值大于0则说明消费过,key可以是消息的md5取值, 或者如果消息id设计合理直接用id做key)
int num = jedis.incrkey); ifnum == 1){ //消费 }else{ //忽略,重复消费 }
方式三:数据库去重表
设计一个去重表,某个字段使用Message的key做唯一索引,因为存在唯一索引,所以重复消费会失败
CREATE TABLE message_record
id
int11) unsigned NOT NULL AUTO_INCREMENT, key
varchar128) DEFAULT NULL, create_time
datetime DEFAULT NULL, PRIMARY KEY id
), UNIQUE KEY key
key
) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
三、保证消息队列的高可用行
建立高可用集群
四、消息积压在消息队列里面
1.大量消息在mq里积压了几个小时了还没解决
场景:几千万条数据在MQ里积压了七八个小时,从下午4点多,积压到了晚上很晚,10点多,11点多。线上故障了,这个时候要不然就是修复consumer的问题,让他恢复消费速度,然后傻傻的等待几个小时消费完毕。这个肯定不行。一个消费者一秒是1000条,一秒3个消费者是3000条,一分钟是18万条,1000多万条。
所以如果你积压了几百万到上千万的数据,即使消费者恢复了,也需要大概1小时的时间才能恢复过来。
解决方案:”
这种时候只能操作临时扩容,以更快的速度去消费数据了。具体操作步骤和思路如下:
①先修复consumer的问题,确保其恢复消费速度,然后将现有consumer都停掉。
②临时建立好原先10倍或者20倍的queue数量新建一个topic,partition是原来的10倍)。
③然后写一个临时分发消息的consumer程序,这个程序部署上去消费积压的消息,消费之后不做耗时处理,直接均匀轮询写入临时建好分10数量的queue里面。
④紧接着征用10倍的机器来部署consumer,每一批consumer消费一个临时queue的消息。
⑤这种做法相当于临时将queue资源和consumer资源扩大10倍,以正常速度的10倍来消费消息。
⑥等快速消费完了之后,恢复原来的部署架构,重新用原来的consumer机器来消费消息。
2.消息设置了过期时间,过期就丢了怎么办
假设你用的是rabbitmq,rabbitmq是可以设置过期时间的,就是TTL,如果消息在queue中积压超过一定的时间就会被rabbitmq给清理掉,这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在mq里,而是大量的数据会直接搞丢。
解决方案:
这种情况下,实际上没有什么消息挤压,而是丢了大量的消息。所以第一种增加consumer肯定不适用。
这种情况可以采取 “批量重导” 的方案来进行解决。
在流量低峰期比如夜深人静时),写一个程序,手动去查询丢失的那部分数据,然后将消息重新发送到mq里面,把丢失的数据重新补回来。
3.积压消息长时间没有处理,mq放不下了怎么办
如果走的方式是消息积压在mq里,那么如果你很长时间都没处理掉,此时导致mq都快写满了,咋办?这个还有别的办法吗?
解决方案:
这个就没有办法了,肯定是第一方案执行太慢,这种时候只好采用 “丢弃+批量重导” 的方式来解决了。
首先,临时写个程序,连接到mq里面消费数据,收到消息之后直接将其丢弃,快速消费掉积压的消息,降低MQ的压力,然后走第二种方案,在晚上夜深人静时去手动查询重导丢失的这部分数据。