应用程序事件

Doubao 中英对照 Application Events

以下 Spring 应用程序事件由监听容器及其消费者发布：

• ConsumerStartingEvent：在消费者线程首次启动时、开始轮询之前发布。
• ConsumerStartedEvent：在消费者即将开始轮询时发布。
• ConsumerFailedToStartEvent：如果在 consumerStartTimeout 容器属性时限内未发布 ConsumerStartingEvent，则发布。此事件可能表明所配置的任务执行器线程数不足，无法支撑其所用于的容器及其并发数。出现此情况时还会记录一条错误消息。
• ListenerContainerIdleEvent：在 idleEventInterval 内未收到消息时发布（如果已配置）。
• ListenerContainerNoLongerIdleEvent：在先前发布过 ListenerContainerIdleEvent 之后消费到一条记录时发布。
• ListenerContainerPartitionIdleEvent：在 idlePartitionEventInterval 内未从该分区收到消息时发布（如果已配置）。
• ListenerContainerPartitionNoLongerIdleEvent：在从先前发布过 ListenerContainerPartitionIdleEvent 的分区消费到一条记录时发布。
• NonResponsiveConsumerEvent：当消费者似乎阻塞在 poll 方法中时发布。
• ConsumerPartitionPausedEvent：在某个分区被暂停时，由每个消费者发布。
• ConsumerPartitionResumedEvent：在某个分区被恢复时，由每个消费者发布。
• ConsumerPausedEvent：在容器被暂停时，由每个消费者发布。
• ConsumerResumedEvent：在容器被恢复时，由每个消费者发布。
• ConsumerStoppingEvent：在每个消费者即将停止之前发布。
• ConsumerStoppedEvent：在消费者关闭之后发布。请参阅 线程安全。
• ConsumerRetryAuthEvent：在消费者的认证或授权失败并重试时发布。
• ConsumerRetryAuthSuccessfulEvent：在认证或授权重试成功时发布。仅当此前出现过 ConsumerRetryAuthEvent 时才可能发生。
• ContainerStoppedEvent：在所有消费者都已停止时发布。
• ConcurrentContainerStoppedEvent：在 ConcurrentMessageListenerContainer 停止时发布。

important

默认情况下，应用程序上下文的事件多播器会在调用线程上调用事件监听器。如果你将多播器更改为使用异步执行器，那么当事件包含对消费者的引用时，你绝不能调用任何 Consumer 方法。

ListenerContainerIdleEvent 具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• id：监听器 ID（或容器 bean 名称）。
• idleTime：事件发布时容器已空闲的时间。
• topicPartitions：事件生成时分配给该容器的主题和分区。
• consumer：对 Kafka Consumer 对象的引用。例如，如果此前调用过消费者的 pause() 方法，则在收到事件时可以 resume()。
• paused：容器当前是否已暂停。更多信息请参阅 暂停和恢复监听容器。

ListenerContainerNoLongerIdleEvent 具有相同的属性，但 idleTime 和 paused 除外。

ListenerContainerPartitionIdleEvent 具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• id：监听器 ID（或容器 bean 名称）。
• idleTime：事件发布时分区消费已空闲的时间。
• topicPartition：触发该事件的主题和分区。
• consumer：对 Kafka Consumer 对象的引用。例如，如果此前调用过消费者的 pause() 方法，则在收到事件时可以 resume()。
• paused：该分区消费对该消费者而言当前是否已暂停。更多信息请参阅 暂停和恢复监听容器。

ListenerContainerPartitionNoLongerIdleEvent 具有相同的属性，但 idleTime 和 paused 除外。

NonResponsiveConsumerEvent 具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• id：监听器 ID（或容器 bean 名称）。
• timeSinceLastPoll：容器上次调用 poll() 之前那一刻的时间。
• topicPartitions：事件生成时分配给该容器的主题和分区。
• consumer：对 Kafka Consumer 对象的引用。例如，如果此前调用过消费者的 pause() 方法，则在收到事件时可以 resume()。
• paused：容器当前是否已暂停。更多信息请参阅 暂停和恢复监听容器。

ConsumerPausedEvent、ConsumerResumedEvent 和 ConsumerStopping 事件具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• partitions：所涉及的 TopicPartition 实例。

ConsumerPartitionPausedEvent、ConsumerPartitionResumedEvent 事件具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• partition：所涉及的 TopicPartition 实例。

ConsumerRetryAuthEvent 事件具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。
• reason：
  • AUTHENTICATION —— 该事件是因认证异常而发布的。
  • AUTHORIZATION —— 该事件是因授权异常而发布的。

ConsumerStartingEvent、ConsumerStartedEvent、ConsumerFailedToStartEvent、ConsumerStoppedEvent、ConsumerRetryAuthSuccessfulEvent 和 ContainerStoppedEvent 事件具有以下属性：

• source：发布该事件的监听容器实例。
• container：监听容器；如果源容器是子容器，则为父监听容器。

所有容器（无论是子容器还是父容器）都会发布 ContainerStoppedEvent。对于父容器，source 和 container 属性是相同的。

此外，ConsumerStoppedEvent 还具有以下附加属性：

• reason：
  • NORMAL —— 消费者正常停止（容器被停止）。
  • ABNORMAL —— 消费者异常停止（容器被异常停止）。
  • ERROR —— 抛出了一个 java.lang.Error。
  • FENCED —— 事务生产者被围栏隔离，且 stopContainerWhenFenced 容器属性为 true。
  • AUTH —— 抛出了 AuthenticationException 或 AuthorizationException，且未配置 authExceptionRetryInterval。
  • NO_OFFSET —— 某个分区没有偏移量，且 auto.offset.reset 策略为 none。

你可以使用此事件在出现此类情况后重启容器：

if (event.getReason().equals(Reason.FENCED)) {
    event.getSource(MessageListenerContainer.class).start();
}

检测空闲和无响应的消费者

异步消费者虽然高效，但一个问题是如何检测它们何时空闲。如果在一段时间内没有消息到达，你可能希望采取某些动作。

你可以将监听容器配置为在一段时间内没有消息投递时发布一个 ListenerContainerIdleEvent。容器空闲时，每隔 idleEventInterval 毫秒发布一个事件。

要配置此特性，请在容器上设置 idleEventInterval。以下示例展示了如何做到这一点：

@Bean
public KafkaMessageListenerContainer(ConsumerFactory<String, String> consumerFactory) {
    ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
    ...
    containerProps.setIdleEventInterval(60000L);
    ...
    KafkaMessageListenerContainer<String, String> container = new KafKaMessageListenerContainer<>(consumerFactory, containerProps);
    return container;
}

以下示例展示了如何为 @KafkaListener 设置 idleEventInterval：

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
    ...
    factory.getContainerProperties().setIdleEventInterval(60000L);
    ...
    return factory;
}

在以上每种情况下，容器空闲时每分钟发布一次事件。

如果由于某种原因消费者的 poll() 方法没有退出，就不会收到消息，也无法生成空闲事件（在 kafka-clients 早期版本中，当代理不可达时这是个问题）。在这种情况下，如果一次 poll 在 pollTimeout 属性的 3x 时间内没有返回，容器就会发布一个 NonResponsiveConsumerEvent。默认情况下，此检查在每个容器中每 30 秒执行一次。你可以在配置监听容器时，通过在 ContainerProperties 中设置 monitorInterval（默认 30 秒）和 noPollThreshold（默认 3.0）属性来修改此行为。noPollThreshold 应大于 1.0，以避免因竞态条件而产生虚假事件。收到这样的事件后，你可以停止容器，从而唤醒消费者使其能够停止。

从 2.6.2 版本开始，如果某个容器已发布过 ListenerContainerIdleEvent，那么在随后收到一条记录时，它会发布一个 ListenerContainerNoLongerIdleEvent。

事件消费

你可以通过实现 ApplicationListener 来捕获这些事件——可以是通用监听器，也可以是收窄为只接收此特定事件的监听器。你也可以使用 Spring Framework 4.2 引入的 @EventListener。

下一个示例将 @KafkaListener 和 @EventListener 组合到同一个类中。你应当明白，应用程序监听器会接收所有容器的事件，因此如果你想根据哪个容器空闲来采取特定动作，可能需要检查监听器 ID。为此你也可以使用 @EventListener 的 condition。

有关事件属性的信息，请参阅 应用程序事件。

该事件通常在消费者线程上发布，因此与 Consumer 对象交互是安全的。

以下示例同时使用了 @KafkaListener 和 @EventListener：

public class Listener {

    @KafkaListener(id = "qux", topics = "annotated")
    public void listen4(@Payload String foo, Acknowledgment ack) {
        ...
    }

    @EventListener(condition = "event.listenerId.startsWith('qux-')")
    public void eventHandler(ListenerContainerIdleEvent event) {
        ...
    }

}

important

事件监听器会看到所有容器的事件。因此，在上一示例中，我们根据监听器 ID 对收到的事件进行了收窄。由于为 @KafkaListener 创建的容器支持并发，实际容器被命名为 id-n，其中 n 是每个实例的唯一值以支持并发。这就是我们在条件中使用 startsWith 的原因。

警告

如果你希望使用空闲事件来停止监听容器，就不应在调用监听器的线程上调用 container.stop()。这样做会导致延迟和不必要的日志消息。相反，你应将事件移交到另一个线程，由该线程停止容器。此外，如果该容器是子容器，你不应 stop() 该容器实例。而应停止并发容器。

空闲时的当前位置

请注意，你可以通过在监听器中实现 ConsumerSeekAware 来在检测到空闲时获取当前位置。请参阅 seek 中的 onIdleContainer()。