5

我正在使用 Kafka 0.8 beta，我只是想弄乱发送不同的对象，使用我自己的编码器对它们进行序列化，然后将它们发送到现有的代理配置。现在我正试图让 DefaultEncoder 正常工作。

我有代理和所有设置并为 StringEncoder 工作，但我无法获得任何其他数据类型，包括纯字节 []，以由代理发送和接收。

我的生产者代码是：

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Date;
import java.util.Properties;
import java.util.Random;


public class ProducerTest {
    public static void main(String[] args) {
        long events = 5;
        Random rnd = new Random();
        rnd.setSeed(new Date().getTime());
        Properties props = new Properties();
        props.setProperty("metadata.broker.list", "localhost:9093,localhost:9094");
        props.setProperty("serializer.class", "kafka.serializer.DefaultEncoder");
        props.setProperty("partitioner.class", "example.producer.SimplePartitioner");
        props.setProperty("request.required.acks", "1");
        props.setProperty("producer.type", "async");
        props.setProperty("batch.num.messages", "4");

        ProducerConfig config = new ProducerConfig(props);
        Producer<byte[], byte[]> producer = new Producer<byte[], byte[]>(config);
        for (long nEvents = 0; nEvents < events; nEvents++) {
            byte[] a = "Hello".getBytes();
            byte[] b = "There".getBytes();

            KeyedMessage<byte[], byte[]> data = new KeyedMessage<byte[], byte[]>("page_visits", a, b);
            producer.send(data);
        }
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        producer.close();
    }
}

我使用了与此处给出的示例相同的 SimplePartitioner ，并用字符串替换所有字节数组并将序列化程序更改为 kafka.serializer.StringEncoder 工作得很好。

供参考，SimplePartitioner：

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

public class SimplePartitioner implements Partitioner<String> {
    public SimplePartitioner (VerifiableProperties props) {

    }

    public int partition(String key, int a_numPartitions) {
        int partition = 0;
        int offset = key.lastIndexOf('.');
        if (offset > 0) {
           partition = Integer.parseInt( key.substring(offset+1)) % a_numPartitions;
        }
       return partition;
  }

}

我究竟做错了什么？

4

1 回答 1

6

答案是分区类SimplePartitioner仅适用于字符串。当我尝试异步运行 Producer 时，它会创建一个单独的线程来处理编码和分区，然后再发送到代理。当该线程意识到 SimplePartitioner 仅适用于字符串时遇到了障碍，但由于它是一个单独的线程，因此不会引发异常，因此线程只是退出而没有任何错误行为的迹象。

如果我们将 SimplePartitioner 更改为接受 byte[]，例如：

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

public class SimplePartitioner implements Partitioner<byte[]> {
    public SimplePartitioner (VerifiableProperties props) {

    }

    public int partition(byte[] key, int a_numPartitions) {
        int partition = 0;
        return partition;
    }

}

这现在完美无缺。

于 2013-09-26T22:59:10.900 回答