当用自定义编码器实例化kafka生产者时，不能实例化

Question

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

首先是我的自定义编码器，解码器也是这样（对于我的消息类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;
}}

我不确定这是否是实例化生产者/消费者的适当方法。但这种方式行不通。因为我的kafkaProducer无法实例化。

一些调试信息：

Answer 1

您应该实现org.apache.kafka.common.serialization.serializer而不是encoder。

请参阅CustomSerializer示例。