在1個topic中，有3個partition，那么如何保證數(shù)據(jù)的順序消費？

生產(chǎn)者在寫的時候，可以指定一個 key，被分發(fā)到同一個 partition 中去，而且這個 partition 中的數(shù)據(jù)一定是有順序的。

消費者從 partition 中取出來數(shù)據(jù)的時候，也一定是有順序的。到這里，順序還是沒有錯亂的。

但是消費者里可能會有多個線程來并發(fā)處理消息，而多個線程并發(fā)處理的話，順序可能就亂掉了。

解決方案

寫?n 個 queue，將具有相同key的數(shù)據(jù)都存儲在同一個 queue，然后對于 n 個線程，每個線程分別消費一個 queue 即可，并手動提交位點。由于 kafka consumer 實例不支持多線程同時提交位點，這里采取全局記數(shù)器的方式，在每一批次記錄的消費過程中，每消費完一條記錄則全局記數(shù)器加 1，全局記數(shù)器等于這一批記錄的總條數(shù)時提交位點。

在Java中，可以使用多線程和隊列來實現(xiàn)對具有相同 key 的數(shù)據(jù)進(jìn)行消費，并通過手動提交位點來保證數(shù)據(jù)的消費。以下是一個帶有手動位點提交的解決方案的示例代碼：

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class DataConsumer {
    private Map<String, BlockingQueue<String>> queues;
    private Map<String, Integer> offsets;

    public DataConsumer(int numThreads) {
        queues = new HashMap<>();
        offsets = new HashMap<>();

        // 創(chuàng)建N個隊列和位點
        for (int i = 0; i < numThreads; i++) {
            BlockingQueue<String> queue = new LinkedBlockingQueue<>();
            String key = Integer.toString(i);
            queues.put(key, queue);
            offsets.put(key, 0);

            // 創(chuàng)建并啟動消費線程
            Thread consumerThread = new Thread(new Consumer(queue, key));
            consumerThread.start();
        }
    }

    public void consumeData(String key, String data) {
        BlockingQueue<String> queue = queues.get(key);
        if (queue != null) {
            try {
                // 將數(shù)據(jù)放入對應(yīng)的隊列
                queue.put(data);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void commitOffset(String key, int offset) {
        offsets.put(key, offset);
        System.out.println("Committed offset for key " + key + ": " + offset);
    }

    private static class Consumer implements Runnable {
        private final BlockingQueue<String> queue;
        private final String key;
        private int offset;

        public Consumer(BlockingQueue<String> queue, String key) {
            this.queue = queue;
            this.key = key;
            this.offset = 0;
        }

        @Override
        public void run() {
            // 消費隊列中的數(shù)據(jù)
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    String data = queue.take();
                    // 進(jìn)行消費邏輯
                    System.out.println("Consumed data: " + data);
                    offset++;

                    // 模擬提交位點
                    if (offset % 10 == 0) {
                        DataConsumer.getInstance().commitOffset(key, offset);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    private static DataConsumer instance;

    public static synchronized DataConsumer getInstance() {
        if (instance == null) {
            instance = new DataConsumer(3);
        }
        return instance;
    }

    public static void main(String[] args) {
        DataConsumer dataConsumer = DataConsumer.getInstance();

        // 模擬產(chǎn)生數(shù)據(jù)
        for (int i = 0; i < 30; i++) {
            dataConsumer.consumeData(Integer.toString(i % 3), "Data " + (i + 1));
        }
    }
}

在以上代碼中，DataConsumer 類維護(hù)了一個 Map 來存儲隊列和位點的關(guān)系。每個消費者線程都有一個對應(yīng)的位點來記錄消費的進(jìn)度。

在 commitOffset 方法中，根據(jù) key 提交位點的偏移值。

消費線程在每次成功消費一條數(shù)據(jù)后，更新位點，并判斷是否滿足提交位點的條件。這里模擬每消費10條數(shù)據(jù)提交一次位點。

在 main 方法中，通過 consumeData 方法模擬了產(chǎn)生了30條數(shù)據(jù)，并將它們放入不同的隊列中進(jìn)行消費。文章來源地址http://www.zghlxwxcb.cn/news/detail-729626.html

到了這里，關(guān)于實現(xiàn) Kafka 分區(qū)內(nèi)消費者多線程順序消費的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！