文档章节

消息队列的消费语义和投递语义

编辑之路
 编辑之路
发布于 06/19 14:41
字数 1311
阅读 2108
收藏 24

引言

所谓的消费语义,指的就是如下三种情况

  • 如何保证消息最多消费一次
  • 如何保证消息至少消费一次
  • 如何保证消息恰好消费一次

其实类似还有一个投递语义

  • 如何保证消息最多投递一次
  • 如何保证消息至少投递一次
  • 如何保证消息恰好投递一次

说句实在话,其实还是老问题,只是换了一种问法!
OK,开始我们的正文

正文

我们先做如下约定

  • Producer代表生产者
  • Consumer代表消费者
  • Message Queue代表消息队列

投递语义

我们先从投递语义开始讲起,因为要先把这个概念讲明白了,才能讲消费语义。恰巧,kafka实现了这三种语义,我们以kafka来说明。

如何保证消息最多投递一次?
简单,就是我已经投出去了,收没收到不管了,会存在消息丢失。
我们在初始化Producer时可以通过配置request.required.acks不同的值,来实现不同的发送模式。
这里将request.required.acks设为0,意思就是Producer不等待Leader确认,只管发出即可;最可能丢失消息。如果丢了消息,就是投递0次。如果没丢,就是投递1次。符合最多投递一次的含义。

如何保证消息至少投递一次?
这里将request.required.acks设为-1。ProducerkafkaLeader(主)节点发送消息后,会等follower(从)节点同步完数据以后,再给Producer返回ACK确认消息。
但是这里是有几率出现重复消费的问题的。
例如,kafka保存消息后,发送ACK前宕机,Producer认为消息未发送成功并重试,造成数据重复!
那么,在这种情况下,就会出现大于1次的投递情况,符合至少投递一次的含义。

如何保证消息恰好投递一次?
kafka在0.11.0.0版本之后支持恰好投递一次的语义。
我们将enable.idempotence设置为ture,此时就会默认把request.required.acks设为-1,可以达到恰好投递一次的语义。
如何做到的?
为了实现Producer的幂等语义,Kafka引入了Producer ID(即PID)和Sequence Number。
kafka为每个Producer分配一个pid,作为该Producer的唯一标识。
Producer会为每一个<topic,partition>维护一个单调递增的seq。
类似的,Message Queue也会为每个<pid,topic,partition>记录下最新的seq。
当req_seq == message_seq+1时,Message Queue才会接受该消息。因为:

  • (1)消息的seq比Message Queue的seq大一以上,说明中间有数据还没写入,即乱序了。
  • (2)消息的seq比Message Queue的seq小,那么说明该消息已被保存。

消费语义

这里我们还是做一个定义如下所示

  • consumer.poll()表示消费者获取消息内容
  • processMsg(message)表示下游系统进行消费消息
  • consumer.commit()表示消费者往消息队列提交确认信息,消息队列接到确认消息,删除该消息。

注意了,我是以processMsg函数,即处理消息的过程,定义为消费消息。
如何保证消息最多消费一次?
Producer:满足最多投递一次的语义即可,即只管发消息,不需要等待消息队列返回确认消息。
Message Queue:接到消息后往内存中一放就行,不用持久化存储。
Consumer:拉取到消息以后,直接给消息队列返回确认消息即可。至于后续消费消息成功与否,无所谓的。即按照以下顺序执行

consumer.poll();
consumer.commit();
processMsg(message);

如何保证消息至少消费一次?
Producer:满足至少投递一次语义即可,即发送消息后,需要等待消息队列返回确认消息。如果超时没收到确认消息,则重发。
Message Queue:接到消息后,进行持久化存储,而后返回生产者确认消息。
Consumer:拉取到消息后,进行消费,消费成功后,再返回确认消息。即按照如下顺序执行

consumer.poll();
processMsg(message);
consumer.commit();

由于这里Producer满足的是至少投递一次语义,因此消息队列中是有重复消息的。所以我们的Consumer会出现重复消费的情形!

如何保证消息恰好消费一次?
在保证至少消费一次的基础上,processMsg满足幂等性操作即可。
如何保证幂等性操作?
老问题了,比如有状态的消息啊。比如唯一表啊。大家搜一搜,一大堆答案,不想重复说了。

总结

本文讲的是消息队列的消费语义和投递语义的含义,希望大家有所收获。

出处:https://www.cnblogs.com/rjzheng/p/11050673.html

© 著作权归作者所有

编辑之路
粉丝 52
博文 48
码字总数 104142
作品 0
长沙
私信 提问
加载中

评论(2)

SilverMissile
SilverMissile
可以用坐标轴来理解,1左边最多一次,1右边至少一次,1上有且仅有一次
gitOpen_1
gitOpen_1
恰好投递一次,不久是TCP的序号和确认机制么😂
Spring AMQP 源码分析MessageListener

准备 ## 目标 了解 Spring AMQP 如何实现异步消息投递(推模式) ## 前置知识 《RabbitMQ入门05多线程消费同一队列》 ## 相关资源 Quick Tour for the impatient:http://docs.spring.io/spr...

writeademo
2018/08/12
0
0
Kafka消息投递语义-消息不丢失,不重复,不丢不重

介绍 kafka支持3种消息投递语义: At most once——最多一次,消息可能会丢失,但不会重复 At least once——最少一次,消息不会丢失,可能会重复 Exactly once——只且一次,消息不丢失不重...

watermelon11
02/19
0
0
深度 | 金融级消息队列的演进 — 蚂蚁金服的实践之路

消息队列作为一个数据的集散中心,承载了越来越多的场景和数据,从最开始的 OLTP 到 OLAP,甚至再到物联网、人工智能、机器学习等场景,都有很大的想像空间。 在能力上,消息队列现在拥有了数...

花肉酱
2018/06/14
0
0
springboot rabbitmq 之死信队列(延迟消费消息)

之前探讨了springboot 集成 rabbitmq 以及开启ack模式 传送门:https://my.oschina.net/u/2948566/blog/1624963 接着该篇 搞一下 死信队列 概念 死信队列 听上去像 消息“死”了 其实也有点这...

NotFound403
2018/02/28
0
2
rabbitMQ 基础之Exchange Type

首先先介绍下rabbitmq的一些基础概念 1、队列、生产者、消费者 队列是RabbitMQ的内部对象,用于存储消息。P(生产者)生产消息并投递到队列中,C(消费者)可以从队列中获取消息并消费。 多个...

zhu_kai1
01/06
0
0

没有更多内容

加载失败,请刷新页面

加载更多

php 遇到 No input file specified的解决方法

(一)IIS Noinput file specified 方法一:改PHP.ini中的doc_root行,打开ini文件注释掉此行,然后重启IIS 方法二: 请修改php.ini 找到 ; cgi.force_redirect = 1 去掉前面分号,把后面的1...

chenhongjiang
今天
5
0
MySQL 基础

一、常用命令 在命令行中,配置好环境变量后,通过cmd可以直接进入mysql命令行模式,同时列举几种常用命令 # 进入mysql数据库,密码可以先不写,打完-p后再输入,防止被别人看到mysql -u账...

华山猛男
今天
6
0
简单的博客系统(四)Django请求HTML页面视图信息--基于函数的视图

1. 编写用于查询数据的功能函数 应用目录 下的 views.py 文件通常用于保存响应各种请求的函数或类 from django.shortcuts import renderfrom .models import BlogArticles# Create your ...

ZeroBit
今天
5
0
用脚本将本地照片库批量导入到Day One中

因为目前iCloud 空间已经不足,其中95%都是照片,之前入手了DayOne,且空间没有限制,订阅费一年也不少,再加上DayOne作为一款日记App 也比较有名,功能方面最大的就是地理视图与照片视图,尤...

在山的那边
昨天
19
0
jupyter部署安装

python373 -m ipykernel install --name python373 ipython kernelspec list sc create myjupyterservice binpath="D:\apply\Python373\Scripts\jupyter-notebook --config=V:/my_work/jupyt......

mbzhong
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部