producer.send 不接受 KeyedMessage 类型答案

【问题标题】：producer.send doesn't accept KeyedMessage typeproducer.send 不接受 KeyedMessage 类型
【发布时间】：2019-11-17 23:58:33
【问题描述】：

我尝试运行此代码，但由于 producer.send() 不接受 KeyedMessage 类型，它不起作用。

我尝试导入 kafka.javaapi.producer.Producer 而不是 kafka.producer.Producer；但还是不行

代码是：

package sources;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;

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

public class ProducerCode {

    private static Producer<Integer, String> producer;
    private static final String topic= "mytopic";

    public void initialize() {
        Properties producerProps = new Properties();
        producerProps.put("metadata.broker.list", "localhost:9092");
        producerProps.put("serializer.class", "kafka.serializer.StringEncoder");
        producerProps.put("request.required.acks", "1");
        // ProducerConfig producerConfig = new ProducerConfig(producerProps);
        // have a change here **
        producer = new Producer<Integer, String>(new ProducerConfig(producerProps));
    }

    public void publishMesssage() throws Exception{            
        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));               
        while (true){
            System.out.print("Enter message to send to kafka broker (Press 'Y' to close producer): ");
            String msg = null;
            msg = reader.readLine(); // Read message from console
            //Define topic name and message
            KeyedMessage<Integer, String> keyedMsg = new KeyedMessage<Integer, String>(topic, msg);

            producer.send(keyedMsg);
            // producer.send(keyedMsg); // This publishes message on given topic

            if("Y".equals(msg)){ break; }
            System.out.println("--> Message [" + msg + "] sent.Check message on Consumer's program console");
         }
         return;
    }

    public static void main(String[] args) throws Exception  {

        KafkaProducer kafkaProducer = new KafkaProducer();
        // Initialize producer
        kafkaProducer.initialize();            
        // Publish message
        kafkaProducer.publishMesssage();
        //Close the producer
        producer.close();
    }
}

【问题讨论】：

你用的是什么版本的Kafka？
我使用的是版本 10 ( kafka-0.10.0.0 )

标签： java apache-kafka producer

【解决方案1】：

你必须使用ProducerRecord（而不是KeyedMessage）和构造函数ProducerRecord(String topic, K key, V value)

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<String, String>("my-topic", "key", "value"));

见https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

【讨论】：