rabbitmq数据实战

rabbitmq数据实战

如何保证消息的顺序性

业务场景:我们需要根据mysql的binlog日志同步一个数据库的数据到另一个库中,加如在binlog中对同一条数据做了insert,update,delete操作,我们往MQ顺序写入了insert,update,delete操作的三条消息,那么根据分析,最终同步到另一个库中,这条数据是被删除了的。但是,如果这三条消息不是按照insert,update,delete顺序被消费,而是按照delete,insert,update的顺序被消费,那么最终这条数据是会保存到新库中的。这就导致了数据错乱了。下面讲解下RabbitMQ是如何保证消息的顺序性。

注意:queue(队列)中的消息只能被一个消费者所消费,然后消费者在消费消息的过程中是无序的。如上图所示,如果按照BAC的消费顺序,那么最终数据库中是被保存这条数据的。这和我们预期的结果不符,如果这样的情况很多,那么就造成了数据库中的数据完成不对,同步工作也是白费了。

如图所示,RabbitMQ保证消息的顺序性,就是拆分多个 queue,每个 queue 对应一个 consumer(消费者),就是多一些 queue 而已,确实是麻烦点;或者就一个 queue 但是对应一个 consumer,然后这个 consumer 内部用内存队列做排队,然后分发给底层不同的 worker 来处理。

如何保证消息不丢失


1,生产者发送消息至MQ的数据丢失

解决方法:在生产者端开启comfirm 确认模式,你每次写的消息都会分配一个唯一的 id,

然后如果写入了 RabbitMQ 中,RabbitMQ 会给你回传一个 ack 消息,告诉你说这个消息 ok 了

2,MQ收到消息,暂存内存中,还没消费,自己挂掉,数据会都丢失

解决方式:MQ设置为持久化。将内存数据持久化到磁盘中

3,消费者刚拿到消息,还没处理,挂掉了,MQ又以为消费者处理完

解决方式:用 RabbitMQ 提供的 ack 机制,简单来说,就是你必须关闭 RabbitMQ 的自动 ack,可以通过一个 api 来调用就行,然后每次你自己代码里确保处理完的时候,再在程序里 ack 一把。这样的话,如果你还没处理完,不就没有 ack 了?那 RabbitMQ 就认为你还没处理完,这个时候 RabbitMQ 会把这个消费分配给别的 consumer 去处理,消息是不会丢的。

如何保证消息不被重复消费

业务场景:假设你有个系统,消费一条消息就往数据库里插入一条数据,要是你一个消息重复两次,你不就插入了两条,这数据不就错了?但是你要是消费到第二次的时候,自己判断一下是否已经消费过了,若是就直接扔了,这样不就保留了一条数据,从而保证了数据的正确性。

一条数据重复出现两次,数据库里就只有一条数据,这就保证了系统的幂等性。

幂等性,通俗点说,就一个数据,或者一个请求,给你重复来多次,你得确保对应的数据是不会改变的,不能出错。

解决方案:

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