如何使用spring从kafka入站消息中提取数据答案

【问题标题】：How to extract the data from a kafka inbound message with spring如何使用spring从kafka入站消息中提取数据
【发布时间】：2015-07-28 05:53:10
【问题描述】：

我正在用 spring 构建一个 Kafka 消费者。我的配置似乎很简单。消息被消费并保存在文件中。但是，有效负载很神秘，我无法获取数据（简短的“hello world”消息）。

这是我在访问有效负载时得到的（例如，当我在入站 kafka 和出站文件中设置转换器 bean 时：

{test-topic={0=[[B@713c9d72, [B@7d656f90, [B@26bb8c83, [B@4b959d83 [B@5ed74e8e]}}

我的问题是：如何访问实际的有效负载（“hellow world”字符串）？

我的配置是：

<int:channel id="inputFromKafka">
    <int:queue />
</int:channel>
        <int:poller 
        max-messages-per-poll="5" default = "true" fixed-delay="10" time-unit="MILLISECONDS"/>

<int-kafka:inbound-channel-adapter
    id="kafkaInboundChannelAdapter" kafka-consumer-context-ref="consumerContext"
    auto-startup="true" channel="inputFromKafka">

<int-kafka:consumer-context id="consumerContext"
    consumer-timeout="40000" zookeeper-connect="zookeeperConnect">
    <int-kafka:consumer-configurations>
        <int-kafka:consumer-configuration
            group-id="group12" max-messages="5">
            <int-kafka:topic id="test-topic" streams="1" />
        </int-kafka:consumer-configuration>
    </int-kafka:consumer-configurations>
</int-kafka:consumer-context>
<int-kafka:zookeeper-connect id="zookeeperConnect"
    zk-connect="localhost:2181" zk-connection-timeout="6000"
    zk-session-timeout="6000" zk-sync-time="2000" />


<file:outbound-channel-adapter id="filesOut"
    directory="/tmp/fromKafka">
    </file:outbound-channel-adapter>

【问题讨论】：

标签： spring-integration apache-kafka

【解决方案1】：

你看到的是原始的byte[]。

添加...

<bean id="decoder" 
    class="org.springframework.integration.kafka.serializer.common.StringDecoder" />

和

<int-kafka:consumer-configuration
                     value-decoder="decoder"
        ...

【讨论】：