Apache Kafka Java Producer Scala Consumer 缺少流答案

【问题标题】：Apache Kafka Java Producer Scala Consumer missing streamsApache Kafka Java Producer Scala Consumer 缺少流
【发布时间】：2016-11-03 05:17:11
【问题描述】：

我正在使用 Apache Kafka。我创建了一个war文件，其中生产者用Java编码，消费者用Scala编码。 Producer 正在从 HTML 页面获取数据。可以看到producer发布的大部分数据都在consumer上找到，但也有部分数据丢失。

这是我的生产者代码

文件 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

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

【问题讨论】：

标签： java apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

您可以尝试增加“acks”配置以确保更耐用。最重要的是，您应该调用带有回调函数的 'send' 方法来处理那些未成功发布到 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());
               }
           });

【讨论】：