Kafka Producer Java API 没有将消息分发到所有主题分区

Question

我是 Kafka 的新手，今天我尝试创建 Java 生产者，用于在不同分区上生成关于 Kafka 主题的消息。

首先，我创建了一个包 raggieKafka，在该包下我创建了 2 个 classes：TestProducer 和 SimplePartitioner。

TestProducer class 具有以下代码：

package raggieKafka;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer{

    public static void main(String args[]) throws Exception
    {
        long events = 0;

        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
        events = Integer.parseInt(reader.readLine());
        Random rnd = new Random();

        Properties props = new Properties();
        props.put("metadata.broker.list", "localhost:9092");
        props.put("topic.metadata.refresh.interval.ms", "1");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "raggieKafka.SimplePartitioner");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> prod = new Producer<String, String>(config);

        for(long i = 0; i < events; i++)
        {
            long runtime = new Date().getTime();
            String ip = "192.168.2." + rnd.nextInt(255);
            String msg = runtime + ",www.example.com, " + ip;
            KeyedMessage<String,String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
            prod.send(data);
        }
        prod.close();
    }
}

和SimplePartitioner class有以下代码：

package raggieKafka;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

public class SimplePartitioner implements Partitioner{

    public SimplePartitioner(VerifiableProperties props)
    {

    }

    public int partition(Object Key, int a_numPartitions)
    {
        int partition = 0;
        String stringKey = (String) Key;
        int offset = stringKey.indexOf(stringKey);

        if(offset > 0)
        {
            partition = Integer.parseInt(stringKey.substring(offset+1)) % a_numPartitions;
        }
        return partition;
    }   
}

在编译这些 Java 程序之前，我在 Kafka Broker 上创建了主题：

C:\kafka_2.11-0.9.0.1>.\bin\windows\kafka-topics.bat --create --topic page_visit
s --zookeeper localhost:2181 --partitions 5 --replication-factor 1
WARNING: Due to limitations in metric names, topics with a period ('.') or under
score ('_') could collide. To avoid issues it is best to use either, but not bot
h.
Created topic "page_visits".

现在，当我编译 java 程序时，它会将所有消息仅放入 1 个分区，即 page_visits-0，所有消息都发布在该分区下，但其余所有分区仍为空。

有人能告诉我为什么我的 Java 生产者没有将我的所有消息分发到其他分区吗？

其实我看了google然后又加了一个属性:

props.put("topic.metadata.refresh.interval.ms", "1");

但 Producer 仍然没有为所有主题生成消息。

请帮忙。

Answer 1

您的 SimplePartitioner 代码在以下行中存在错误

int offset = stringKey.indexOf(stringKey);

它总是 returns 0 所以你的偏移量总是等于 0 并且因为它永远不会大于 0 你的 if 块将不会被执行。最后它总是 returns 你分区 0.

解决方案： 由于您的密钥是 ip 地址，因此以下更改可以按预期工作。

int offset = stringKey.lastIndexOf('.');

希望对您有所帮助！

Kafka Producer Java API 没有将消息分发到所有主题分区

Kafka Producer Java API is not distributing messages to all topic partitions

apache-kafka

kafka-producer-api