【原創】Kafka Consumer多線程實例

　　Kafka 0.9版本開始推出了Java版本的consumer，優化了coordinator的設計以及擺脫了對zookeeper的依賴。社區最近也在探討正式用這套consumer API替換Scala版本的consumer的計划。鑒於目前這方面的資料並不是很多，本文將嘗試給出一個利用KafkaConsumer編寫的多線程消費者實例，希望對大家有所幫助。

    這套API最重要的入口就是KafkaConsumer(o.a.k.clients.consumer.KafkaConsumer)，普通的單線程使用方法官網API已有介紹，這里不再贅述了。因此，我們直奔主題——討論一下如何創建多線程的方式來使用KafkaConsumer。KafkaConsumer和KafkaProducer不同，后者是線程安全的，因此我們鼓勵用戶在多個線程中共享一個KafkaProducer實例，這樣通常都要比每個線程維護一個KafkaProducer實例效率要高。但對於KafkaConsumer而言，它不是線程安全的，所以實現多線程時通常由兩種實現方法：

1 每個線程維護一個KafkaConsumer

2  維護一個或多個KafkaConsumer，同時維護多個事件處理線程(worker thread)

當然，這種方法還可以有多個變種：比如每個worker線程有自己的處理隊列。consumer根據某種規則或邏輯將消息放入不同的隊列。不過總體思想還是相同的，故這里不做過多展開討論了。

　　下表總結了兩種方法的優缺點： 

	優點	缺點
方法1(每個線程維護一個KafkaConsumer)	方便實現 速度較快，因為不需要任何線程間交互 易於維護分區內的消息順序	更多的TCP連接開銷(每個線程都要維護若干個TCP連接) consumer數受限於topic分區數，擴展性差 頻繁請求導致吞吐量下降 線程自己處理消費到的消息可能會導致超時，從而造成rebalance
方法2 (單個(或多個)consumer，多個worker線程)	可獨立擴展consumer數和worker數，伸縮性好	實現麻煩 通常難於維護分區內的消息順序 處理鏈路變長，導致難以保證提交位移的語義正確性

 

下面我們分別實現這兩種方法。需要指出的是，下面的代碼都是最基本的實現，並沒有考慮很多編程細節，比如如何處理錯誤等。

方法1

ConsumerRunnable類

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class ConsumerRunnable implements Runnable {

    // 每個線程維護私有的KafkaConsumer實例
    private final KafkaConsumer<String, String> consumer;

    public ConsumerRunnable(String brokerList, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");        //本例使用自動提交位移
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        this.consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));   // 本例使用分區副本自動分配策略
    }

    @Override
    public void run() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(200);   // 本例使用200ms作為獲取超時時間
            for (ConsumerRecord<String, String> record : records) {
                // 這里面寫處理消息的邏輯，本例中只是簡單地打印消息
                System.out.println(Thread.currentThread().getName() + " consumed " + record.partition() +
                        "th message with offset: " + record.offset());
            }
        }
    }
}

ConsumerGroup類

package com.my.kafka.test;

import java.util.ArrayList;
import java.util.List;

public class ConsumerGroup {

    private List<ConsumerRunnable> consumers;

    public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
        consumers = new ArrayList<>(consumerNum);
        for (int i = 0; i < consumerNum; ++i) {
            ConsumerRunnable consumerThread = new ConsumerRunnable(brokerList, groupId, topic);
            consumers.add(consumerThread);
        }
    }

    public void execute() {
        for (ConsumerRunnable task : consumers) {
            new Thread(task).start();
        }
    }
}

ConsumerMain類

public class ConsumerMain {

    public static void main(String[] args) {
        String brokerList = "localhost:9092";
        String groupId = "testGroup1";
        String topic = "test-topic";
        int consumerNum = 3;

        ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
        consumerGroup.execute();
    }
}

 

方法2

Worker類

import org.apache.kafka.clients.consumer.ConsumerRecord;

public class Worker implements Runnable {

    private ConsumerRecord<String, String> consumerRecord;

    public Worker(ConsumerRecord record) {
        this.consumerRecord = record;
    }

    @Override
    public void run() {
        // 這里寫你的消息處理邏輯，本例中只是簡單地打印消息
        System.out.println(Thread.currentThread().getName() + " consumed " + consumerRecord.partition()
            + "th message with offset: " + consumerRecord.offset());
    }
}

ConsumerHandler類

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ConsumerHandler {

    // 本例中使用一個consumer將消息放入后端隊列，你當然可以使用前一種方法中的多實例按照某張規則同時把消息放入后端隊列
    private final KafkaConsumer<String, String> consumer;
    private ExecutorService executors;

    public ConsumerHandler(String brokerList, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
    }

    public void execute(int workerNum) {
        executors = new ThreadPoolExecutor(workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(200);
            for (final ConsumerRecord record : records) {
                executors.submit(new Worker(record));
            }
        }
    }

    public void shutdown() {
        if (consumer != null) {
            consumer.close();
        }
        if (executors != null) {
            executors.shutdown();
        }
        try {
            if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("Timeout.... Ignore for this case");
            }
        } catch (InterruptedException ignored) {
            System.out.println("Other thread interrupted this shutdown, ignore for this case.");
            Thread.currentThread().interrupt();
        }
    }

}

Main類

public class Main {

    public static void main(String[] args) {
        String brokerList = "localhost:9092,localhost:9093,localhost:9094";
        String groupId = "group2";
        String topic = "test-topic";
        int workerNum = 5;

        ConsumerHandler consumers = new ConsumerHandler(brokerList, groupId, topic);
        consumers.execute(workerNum);
        try {
            Thread.sleep(1000000);
        } catch (InterruptedException ignored) {}
        consumers.shutdown();
    }
}

 

　　總結一下，這兩種方法或是模型都有各自的優缺點，在具體使用時需要根據自己實際的業務特點來選取對應的方法。就我個人而言，我比較推崇第二種方法以及背后的思想，即不要將很重的處理邏輯放入消費者的代碼中，很多Kafka consumer使用者碰到的各種rebalance超時、coordinator重新選舉、心跳無法維持等問題都來源於此。