Apache Kafka Java生产商Scala消费者缺少流

Question

我用的是阿帕奇·Kafka。我创建了一个war文件，其中生产者用Java编码，消费者用Scala编码。制作人正在从HTML页面获取数据。我可以看到，生产商发布的大部分数据都是关于消费者的，但有些数据缺失。

这是我的制片人代码

文件1

package com.cts.rest;

import java.util.Properties;

import kafka.producer.ProducerConfig;

public class Configuration {

static ProducerConfig setKafkaProducerParameter() {
    Properties properties = new Properties();
    properties.put("zk.connect", "localhost:2181");
    properties.put("metadata.broker.list", "localhost:9092");
    properties.put("serializer.class", "kafka.serializer.StringEncoder");
    properties.put("acks", 0);  
    ProducerConfig producerConfig = new ProducerConfig(properties);
    return producerConfig;
    }

}

文件2

package com.cts.rest;

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


public class RTTSKProducer {

static void sendDataToProducer(String line){

    ProducerConfig producerConfig = configuration.setKafkaProducerParameter();
    Producer<String, String> producer = new Producer<String, String>(producerConfig);       

    String topic = "jsondata";      
    KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, line);
    System.out.print(msg);
    producer.send(msg);
    producer.close();
            }
    }

现在，我使用以下命令检查消费者的消息。

bin\windows\kafka-console-consumer.bat --zookeeper localhost:2181 --topic jsondata --from-beginning

我是否缺少任何生产者配置？

Answer 1

您可以尝试增加ack配置以确保更耐用。最重要的是，您应该调用带有回调函数的“发送”方法来处理那些没有成功发布到Kafka的消息，如下所示：

producer.send(myRecord,
           new Callback() {
               public void onCompletion(RecordMetadata metadata, Exception e) {
                   if(e != null)
                       e.printStackTrace();
                   System.out.println("The offset of the record we just sent is: " + metadata.offset());
               }
           });