Kafka Java 客戶端完整實戰：生產者/消費者/分區策略與集群配置詳解

前言：LinkedIn 打造的高吞吐分散式日誌系統

Kafka 犧牲強一致性換取極致吞吐，支援離線/線上雙模式消費。本文基於 kafka_2.10-0.8.1.1（3 Broker + 1 ZooKeeper）給出生產者、消費者、自訂分區器完整可運行代碼。

環境依賴

    org.apache.kafka
    kafka_2.10
    0.8.1.1

生產者 ProducerDemo.java

public class ProducerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("metadata.broker.list", "172.168.63.221:9092,172.168.63.233:9092,172.168.63.234:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "com.catt.kafka.demo.PartitionerDemo"); // 可選
        props.put("request.required.acks", "1"); // 0=fire-forget, 1=leader ack, -1=all replica ack

        ProducerConfig config = new ProducerConfig(props);
        Producer producer = new Producer<>(config);

        long start = System.currentTimeMillis();
        for (long i = 0; i < 100; i++) {
            String ip = "192.168.2." + i;
            String msg = System.currentTimeMillis() + ",www.example.com," + ip;
            KeyedMessage data = new KeyedMessage<>("page_visits", ip, msg);
            producer.send(data);
        }
        System.out.println("耗時: " + (System.currentTimeMillis() - start) + "ms");
        producer.close();
    }
}

消費者 ConsumerDemo.java

public class ConsumerDemo {
    private final ConsumerConnector consumer;
    private final String topic;
    private ExecutorService executor;

    public ConsumerDemo(String zk, String groupId, String topic) {
        this.consumer = Consumer.createJavaConsumerConnector(createConsumerConfig(zk, groupId));
        this.topic = topic;
    }

    private static ConsumerConfig createConsumerConfig(String zk, String groupId) {
        Properties props = new Properties();
        props.put("zookeeper.connect", zk);
        props.put("group.id", groupId);
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        return new ConsumerConfig(props);
    }

    public void run(int numThreads) {
        Map topicCount = new HashMap<>();
        topicCount.put(topic, numThreads);
        Map>> streamsMap = consumer.createMessageStreams(topicCount);
        List> streams = streamsMap.get(topic);

        executor = Executors.newFixedThreadPool(numThreads);
        int threadNum = 0;
        for (final KafkaStream stream : streams) {
            executor.submit(new ConsumerMsgTask(stream, threadNum++));
        }
    }

    public void shutdown() {
        if (consumer != null) consumer.shutdown();
        if (executor != null) executor.shutdown();
    }

    public static void main(String[] args) {
        // args: zk, groupId, topic, threads
        ConsumerDemo demo = new ConsumerDemo(args[0], args[1], args[2]);
        demo.run(Integer.parseInt(args[3]));
        try { Thread.sleep(10000); } catch (InterruptedException ignored) {}
        demo.shutdown();
    }
}

訊息處理任務 ConsumerMsgTask.java

public class ConsumerMsgTask implements Runnable {
    private final KafkaStream stream;
    private final int threadNumber;

    public ConsumerMsgTask(KafkaStream stream, int threadNumber) {
        this.stream = stream; this.threadNumber = threadNumber;
    }

    @Override public void run() {
        ConsumerIterator it = stream.iterator();
        while (it.hasNext()) {
            System.out.println("Thread " + threadNumber + ": " + new String(it.next().message()));
        }
        System.out.println("Shutting down Thread: " + threadNumber);
    }
}

自訂分區器 PartitionerDemo.java

public class PartitionerDemo implements Partitioner {
    public PartitionerDemo(VerifiableProperties props) {}

    @Override public int partition(Object key, int numPartitions) {
        int partition = 0;
        if (key instanceof String) {
            String k = (String) key;
            int dot = k.lastIndexOf('.');
            if (dot > 0) partition = Integer.parseInt(k.substring(dot + 1)) % numPartitions;
        } else {
            partition = key.toString().length() % numPartitions;
        }
        return partition;
    }
}

關鍵配置解析

遷移提醒：新版 Kafka Client (0.10+)

• 套件改為 org.apache.kafka:kafka-clients
• 生產者用 KafkaProducer、ProducerRecord
• 消費者用 KafkaConsumer、subscribe()、poll() 迴圈
• 不再依賴 ZooKeeper 管理 Offset，改存在 __consumer_offsets Topic