配置kafka发送自定义类型数据

Question

我对卡夫卡很陌生

我已经使用官方示例创建了生产者和消费者组，尽管我想从生产者和消费者发送节俭包以获取包并存储在包数组中。

我已将生产者端代码编写为

KeyedMessage<String, Bundle> data = new KeyedMessage<String, Bundle>("bundles", "Bundle", bundle); 
        producer.send(data);

但在消费者方面我有

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, new Integer(NO_OF_THREADS));
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<KafkaStream<byte[], byte[]>> messageStreams = consumerMap.get(topic);

我可以让 kafka 消费者以捆绑类型而不是 byte[] 数组的形式获取数据吗？

Answer 1

您可以使用以下方法将值直接解码为Bundle类型：

public interface kafka.javaapi.consumer.ConsumerConnector {
  ...
  public <K,V> Map<String, List<KafkaStream<K,V>>> 
     createMessageStreams(
         Map<String, Integer> topicCountMap, Decoder<K> keyDecoder, Decoder<V> valueDecoder);

在这种情况下，您需要使用相应的 Thrift API 实现您自己的 Decoder<Bundle> 类型的 valueDecoder。

请参阅 Kafka 文档中的 High Level Consumer API 描述。