udp怎么保证不丢包_在 Flink 算子中使用多线程如何保证不丢数据?

分析痛点

笔者线上有一个 Flink 任务消费 Kafka 数据,将数据转换后,在 Flink 的 Sink 算子内部调用第三方 api 将数据上报到第三方的数据分析平台。这里使用批量同步 api,即:每 50 条数据请求一次第三方接口,可以通过批量 api 来提高请求效率。由于调用的外网接口,所以每次调用 api 比较耗时。假如批次大小为 50,且请求接口的平均响应时间为 50ms,使用同步 api,因此第一次请求响应以后才会发起第二次请求。请求示意图如下所示:

ff9fabfac091201a2487e2b9fb006ec3.png

平均下来,每 50 ms 向第三方服务器发送 50 条数据,也就是每个并行度 1 秒钟处理 1000 条数据。假设当前业务数据量为每秒 10 万条数据,那么 Flink Sink 算子的并行度需要设置为 100 才能正常处理线上数据。从 Flink 资源分配来讲,100 个并行度需要申请 100 颗 CPU,因此当前 Flink 任务需要占用集群中 100 颗 CPU 以及不少的内存资源。请问此时 Flink Sink 算子的 CPU 或者内存压力大吗?

上述请求示意图可以看出 Flink 任务发出请求到响应这 50ms 期间,Flink Sink 算子只是在 wait,并没有实质性的工作。因此,CPU 使用率肯定很低,当前任务的瓶颈明显在网络 IO。最后结论是 Flink 任务申请了 100 颗 CPU,导致 yarn 或其他资源调度框架没有资源了,但是这 100 颗 CPU 的使用率并不高,这里能不能优化通过提高 CPU 的使用率,从而少申请一些 CPU 呢?

同步批量请求优化为异步请求

首先可以想到的是将同步请求改为异步请求,使得任务不会阻塞在网络请求这一环节,请求示意图如下所示。

279116aa6ae0af2876be50c29ce8b68b.png

异步请求相比同步请求而言,优化点在于每次发出请求时,不需要等待请求响应后再发送下一次请求,而是当下一批次的 50 条数据准备好之后,直接向第三方服务器发送请求。每次发送请求后,Flink Sink 算子的客户端需要注册监听器来等待响应,当响应失败时需要做重试或者回滚策略。

通过异步请求的方式,可以优化网络瓶颈,假如 Flink Sink 算子的单个并行度平均 10ms 接收到 50 条数据,那么使用异步 api 的方式平均 1 秒可以处理 5000 条数据,整个 Flink 任务的性能提高了 5 倍。对于每秒 10 万数据量的业务,这里仅需要申请 20 颗 CPU 资源即可。关于异步 api 的具体使用,可以根据场景具体设计,这里不详细讨论。

多线程 Client 模式

对于一些不支持异步 api 的场景,可能并不能使用上述优化方案,同样,为了提高 CPU 使用率,可以在 Flink Sink 端使用多线程的方案。如下图所示,可以在 Flink Sink 端开启 5 个请求第三方服务器的 Client 线程:Client1、Client2、Client3、Client4、Client5。

这五个线程内分别使用同步批量请求的 Client,单个 Client 还是保持 50 条记录为一个批次,即 50 条记录请求一次第三方 api。请求第三方 api 耗时主要在于网络 IO(性能瓶颈在于网络请求延迟),因此如果变成 5 个 Client 线程,每个 Client 的单次请求平均耗时还能保持在 50ms,除非网络请求已经达到了带宽上限或整个任务又遇到其他瓶颈。所以,多线程模式下使用同步批量 api 也能将请求效率提升 5 倍。

86f47dfc885429395ae5bf2bf685d593.png

说明:多线程的方案,不仅限于请求第三方接口,对于非 CPU 密集型的任务也可以使用该方案,在降低 CPU 数量的同时,单个 CPU 承担多个线程的工作,从而提高 CPU 利用率。例如:请求 HBase 的任务或磁盘 IO 是瓶颈的任务,可以降低任务的并行度,使得每个并行度内处理多个线程。

Flink 算子内多线程实现

Sink 算子的单个并行度内现在有 5 个 Client 用于消费数据,但 Sink 算子的数据都来自于上游算子。如下图所示,一个简单的实现方式是 Sink 算子接收到上游数据后通过轮循或随机的策略将数据分发给 5 个 Client 线程。

e7bae013923374fc918b7fe129dc0251.png

但是轮循或者随机策略会存在问题,假如 5 个 Client 中 Client3 线程消费较慢,会导致给 Client3 分发数据时被阻塞,从而使得其他正常消费的线程 Client1、2、4、5 也被分发不到数据。

为了解决上述问题,可以在 Sink 算子内申请一个数据缓冲队列,队列有先进先出(FIFO)的特性。Sink 算子接收到的数据直接插入到队列尾部,五个 Client 线程不断地从队首取数据并消费,即:Sink 算子先接收的数据 Client 先消费,后接收的数据 Client 后消费。

  • 若队列一直是满的,说明 Client 线程消费较慢、Sink 算子上游生产数据较快。
  • 若队列一直为空,说明 Client 线程消费较快、Sink 算子的上游生产数据较慢。

五个线程共用同一个队列完美地解决了单个线程消费慢的问题,当 Client3 线程阻塞时,不影响其他线程从队列中消费数据。这里使用队列还起到了削峰填谷的作用。

9cf01c56c1c9486593efcb25b3e37670.png

代码实现

原理明白了,具体代码如下所示,首先是消费数据的 Client 线程代码,代码逻辑很简单,一直从 bufferQueue 中 poll 数据,取出数据后,执行相应的消费逻辑即可,在本案例中消费逻辑便是 Client 积攒批次并调用第三方 api。

public class MultiThreadConsumerClient implements Runnable {private LinkedBlockingQueue<String> bufferQueue;public MultiThreadConsumerClient(LinkedBlockingQueue<String> bufferQueue) {this.bufferQueue = bufferQueue;}@Overridepublic void run() {String entity;while (true){// 从 bufferQueue 的队首消费数据entity = bufferQueue.poll();// 执行 client 消费数据的逻辑doSomething(entity);}}// client 消费数据的逻辑private void doSomething(String entity) {// client 积攒批次并调用第三方 api}
}

Sink 算子代码如下所示,在 open 方法中需要初始化线程池、数据缓冲队列并创建开启消费者线程,在 invoke 方法中只需要往 bufferQueue 的队尾添加数据即可。

public class MultiThreadConsumerSink extends RichSinkFunction<String> {// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM = 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY = 5000;private LinkedBlockingQueue<String> bufferQueue;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue = Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient = new MultiThreadConsumerClient(bufferQueue);for (int i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {threadPoolExecutor.execute(consumerClient);}}@Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}
}

代码逻辑相对比较简单,请问上述 Sink 能保证 Exactly Once 吗?

答:不能保证 Exactly Once,Flink 要想端对端保证 Exactly Once,必须要求外部组件支持事务,这里第三方接口明显不支持事务。

那么上述 Sink 能保证 At Lease Once 吗?言外之意,上述 Sink 会丢数据吗?

答:会丢数据。因为上述案例中使用的批量 api 来消费数据,假如批量 api 是每积攒 50 条数据请求一次第三方接口,当做 Checkpoint 时可能只积攒了 30 条数据,所以做 Checkpoint 时内存中可能还有数据未发送到外部系统。而且数据缓冲队列中可能还有缓存的数据,因此上述 Sink 在做 Checkpoint 时会出现 Checkpoint 之前的数据未完全消费的情况。

例如,Flink 任务消费的 Kafka 数据,当做 Checkpoint 时,Flink 任务消费到 offset 为 10000 的位置,但实际上 offset 10000 之前的一小部分数据可能还在数据缓冲队列中尚未完全消费,或者因为没积攒够一定批次所以数据缓存在 client 中,并未请求到第三方。当任务失败后,Flink 任务从 Checkpoint 处恢复,会从 offset 为 10000 的位置开始消费,此时 offset 10000 之前的一小部分缓存在内存缓冲队列中的数据不会再被消费,于是就出现了丢数据情况。

d75b75a7c4889e538882e2ab7a4045f1.png

处理丢数据情况

如何保证数据不丢失呢?很简单,可以在 Checkpoint 时强制将数据缓冲区的数据全部消费完,并对 client 执行 flush 操作,保证 client 端不会缓存数据。

实现思路:Sink 算子可以实现 CheckpointedFunction 接口,当做 Checkpoint 时,会调用 snapshotState 方法,方法内可以触发 client 的 flush 操作。但 client 在 MultiThreadConsumerClient 对应的五个线程中,需要考虑线程同步的问题,即:Sink 算子的 snapshotState 方法中做一个操作,要使得五个 Client 线程感知到当前正在执行 Checkpoint,此时应该把数据缓冲区的数据全部消费完,并对 client 执行过 flush 操作。

如何实现呢?需要借助 CyclicBarrier。CyclicBarrier 会让所有线程都等待某个操作完成后才会继续下一步行动。在这里可以使用 CyclicBarrier,让 Checkpoint 等待所有的 client 将数据缓冲区的数据全部消费完并对 client 执行过 flush 操作,言外之意,offset 10000 之前的数据必须全部消费完成才允许 Checkpoint 执行完成。这样就可以保证 Checkpoint 时不会有数据被缓存在内存,可以保证数据源 offset 10000 之前的数据都消费完成。

MultiThreadConsumerSink 具体代码如下所示:

public class MultiThreadConsumerSink extends RichSinkFunction<String> {// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM = 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY = 5000;private LinkedBlockingQueue<String> bufferQueue;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue = Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient = new MultiThreadConsumerClient(bufferQueue);for (int i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {threadPoolExecutor.execute(consumerClient);}}@Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}
}

MultiThreadConsumerSink 实现了 CheckpointedFunction 接口,在 open 方法中增加了 CyclicBarrier 的初始化,CyclicBarrier 预期容量设置为 client 线程数加一,表示当 client 线程数加一个线程都执行了 await 操作时,所有的线程的 await 方法才会执行完成。这里为什么要加一呢?因为除了 client 线程外, snapshotState 方法中也需要执行过 await。

当做 Checkpoint 时 snapshotState 方法中执行 clientBarrier.await(),等待所有的 client 线程将缓冲区数据消费完。snapshotState 方法执行过程中 invoke 方法不会被执行,即:Checkpoint 过程中数据缓冲队列不会增加数据,所以 client 线程很快就可以将缓冲队列中的数据消费完。

MultiThreadConsumerClient 具体代码如下所示:

public class MultiThreadConsumerSink extends RichSinkFunction<String> implements CheckpointedFunction {private Logger LOG = LoggerFactory.getLogger(MultiThreadConsumerSink.class);// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM = 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY = 5000;private LinkedBlockingQueue<String> bufferQueue;private CyclicBarrier clientBarrier;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue = Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// barrier 需要拦截 (DEFAULT_CLIENT_THREAD_NUM + 1) 个线程this.clientBarrier = new CyclicBarrier(DEFAULT_CLIENT_THREAD_NUM + 1);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient = new MultiThreadConsumerClient(bufferQueue, clientBarrier);for (int i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {threadPoolExecutor.execute(consumerClient);}}@Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}@Overridepublic void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {LOG.info("snapshotState : 所有的 client 准备 flush !!!");// barrier 开始等待clientBarrier.await();}@Overridepublic void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {}}

从数据缓冲队列中 poll 数据时,增加了 timeout 时间为 50ms。如果从队列中拿到数据,则执行消费数据的逻辑,若拿不到数据说明数据缓冲队列中数据消费完了。此时需要判断是否有等待的 CyclicBarrier,如果有等待的 CyclicBarrier 说明此时正在执行 Checkpoint,所以 client 需要执行 flush 操作。flush 完成后,Client 线程执行 barrier.await() 操作。当所有的 Client 线程都执行到 await 时,所有的 barrier.await() 都会被执行完。此时 Sink 算子的 snapshotState 方法就会执行完。通过这种策略可以保证 Checkpoint 时将数据缓冲区中的数据消费完,client 执行 flush 操作可以保证 client 端不会缓存数据。

总结

分析到这里,我们设计的 Sink 终于可以保证不丢失数据了。对 CyclicBarrier 不了解的同学请 Google 或百度查询。再次强调这里多线程的方案,不仅限于请求第三方接口,对于非 CPU 密集型的任务都可以使用该方案来提高 CPU 利用率,且该方案不仅限于 Sink 算子,各种算子都适用。本文主要希望帮助大家理解 Flink 中使用多线程的优化及在 Flink 算子中使用多线程如何保证不丢数据。

原文链接

本文为阿里云内容,未经允许不得转载

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/501827.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

php 没有libmysql.dll,PHP5.3以上版本没有libmysql.dll,以及由此带来的困扰

有朋友下载了PHP5.3&#xff0c;PHP5.4版本想加载mysql支持的时候发现没有libmysql.dll文件&#xff0c;无法完成mysql配置&#xff0c;其实PHP5.3版本开始&#xff0c;使用mysqlnd库&#xff0c;不再使用libmysql.dll用来访问MySQL数据库&#xff0c;大大提升了PHP的数据库访问…

堆排序时间复杂度_图解堆结构、堆排序及堆的应用

前言这次我们介绍另一种时间复杂度为 O(nlogn) 的选择类排序方法叫做堆排序。我将从以下几个方面介绍&#xff1a;堆的结构堆排序优化的堆排序原地堆排序堆的应用堆的结构什么是堆&#xff1f;我给出了百度的定义&#xff0c;如下&#xff1a;堆(Heap)是计算机科学中一类特殊的…

农历php,PHP农历公历转换

/*云南省曲靖师范学院计算机科学与工程学院-杨海熙编写2009-9-3*/class Lunar{private $_SMDay array(1 > 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31);//定义公历月分天数private $_LStart 1950 ;//农历从1950年开始private $_LMDay array(//差&#xff1a;该年的…

python迷宫起点终点所有路径_通向终点的路不止一条!python迷宫。

马上注册&#xff0c;结交更多好友&#xff0c;享用更多功能^_^您需要 登录 才可以下载或查看&#xff0c;没有帐号&#xff1f;立即注册x本帖最后由 wei_Y 于 2014-11-24 19:53 编辑2008121223547917.jpg (56.62 KB, 下载次数: 3)2014-11-17 19:22 上传其实木有上面那么高端啦…

恶意软件分析沙箱在网络安全策略中处于什么位置?

恶意软件分析沙箱提供了一种全面的恶意软件分析方法&#xff0c;包括静态和动态技术。这种全面的评估可以更全面地了解恶意软件的功能和潜在影响。然而&#xff0c;许多组织在确定在其安全基础设施中实施沙箱的最有效方法方面面临挑战。让我们看一下可以有效利用沙盒解决方案的…

php websocket 帧封装,swoole websocket封装类和调用

上代码 ws.php/*** ws 优化 基础类库* User: singwa* Date: 18/3/2* Time: 上午12:34*/class Ws {CONST HOST "0.0.0.0";CONST PORT 9512;public $ws null;public function __construct() {$this->ws new swoole_websocket_server("0.0.0.0", 9512)…

夸克浏览器怎么安装脚本_广告看烦了?别砸手机!这五款浏览器能拯救你

哈喽大家好&#xff0c;欢迎来到黑马公社。随着各种良莠不齐的内容开始泛滥&#xff0c;黑马发现自己很难通过网络第一时间找到自己想要的内容。在电脑上&#xff0c;黑马为自己的每个浏览器都安装了不下三个广告屏蔽插件&#xff0c;而在手机上&#xff0c;很难。先不说手机浏…

php 今天 明天 后天 显示10天,【微信小程序】实现含有今天,明天,后天的日期组件...

封面图.JPG前言做过微信小程序的前端er都知道&#xff0c;小程序有个日期组件&#xff0c;叫picker&#xff0c;但是&#xff0c;需求方要求日期和时间都要显示的&#xff0c;用picker组件的话&#xff0c;那就用到两个picker&#xff0c;date和time&#xff0c;就是说要让用户…

python dlib 年龄 性别_python dlib学习(一):人脸检测

1、环境安装Windows&#xff1a; 旧版本安装pip install xxx.whl。以下是whl文件地址&#xff1a; ​ Python Package Index​ 最新版本安装&#xff1a;不要嫌麻烦&#xff0c;先装上visual studio2015 (C模块)。​ 具体的记不清了&#xff0c;装上cmake和boost&#xff0c;然…

php数组实例,php常用数组函数实例小结

本文实例总结了php常用数组函数。分享给大家供大家参考&#xff0c;具体如下&#xff1a;1. array array_merge(array $array1 [, array $array2 [, $array]])函数功能&#xff1a;将一个或多个数组的单元合并起来&#xff0c;一个数组中的值附加在前一个数组的后面。返回结果的…

python预测控制_无人驾驶——4.控制之MPC模型预测控制

源自&#xff1a;《无人驾驶无人驾驶车辆模型预测控制》——龚建伟参考&#xff1a;https://wenku.baidu.com/view/8e4633d519e8b8f67c1cb9fa.html0.车辆模型汽车的车轮转角为δf分别做垂直于后轮和前轮的射线&#xff0c;这两根射线会交于O点&#xff0c;两轮模型会绕O点进行运…

php实现git服务器,如何搭建和配置Git服务器

本篇文章给大家带来的内容是介绍如何搭建和配置Git服务器&#xff0c;有一定的参考价值&#xff0c;有需要的朋友可以参考一下&#xff0c;希望对你有所帮助。安装Git安装依赖yum install curl-devel expat-devel gcc gcc-cyum install perl perl-devel gettext-devel openssl-…

手机连接投影机的步骤_投影机安装过程详解

投影机安装过程详解一 投影机的安装方式1、桌面摆放桌面投影虽然看起来不是很美观&#xff0c;但可以省去那些繁琐的步骤&#xff0c;只需要准备一张桌子&#xff0c;还可以购买一些专门用来摆放投影机的可移动小车架&#xff0c;把投影机往上一放&#xff0c;连接上线缆就可以…

php memcached close,PHP连接Memcached安装及数据库操作

memcached介绍Memcached是一套开源的高性能分布式内存对象缓存系统,它将所有的数据都存储在内存中,因为在内存中会统一维护一张巨大的Hash表,所以支持任意存储类型的数据。很多网站通过使用 Memcached提高网站的访问速度,尤其是对于大型的需要频繁访问数据的网站。Memcached是典…

坏道修复是不是硬盘东西全部都没有了_硬盘有坏道就不能用了吗?别再吃哑巴亏了,今天跟大家再说一次...

硬盘是电脑的存储硬件&#xff0c;是电脑中核心的硬件之一&#xff0c;目前市场上主要使用的是固态硬盘与机械硬盘两种&#xff0c;固态硬盘的读写速度较快&#xff0c;容量小&#xff0c;价格贵&#xff0c;机械硬盘读写速度慢&#xff0c;容量大价格便宜&#xff0c;现在的电…

html5+php调用android手机图片,html5+exif.js+canvas+php实现手机上传图片,图片损坏无法打开...

上传图片&#xff0c;图片损坏无法打开&#xff0c;图片路径也是正确的&#xff0c;function selectFileImage(fileObj) {var file fileObj.files[0];//图片方向角 added by lzkvar Orientation null;if (file) {console.log("正在上传,请稍后...");var rFilter /…

word 编辑域中的汉字_15条Word常用操作教程,简单实用,纯干货分享,收藏备用!...

点击蓝字关注我们1. 去除页眉横线在页眉插入信息的时候经常会在下面出现一条横线&#xff0c;如果这条横线影响你的视觉。这时你可以采用下述的两种方法去掉&#xff1a;用第一种的朋友比较多&#xff0c;即选中页眉的内容后&#xff0c;选取“格式”选项&#xff0c;选取“边框…

matlab里面连接器是什么,连接器

连接器连接器&#xff0c;即CONNECTOR。国内亦称作接插件、插头和插座。一般是指电连接器。即连接两个有源器件的器件&#xff0c;传输电流或信号。它广泛应用于航空、航天、国防等军用系统中。连接器是我们电子工程技术人员经常接触的一种部件。它的作用非常单纯&#xff1a;在…

安卓抓包软件_Packet Capture安卓抓包神器介绍及使用教程

除了干货&#xff0c;其他什么也没有源码&#xff5c;资源&#xff5c;软件&#xff5c;教程&#xff5c;揭秘关 注Packet Capture是一款安卓抓包软件&#xff0c;能用来提取用户操作程序内容&#xff0c;Packet Capture可以捕获网络数据包&#xff0c;并记录它们使用中间人技术…

队列处理高并发_高并发场景下缓存处理的一些思路

在实际的开发当中&#xff0c;我们经常需要进行磁盘数据的读取和搜索&#xff0c;因此经常会有出现从数据库读取数据的场景出现。但是当数据访问量次数增大的时候&#xff0c;过多的磁盘读取可能会最终成为整个系统的性能瓶颈&#xff0c;甚至是压垮整个数据库&#xff0c;导致…