Kafka-spring 消费者没有阅读消息

Question

我正在发送有关主题的消息，并且可以使用 kafka 控制台和使用 main 方法的 Java 使用者代码读取相同的消息。但是没有从消费者配置 xml 中读取消息。请帮助解决此问题。我正在使用消息 int-kafka:message-driven-channel-adapter。 spring-kafka 版本 1.3.0.RELEASE

         <?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:stream="http://www.springframework.org/schema/integration/stream"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
    http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.0.xsd
    http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-4.0.xsd
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">


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


           <bean id="zkConfiguration"  class="org.springframework.integration.kafka.core.ZookeeperConfiguration">
    <constructor-arg ref="zookeeperConnect" />
</bean>

<bean id="kafkaConnectionFactory" class="org.springframework.integration.kafka.core.DefaultConnectionFactory" >
    <constructor-arg ref="zkConfiguration" />
</bean>

  <bean id="decoder"
class="org.springframework.integration.kafka.serializer.common.StringDecoder" />
<int-kafka:message-driven-channel-adapter id="messageProcessor" 
    channel="inputFromKafka"
    connection-factory="kafkaConnectionFactory"
    queue-size="${queue.size:1024}"
        concurrency="${concurrency:50}"
    auto-startup="true"
    payload-decoder="decoder"
    topics="${topics:nishant}" key-decoder="decoder" />
     <int-kafka:zookeeper-connect id="zookeeperConnect" 
  zk-connect="localhost:2181" zk-connection-timeout="6000"
    zk-session-timeout="6000" zk-sync-time="2000" />


         </beans>

消费者 Java 代码

              @Component(value = "messageProcessor")
              public class MessageConsumer {
            public void reader(Map<String, Map<Integer, String>> payload) {
            System.out.println("TEST");

             }
        }

Answer 1

您的配置有几个问题。

首先，您的“消费者”和通道适配器具有相同的 bean 名称 (id) messageProcessor。在 Spring 中，最后一个获胜，因此一个定义将覆盖另一个。

其次，配置所做的只是将消息转储到队列通道inputFromKafka - 与您的“消费者”没有任何连接。更改 bean 名称并添加服务激活器（并从通道中删除 <int:queue/> 元素）。

<int:channel id="inputFromKafka" />

<int:service-activator ref="myConsumer" input-channel="inputFromKafka" />

将消息从适配器路由到您的消费者 bean 方法（假设有效负载可以转换为您所需的类型）。