现在主流的消息队列产品都提供了非常完善的消息可靠性保证机制,完全可以做到在消息传递过程中,即使发生网络中断或者硬件故障,也能确保消息的可靠传递,不丢消息。
绝大部分丢消息的原因都是由于开发者不熟悉消息队列,没有正确使用和配置消息队列导致的。虽然不同的消息队列提供的 API 不一样,相关的配置项也不同,但是在保证消息可靠传递这块儿,它们的实现原理是一样的。
检查消息丢失的方法
利用消息队列的有序性来验证是否有消息丢失
在 Producer 端,我们给每个发出的消息附加一个连续递增的序号,然后在 Consumer 端来检查这个序号的连续性。如果没有消息丢失,Consumer 收到消息的序号必然是连续递增的,或者说收到的消息,其中的序号必然是上一条消息的序号 +1。如果检测到序号不连续,那就是丢消息了。还可以通过缺失的序号来确定丢失的是哪条消息,方便进一步排查原因。
大多数消息队列的客户端都支持拦截器机制,你可以利用这个拦截器机制,在 Producer 发送消息之前的拦截器中将序号注入到消息中,在 Consumer 收到消息的拦截器中检测序号的连续性,这样实现的好处是消息检测的代码不会侵入到你的业务代码中,待你的系统稳定后,也方便将这部分检测的逻辑关闭或者删除。
确保消息的可靠传递
- 生产阶段: 在这个阶段,从消息在 Producer 创建出来,经过网络传输发送到 Broker 端。
- 存储阶段: 在这个阶段,消息在 Broker 端存储,如果是集群,消息会在这个阶段被复制到其他的副本上。
- 消费阶段: 在这个阶段,Consumer 从 Broker 上拉取消息,经过网络传输发送到 Consumer 上。
一、生产阶段
当你的代码调用发消息方法时,消息队列的客户端会把消息发送到 Broker,Broker 收到消息后,会给客户端返回一个确认响应,表明消息已经收到了
只要 Producer 收到了 Broker 的确认响应,就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后,会自动重试,如果重试再失败,就会以返回值或者异常的方式告知用户。
正确处理返回值或者捕获异常,就可以保证这个阶段的的消息不对丢失
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json
import json
from kafka import KafkaConsumer, KafkaProducer
class KProducer:
def __init__(self, bootstrap_servers, topic, acks):
"""
kafka 生产者
:param bootstrap_servers: 地址
:param topic: topic
"""
self.producer = KafkaProducer(
bootstrap_servers=bootstrap_servers,
key_serializer=lambda k: json.dumps(k).encode('ascii'),
value_serializer=lambda k: json.dumps(k).encode('ascii'),
acks=acks
) # json 格式化发送的内容
self.topic = topic
def sync_producer(self, data_li: list):
"""
同步发送 数据
:param data_li: 发送数据
:return:
"""
for data in data_li:
future = self.producer.send(self.topic, data)
record_metadata = future.get(timeout=10) # 同步确认消费
partition = record_metadata.partition # 数据所在的分区
offset = record_metadata.offset # 数据所在分区的位置
print('save success, partition: {}, offset: {}'.format(partition, offset))
# 异步发送时,则需要在回调方法里进行检查。这个地方是需要特别注意的,很多丢消息的原因就是,我们使用了异步发送,却没有在回调中检查发送结果。
def async_producer(self, data_li: list):
"""
异步发送数据
:param data_li:发送数据
:return:
"""
for data in data_li:
self.producer.send(self.topic, data)
self.producer.flush() # 批量提交
def async_producer_callback(self, data_li: list):
"""
异步发送数据 + 发送状态处理
:param data_li:发送数据
:return:
"""
for data in data_li:
self.producer.send(self.topic, data).add_callback(self.send_success).add_errback(self.send_error)
self.producer.flush() # 批量提交
def send_success(self, *args, **kwargs):
"""异步发送成功回调函数"""
print('save success')
return
def send_error(self, *args, **kwargs):
"""异步发送错误回调函数"""
print('save error')
return
def close_producer(self):
try:
self.producer.close()
except:
pass
if __name__ == '__main__':
send_data_li = ["Egg", "西瓜", "呼伦贝尔"]
kp = KProducer(topic='gourd', bootstrap_servers=['127.0.0.1:9092'], acks="all")
# 同步发送
# kp.sync_producer(send_data_li)
# 异步发送
# kp.async_producer(send_data_li)
# 异步+回调
kp.async_producer_callback(send_data_li)
这里由于已经做了是否发送成功,在实例化producer阶段的acks是可以不用的
二、存储阶段
在存储阶段正常情况下,只要 Broker 在正常运行,就不会出现丢失消息的问题,但是如果 Broker 出现了故障,比如进程死掉了或者服务器宕机了,还是可能会丢失消息的。
如果对消息的可靠性要求非常高,可以通过配置 Broker 参数来避免因为宕机丢消息。
对于单个节点的 Broker,需要配置 Broker 参数,在收到消息后,将消息写入磁盘后再给 Producer 返回确认响应,这样即使发生宕机,由于消息已经被写入磁盘,就不会丢失消息,恢复后还可以继续消费。例如,在 RocketMQ 中,需要将刷盘方式 flushDiskType 配置为 SYNC_FLUSH 同步刷盘。
三、消费阶段
消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递,客户端从 Broker 拉取消息后,执行用户的消费业务逻辑,成功后,才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应,下次拉消息的时候还会返回同一条消息,确保消息不会在网络传输过程中丢失,也不会因为客户端在执行消费逻辑中出错导致丢失。
from kafka import KafkaConsumer
import json
def consumer_demo():
consumer = KafkaConsumer(
'gourd',
bootstrap_servers='47.116.26.72:9092',
group_id='test',
# earliest 从上一次未消费的位置开始读
# latest 当前时刻开始读之后产生的,之前产生的数据不再消费
auto_offset_reset="earliest",
# 是否自动commit,当前消费者消费完该数据后,需要commit,才可以将消费完的信息传回消息队列的控制中心。
# enable_auto_commit设置为True后,消费者将自动commit,并且两次commit的时间间隔为auto_commit_interval_ms。
enable_auto_commit=False, # 取消自动提交
consumer_timeout_ms="10"
)
for message in consumer:
print("receive, key: {}, value: {}".format(
json.loads(message.key.decode()),
json.loads(message.value.decode())
)
)
# 存入数据库
mysql.clinet("insert xxxxx json.loads(message.key.decode()) json.loads(message.value.decode()")
consumer.commit()
if __name__ == '__main__':
consumer_demo()
建议 一个进程一个consumer
正确的顺序是,先是把消息保存到数据库中,然后再发送消费确认响应。这样如果保存消息到数据库失败了,就不会执行消费确认的代码,下次拉到的还是这条消息,直到消费成功。
-
在生产阶段,你需要捕获消息发送的错误,并重发消息。
-
在存储阶段,你可以通过配置刷盘和复制相关的参数,让消息写入到多个副本的磁盘上,来确保消息不会因为某个 Broker 宕机或者磁盘损坏而丢失。
息保存到数据库中,然后再发送消费确认响应。这样如果保存消息到数据库失败了,就不会执行消费确认的代码,下次拉到的还是这条消息,直到消费成功。 -
在生产阶段,你需要捕获消息发送的错误,并重发消息。
-
在存储阶段,你可以通过配置刷盘和复制相关的参数,让消息写入到多个副本的磁盘上,来确保消息不会因为某个 Broker 宕机或者磁盘损坏而丢失。
-
在消费阶段,你需要在处理完全部消费业务逻辑之后,再发送消费确认。