configurelistenercontainer时KAFKA消息错误的集成流

Question

我正在尝试使用IntegrationFlow for Kafka将从Kafka接收到的消息传递到频道。

以下是我的工作代码： -

    @Bean
    public MessageChannel fromKafka() {
        return new DirectChannel();
    }

    @Bean
    public IntegrationFlow topic1ListenerFromKafkaFlow1() throws Exception {
/*        return IntegrationFlows
                .from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
                                KafkaMessageDrivenChannelAdapter.ListenerMode.record, kafkaTopic)
                        .configureListenerContainer( c ->  c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
                                        .id("topic1ListenerContainer"))
                        .recoveryCallback(new ErrorMessageSendingRecoverer(messageFromKafka(),
                                new RawRecordHeaderErrorMessageStrategy()))
                        .retryTemplate(new RetryTemplate())
                        .filterInRetry(true))
                .filter(Message.class, m ->
                                m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
                        f -> f.throwExceptionOnRejection(true))
                .<String, String>transform(String::toUpperCase)
                .channel(c -> c.queue("listeningFromKafkaResults1"))
                .get();*/



        return IntegrationFlows
                .from(Kafka.messageDrivenChannelAdapter(listener(), KafkaMessageDrivenChannelAdapter.ListenerMode.record))
                .channel("fromKafka")
                .get();
    }





    @Bean("listenerkafka")
    public KafkaMessageListenerContainer<String, String> listener() throws Exception {
        ContainerProperties properties = new ContainerProperties(kafkaTopic1);
        properties.setGroupId("kafka-test");
        return new KafkaMessageListenerContainer<>(consumerFactory, properties);
    }

    @ServiceActivator(inputChannel="fromKafka", outputChannel = "somechannel")
    public Message<CreatRequest> fromKafka(Message<?> msg) throws JsonProcessingException {
        CreatRequest  creatRequest =  objectMapper.readValue(msg.getPayload().toString(), CreatRequest.class);
        Message<CreatRequest> message= MessageBuilder.withPayload(creatRequest).build();
        logger.info("Inside fromKafka " + message);
        return  message;
    }

我面临的问题是评论代码在主题1listenerfromkafkaflow中不起作用。 在这里，我找不到C.Ackmode（AbstractMessageListenerContainer.ackmode.manual），

因为它显示了未识别的编译时间错误。 你能在我出错的地方纠正我吗？

另外，我需要将此流传递到另一个线程中，而不是在主线程中。

Answer 1

代替使用KAFKA消息驱动的通道适配器：

https://docs.spring.io/spring-integration/docs/current/referent/referent/html/kafka.html#kafka-inbound

但是，在同一频道上有两个适配器，请求将是圆形的 -罗宾分布在他们之间。如果您希望两者都接收消息，则需要 PublishSubscribechannel 。