簡介： 本人通過分析痛點、同步批量請求優化為異步請求、多線程 Client 模式、Flink 算子內多線程實現以及總結四部分幫助大家理解 Flink 中使用多線程的優化及在 Flink 算子中使用多線程如何保證不丟數據。

分析痛點

筆者線上有一個 Flink 任務消費 Kafka 數據，將數據轉換后，在 Flink 的 Sink 算子內部調用第三方 api 將數據上報到第三方的數據分析平臺。這里使用批量同步 api，即：每 50 條數據請求一次第三方接口，可以通過批量 api 來提高請求效率。由于調用的外網接口，所以每次調用 api 比較耗時。假如批次大小為 50，且請求接口的平均響應時間為 50ms，使用同步 api，因此第一次請求響應以后才會發起第二次請求。請求示意圖如下所示：

平均下來，每 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 呢？

同步批量請求優化為異步請求

首先可以想到的是將同步請求改為異步請求，使得任務不會阻塞在網絡請求這一環節，請求示意圖如下所示。

異步請求相比同步請求而言，優化點在于每次發出請求時，不需要等待請求響應后再發送下一次請求，而是當下一批次的 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 倍。

說明：多線程的方案，不僅限于請求第三方接口，對于非 CPU 密集型的任務也可以使用該方案，在降低 CPU 數量的同時，單個 CPU 承擔多個線程的工作，從而提高 CPU 利用率。例如：請求 HBase 的任務或磁盤 IO 是瓶頸的任務，可以降低任務的并行度，使得每個并行度內處理多個線程。

Flink 算子內多線程實現

Sink 算子的單個并行度內現在有 5 個 Client 用于消費數據，但 Sink 算子的數據都來自于上游算子。如下圖所示，一個簡單的實現方式是 Sink 算子接收到上游數據后通過輪循或隨機的策略將數據分發給 5 個 Client 線程。

但是輪循或者隨機策略會存在問題，假如 5 個 Client 中 Client3 線程消費較慢，會導致給 Client3 分發數據時被阻塞，從而使得其他正常消費的線程 Client1、2、4、5 也被分發不到數據。

為了解決上述問題，可以在 Sink 算子內申請一個數據緩沖隊列，隊列有先進先出（FIFO）的特性。Sink 算子接收到的數據直接插入到隊列尾部，五個 Client 線程不斷地從隊首取數據并消費，即：Sink 算子先接收的數據 Client 先消費，后接收的數據 Client 后消費。

• 若隊列一直是滿的，說明 Client 線程消費較慢、Sink 算子上游生產數據較快。
• 若隊列一直為空，說明 Client 線程消費較快、Sink 算子的上游生產數據較慢。

五個線程共用同一個隊列完美地解決了單個線程消費慢的問題，當 Client3 線程阻塞時，不影響其他線程從隊列中消費數據。這里使用隊列還起到了削峰填谷的作用。

代碼實現

原理明白了，具體代碼如下所示，首先是消費數據的 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 之前的一小部分緩存在內存緩沖隊列中的數據不會再被消費，于是就出現了丟數據情況。

處理丟數據情況

如何保證數據不丟失呢？很簡單，可以在 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 算子中使用多線程如何保證不丟數據。

總結

以上是生活随笔為你收集整理的线程中如何使用对象_在 Flink 算子中使用多线程如何保证不丢数据？的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。