spring integration kafka - 暂停/从我的听众那里寻找消费者一小段时间

Question

我的 Spring 集成图如下所示。

在我的服务激活器中，是否可以根据我的消息获取侦听器，然后将其暂停一段时间，如果更进一步，我的处理速度很慢？我需要这种方式来处理一些溢出机制。

我知道我可以实现一个新的 consumerSeekCallback，但在集成设置中，据我了解，我无法访问 messageDrivenChannelAdapter。

我正在使用链接到消息驱动通道适配器的 ConcurrentMessageListenerContainer。

    <int-kafka:message-driven-channel-adapter
    id="kafkaListenertest" listener-container="containertest" auto-startup="true"
    phase="100" send-timeout="5000" channel="kafkaMessage" error-channel="overflow"  />

<bean id="containertest"
    class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer">
    <constructor-arg ref="kafkaConsumerFactory"/>
    <constructor-arg ref="consumerContainerPropertiestest" />
    <property name="concurrency" value="4"/>
</bean>

<bean id="consumerContainerPropertiestest"
    class="org.springframework.kafka.listener.config.ContainerProperties">
    <constructor-arg name="topics" value="events.test" />
    <property name="ackMode" value="MANUAL_IMMEDIATE"></property>
</bean>

<int:service-activator input-channel="kafkaMessage"
    ref="MyListener" method="handleIncomingKafkaEvent" ></int:service-activator>

<int:channel id="kafkaMessage"></int:channel>

Answer 1

启动版本2.1.3，Spring Kafka在MessageListenerContainer上提供了这个API：

/**
 * Pause this container before the next poll().
 * @since 2.1.3
 */
default void pause() {
    throw new UnsupportedOperationException("This container doesn't support pause");
}

/**
 * Resume this container, if paused, after the next poll().
 * @since 2.1.3
 */
default void resume() {
    throw new UnsupportedOperationException("This container doesn't support resume");
}

/**
 * Return true if {@link #pause()} has been called; the container might not have actually
 * paused yet.
 * @return true if pause has been requested.
 * @since 2.1.5
 */
default boolean isPauseRequested() {
    throw new UnsupportedOperationException("This container doesn't support pause/resume");
}

/**
 * Return true if {@link #pause()} has been called; and all consumers in this container
 * have actually paused.
 * @return true if the container is paused.
 * @since 2.1.5
 */
default boolean isContainerPaused() {
    throw new UnsupportedOperationException("This container doesn't support pause/resume");
}

因此，您确实可以从应用程序的任何位置暂停和恢复您的侦听器容器，将 containertest 注入到适当的服务中。

KafkaMessageDrivenChannelAdapter 还公开了 pause() 和 resume() 挂钩。

另外，在KafkaMessageDrivenChannelAdapter 中，MessagingMessageConverter 将这些标头填充到消息中以供下游处理：

    rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, record.key());
    rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, record.topic());
    rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, record.partition());
    rawHeaders.put(KafkaHeaders.OFFSET, record.offset());
    rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, record.timestampType().name());
    rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, record.timestamp());

    if (acknowledgment != null) {
        rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
    }
    if (consumer != null) {
        rawHeaders.put(KafkaHeaders.CONSUMER, consumer);
    }

因此，您可以获取KafkaHeaders.CONSUMER 标头并使用来自KafkaConsumer 的本机pause()/resume()。