Kafka 生产者发送无效字符

Question

使用以下代码，我发送 Elasticsearch 文档进行索引。我尝试将基本对象转换为 JSON 并通过生产者发送。但是，每条消息（从控制台检查）都会附加一些乱码，例如 - ��t�{"productId":2455

public boolean sendMessage()
{
    PageRequest page = new PageRequest(0, 1); 
    Product p = product.findByName("Cream", page).getContent().get(0);
    String json = "";
    ObjectMapper mapper = new ObjectMapper();
    try {
        json = mapper.writeValueAsString(p);
    } catch (JsonProcessingException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }       
    logger.info("JSON = " + json);

    boolean status =  inputToKafka.send(org.springframework.integration.support.MessageBuilder.withPayload(json).build());
    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return status;
}

出站配置

 <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
       xmlns:task="http://www.springframework.org/schema/task"
       xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">

    <int:channel id="inputToKafka">
        <int:queue/>
    </int:channel>

    <int-kafka:outbound-channel-adapter
            id="kafkaOutboundChannelAdapter"
            kafka-producer-context-ref="kafkaProducerContext"
            channel="inputToKafka">
        <int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
    </int-kafka:outbound-channel-adapter>

    <task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="500"/>

    <int-kafka:producer-context id="kafkaProducerContext">
        <int-kafka:producer-configurations>
            <int-kafka:producer-configuration broker-list="localhost:9092"
                                              topic="test_topic"
                                              compression-codec="default"/>
        </int-kafka:producer-configurations>
    </int-kafka:producer-context>

    </beans>

有什么线索吗？

使用的插件：Spring Extension Kafka

Answer 1

我今天遇到了这个问题，并通过在生产者配置中设置正确的 value-serializer 类来解决它，如下所示：

<int-kafka:producer-configuration
                broker-list="localhost:9092" topic="headers['kafka_topic']"
                key-class-type="java.lang.String" value-class-type="java.lang.String"
                key-serializer="kafkaSerializer" value-serializer="kafkaSerializer"/>

<bean id="kafkaSerializer" class="org.apache.kafka.common.serialization.StringSerializer" />

Answer 2

卡夫卡不做这样的事情。在您将其发送给 Kafka 生产者的字符串消息处进行调试。 如果您从 URL 或 HTML 表单获取此消息，您可能需要先解码您的消息，然后再发送给生产者。

例如URLDecoder.decode(message, "UTF-8")

Answer 3

这些可能是您的控制台不能很好地解释的制表符（因为缩进的 JSON）。

如果您禁用对象映射器生成的输出的缩进，这些字符可能会消失。

try {
    mapper.disable(SerializationFeature.INDENT_OUTPUT);     <---- add this line
    json = mapper.writeValueAsString(p);
} catch (JsonProcessingException e1) {
    // TODO Auto-generated catch block
    e1.printStackTrace();
}