使用自定义编码器实例化kafka producer时，无法实例化

Question

我是卡夫卡的新手。我想在我的 spring 项目中使用 kafka producer/consumer 替换 activeMQ(jms)。我需要的是一个 kafka 生产者将我的消息对象发布到一个主题，一个消费者从该主题订阅它。

首先是我的自定义编码器，解码器也是如此（对于我的消息 class ConfigurationActionMsg）：

@Component
public class ConfigActionMessageEncoder implements Encoder<ConfigurationActionMsg> {

    public ConfigActionMessageEncoder() {
    /* This constructor must be present for successful compile. */
    }
    public ConfigActionMessageEncoder(VerifiableProperties verifiableProperties) {
    /* This constructor must be present for successful compile. */
    }

    @Override
    public byte[] toBytes(ConfigurationActionMsg actionMsg){
        return SerializationUtils.serialize(actionMsg);
    }}

下面是我的处理器和消费者配置

@Configuration
@ComponentScan(basePackages = {"XXX"})
public class KafkaConfig {
@Bean
public KafkaProducer<String,ConfigurationActionMsg> kafkaProducer(){
    Properties props = new Properties();
    props.put("zk.connect", "127.0.0.1:2181");
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "com.atlas.configengine2.XXX.ConfigActionMessageEncoder");

    return new KafkaProducer<>(props);
}

@Bean
public KafkaConsumer<String, ConfigurationActionMsg> kafkaConsumer(){

    Properties props = new Properties();
    props.put("zk.connect", "127.0.0.1:2181");
    props.put("bootstrap.servers", "localhost:9092");
    //We should only have one process running for consumer
    props.put("group.id", "resolverActionTrigger");
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "com.atlas.configengine2.XXX.ConfigActionMessageDecoder");
    KafkaConsumer<String, ConfigurationActionMsg> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList("configAction"));
    return consumer;
}}

我不确定这是否是实例化 producer/consumer 的正确方法。但这种方式行不通。因为我的kafkaProducer无法实例化。

一些调试信息：

Caused by: org.apache.kafka.common.KafkaException: com.atlas.configengine2.jms.ConfigActionMessageEncoder is not an instance of org.apache.kafka.common.serialization.Serializer

但我不确定这是否是唯一的问题？那么自定义编码器怎么写呢？

Answer 1

您应该实施 org.apache.kafka.common.serialization.Serializer 而不是 Encoder。

参见 CustomSerializer 示例。

Answer 2

首先，你应该使用org.apache.kafka.common.serialization.Serializer的子类。

其次，我认为您可以发送对象的字符串表示形式，例如 JSON，并在生产者中跟上 StringSerializer 以避免实现自定义 org.apache.kafka.common.serialization.Serializer 的开销].

Answer 3

 //create properties
        //https://kafka.apache.org/documentation/#producerconfigs
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9094,127.0.0.1:9098");
        prop.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prop.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Employee.class.getName());

        //Create Producer
        //For String
        //KafkaProducer<String, String> producer= new KafkaProducer(prop);
        //For Object
        KafkaProducer<String, Employee> producer = new KafkaProducer(prop);
        Address add = new Address("India");
        Employee emp = new Employee(1, "Arun", add);
        //For String
    // ProducerRecord prodRecord = new ProducerRecord("aryan_topic", "Any String Value");
        ProducerRecord prodRecord = new ProducerRecord("aryan_topic", emp);
    //Send Record

        producer.send(prodRecord);
        System.out.println("Data Send");

}
}
import org.apache.kafka.common.header.Headers;

导入org.apache.kafka.common.serialization.Serializer；

进口java.io.Serializable；导入 java.util.Map;

//由 Arun Singh 开发

public class Employee implements Serializable, Serializer {
//add all unImplemented methods here
}
 and it will work as
//
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.4.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 77a89fcf8d7fa018
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1583515298948
[kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.Metadata - [Producer clientId=producer-1] Cluster ID: h_EDcwYmThiz-iBTc0AsGw
Data Send

使用自定义编码器实例化kafka producer时，无法实例化

when instantiate kafka producer with custom encoder, can't instantiate

apache-kafka

kafka-producer-api