正文

第三集 Spring for Apache Kafka 接受消息

toobug
toobugV管理员/1.27 W阅读/0评论
0325
此篇文章发布距今已超过1483天，您需要注意文章的内容或图片是否可用！

我们可以接受消息通过配置一个MessageListenerContainer 和提供一个消息监听或者通过使用@KafkaListener 注解

3.1 Message Listeners

当我们使用一个消息监听容器的时候，我们必须提供一个监听来接受数据。
当前有八种支持消息监听的接口，以下是这些接口列表：

public interface MessageListener<K, V> { 

    void onMessage(ConsumerRecord<K, V> data);

}

使用自动提交或其中一个容器管理的提交方法时，使用此接口处理从Kafka使用者poll()操作接收的各个ConsumerRecord实例。

public interface AcknowledgingMessageListener<K, V> { 

    void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment);

}

使用其中一种手动提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的各个ConsumerRecord实例。

public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V> { 

    void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);

}

使用自动提交或其中一个容器管理的提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的各个ConsumerRecord实例。 提供对Consumer对象的访问。

public interface AcknowledgingConsumerAwareMessageListener<K, V> extends MessageListener<K, V> { 

    void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);

}

使用其中一种手动提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的各个ConsumerRecord实例。 提供对Consumer对象的访问

public interface BatchMessageListener<K, V> { 

    void onMessage(List<ConsumerRecord<K, V>> data);

}

使用自动提交或其中一个容器管理的提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的所有ConsumerRecord实例。 使用此接口时不支持AckMode.RECORD，因为将为侦听器提供完整的批处理。

public interface BatchAcknowledgingMessageListener<K, V> { 

    void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);

}

使用其中一种手动提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的所有ConsumerRecord实例。

public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { 

    void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);

}

使用自动提交或其中一个容器管理的提交方法时的操作。 使用此接口时不支持AckMode.RECORD，因为将为侦听器提供完整的批处理。 提供对Consumer对象的访问。

public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { 

    void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);

}

使用其中一种手动提交方法时，使用此接口处理从Kafka使用者poll（）操作接收的所有ConsumerRecord实例。 提供对Consumer对象的访问。

Consumer对象不是线程安全的。 您只能在调用侦听器的线程上调用其方法。

3.2 Message Listener Containers

提供了两个消息监听容器实现

  • KafkaMessageListenerContainer
  • ConcurrentMessageListenerContainer

这个KafkaMessageLisenterContainer 在一个线程中接受所有的消息从所有的主题中或者分区中。
这个ConcurrentMessageListenerContainer 代理接受一个或者多个KafkaMessageListenerContainer 实例通过提供多个线程。

3.2.1 使用KafkaMessageListenerContainer

构造方法如下所示：

public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                    ContainerProperties containerProperties)

public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                    ContainerProperties containerProperties,
                    TopicPartitionInitialOffset... topicPartitions)

每个都采用ConsumerFactory和有关主题和分区的信息，以及ContainerProperties对象中的其他配置。 ConcurrentMessageListenerContainer（稍后描述）使用第二个构造函数跨消费者实例分发TopicPartitionInitialOffset。 ContainerProperties具有以下构造函数：

public ContainerProperties(TopicPartitionInitialOffset... topicPartitions)

public ContainerProperties(String... topics)

public ContainerProperties(Pattern topicPattern)

第一个构造函数接受一个TopicPartitionInitialOffset参数数组，以显式指示容器使用哪些分区（使用consumer assign（）方法）和可选的初始偏移量。 正值是默认的绝对偏移量。 默认情况下，负值相对于分区中的当前最后一个偏移量。 提供了一个TopicPartitionInitialOffset的构造函数，它接受一个额外的布尔参数。 如果这是真的，则初始偏移（正或负）相对于该消费者的当前位置。 启动容器时应用偏移量。 第二个采用一系列主题，Kafka根据group.id属性分配分区 - 在整个组中分配分区。 第三个使用正则表达式模式来选择主题。

要将MessageListener分配给容器，可以在创建Container时使用ContainerProps.setMessageListener方法。 以下示例显示了如何执行此操作：

ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
containerProps.setMessageListener(new MessageListener<Integer, String>() {
    ...
});
DefaultKafkaConsumerFactory<Integer, String> cf =
                        new DefaultKafkaConsumerFactory<Integer, String>(consumerProps());
KafkaMessageListenerContainer<Integer, String> container =
                        new KafkaMessageListenerContainer<>(cf, containerProps);
return container;

有关可以设置的各种属性的更多信息，请参阅Javadoc for ContainerProperties。

从版本2.1.1开始，可以使用名为logContainerConfig的新属性。如果启用了true和INFO日志记录，则每个侦听器容器都会写入一条记录其配置属性的日志消息。

默认情况下，在DEBUG日志记录级别执行主题偏移提交的日志记录。从版本2.1.2开始，ContainerProperties中名为commitLogLevel的属性允许您指定这些消息的日志级别。例如，要将日志级别更改为INFO，可以使用containerProperties.setCommitLogLevel（LogIfLevelEnabled.Level.INFO）;.

从2.2版开始，添加了一个名为missingTopicsFatal的新容器属性（默认值：true）。如果代理上不存在任何已配置的主题，则会阻止容器启动。如果容器配置为侦听主题模式（正则表达式），则不适用。以前，容器线程在consumer.poll（）方法中循环，等待在记录许多消息时显示主题。除了日志之外，没有迹象表明存在问题。要还原以前的行为，可以将该属性设置为false。

3.2.2 使用ConcurrentMessageListenerContainer

单个构造函数类似于第一个KafkaListenerContainer构造函数。 以下清单显示了构造函数的签名：

public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                            ContainerProperties containerProperties)

它还具有并发属性。 例如，container.setConcurrency（3）创建三个KafkaMessageListenerContainer实例。

对于第一个构造函数，Kafka使用其组管理功能在消费者之间分配分区。

收听多个主题时，默认分区分发可能与您的预期不同。 例如，如果您有三个主题，每个主题有五个分区，并且您希望使用concurrency =
15，则只能看到五个活动使用者，每个主用户分配一个分区，其他10个使用者处于空闲状态。 这是因为默认的Kafka
PartitionAssignor是RangeAssignor（请参阅其Javadoc）。
对于这种情况，您可能需要考虑使用RoundRobinAssignor，它将分区分配给所有使用者。 然后，为每个使用者分配一个主题或分区。
要更改PartitionAssignor，可以在提供给DefaultKafkaConsumerFactory的属性中设置partition.assignment.strategy使用者属性（ConsumerConfigs.PARTITION_ASSIGNMENT_STRATEGY_CONFIG）。

使用Spring Boot时，您可以按如下方式分配策略：

spring.kafka.consumer.properties.partition.assignment.strategy=\
org.apache.kafka.clients.consumer.RoundRobinAssignor

对于第二个构造函数，ConcurrentMessageListenerContainer在委托KafkaMessageListenerContainer实例中分发TopicPartition实例。

例如，如果提供了六个TopicPartition实例并且并发性为3; 每个容器有两个分区。 对于五个TopicPartition实例，两个容器获得两个分区，第三个获得一个。 如果并发性大于TopicPartitions的数量，则调整并发性以使每个容器获得一个分区。

client.id属性（如果设置）附加-n，其中n是与并发相对应的使用者实例。 启用JMX时，需要为MBean提供唯一的名称。

从版本1.3开始，MessageListenerContainer提供对底层KafkaConsumer的度量的访问。 对于ConcurrentMessageListenerContainer，metrics（）方法返回所有目标KafkaMessageListenerContainer实例的度量标准。 指标分组到Map <MetricName，？ 通过为底层KafkaConsumer提供的client-id扩展Metric>。

3.3 Committing Offsets 提交偏移量

提供了几种用于提交偏移的选项。 如果enable.auto.commit使用者属性为true，则Kafka会根据其配置自动提交偏移量。 如果为false，则容器支持多个AckMode设置（在下一个列表中描述）。

消费者poll（）方法返回一个或多个ConsumerRecords。 为每条记录调用MessageListener。 以下列表描述了容器为每个AckMode采取的操作：

  • RECORD：在处理记录后侦听器返回时提交偏移量。

  • BATCH：在处理poll（）返回的所有记录时提交偏移量。

  • TIME：处理poll（）返回的所有记录时的偏移量，只要超过自上次提交以来的ackTime。

  • COUNT：只要自上次提交后已收到ackCount记录，就会在处理poll（）返回的所有记录时提交偏移量。

  • COUNT_TIME：类似于TIME和COUNT，但如果任一条件为真，则执行提交。

  • MANUAL：消息监听器负责确认（）确认。 之后，应用与BATCH相同的语义。

  • MANUAL_IMMEDIATE：在侦听器调用Acknowledgment.acknowledge（）方法时立即提交偏移量。

MANUAL和MANUAL_IMMEDIATE要求侦听器是AcknowledgingMessageListener或BatchAcknowledgingMessageListener。
请参阅消息监听器。

根据syncCommits容器属性，使用使用者上的commitSync（）或commitAsync（）方法。

确认具有以下方法：

public interface Acknowledgment {

    void acknowledge();

}

此方法使侦听器可以控制何时提交偏移。

Listener Container Auto Startup

侦听器容器实现SmartLifecycle，默认情况下autoStartup为true。 容器在后期启动（Integer.MAX-VALUE - 100）。 应该在早期阶段启动实现SmartLifecycle以处理来自侦听器的数据的其他组件。 -100为以后的阶段留出了空间，使组件能够在容器之后自动启动。

@KafkaListener 注解

@KafkaListener注释用于将bean方法指定为侦听器容器的侦听器。 该bean包含在MessagingMessageListenerAdapter中，该MessagingMessageListenerAdapter配置有各种功能，例如转换器以在必要时转换数据以匹配方法参数。

您可以使用＃{…}或属性占位符（$ {…}）使用SpEL在注释上配置大多数属性。 有关更多信息，请参阅Javadoc。

Record Listeners

@KafkaListener注解为简单的POJO侦听器提供了一种机制。 以下示例显示了如何使用它：

public class Listener {

    @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
    public void listen(String data) {
        ...
    }

}

此机制需要在其中一个@Configuration类和一个侦听器容器工厂上使用@EnableKafka注释，该工厂用于配置基础ConcurrentMessageListenerContainer。 默认情况下，需要名为kafkaListenerContainerFactory的bean。 以下示例显示如何使用ConcurrentMessageListenerContainer：

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        ...
        return props;
    }
}

请注意，要设置容器属性，必须在工厂中使用getContainerProperties（）方法。 它用作注入容器的实际属性的模板。

从版本2.1.1开始，您现在可以为注释创建的使用者设置client.id属性。 clientIdPrefix以-n为后缀，其中n是表示使用并发时的容器编号的整数。

从2.2版开始，您现在可以通过使用注释本身的属性来覆盖容器工厂的并发和autoStartup属性。 属性可以是简单值，属性占位符或SpEL表达式。 以下示例显示了如何执行此操作：

@KafkaListener(id = "myListener", topics = "myTopic",
        autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")
public void listen(String data) {
    ...
}

您还可以使用显式主题和分区（以及可选的初始偏移量）配置POJO侦听器。 以下示例显示了如何执行此操作：

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

您可以在分区或partitionOffsets属性中指定每个分区，但不能同时指定两者。

使用手动AckMode时，您还可以向监听器提供确认。 以下示例还说明了如何使用其他容器工厂。

@KafkaListener(id = "cat", topics = "myTopic",
          containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
    ...
    ack.acknowledge();
}

最后，可以从邮件头中获取有关邮件的元数据。 您可以使用以下标头名称来检索邮件的标头：

  • KafkaHeaders.RECEIVED_MESSAGE_KEY
  • KafkaHeaders.RECEIVED_TOPIC
  • KafkaHeaders.RECEIVED_PARTITION_ID
  • KafkaHeaders.RECEIVED_TIMESTAMP
  • KafkaHeaders.TIMESTAMP_TYPE

以下示例显示了如何使用标头：

@KafkaListener(id = "qux", topicPattern = "myTopic1")
public void listen(@Payload String foo,
        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
        ) {
    ...

Batch listeners

从1.1版开始，您可以配置@KafkaListener方法以接收从消费者调查中收到的整批消费者记录。 要配置侦听器容器工厂以创建批处理侦听器，可以设置batchListener属性。 以下示例显示了如何执行此操作：

@Bean
public KafkaListenerContainerFactory<?> batchFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);  // <<<<<<<<<<<<<<<<<<<<<<<<<
    return factory;
}

以下示例显示如何接收有效负载列表

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {
    ...
}

主题，分区，偏移等在与有效负载并行的标头中可用。 以下示例显示了如何使用标头：

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list,
        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<Integer> keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
        @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
    ...
}

或者，您可以在每条消息中接收带有每个偏移量和其他详细信息的消息<？>对象列表，但它必须是唯一的参数（除了可选的确认，使用手动提交时，和/或消费者<？，？> 参数）在方法上定义。 以下示例显示了如何执行此操作：

@KafkaListener(id = "listMsg", topics = "myTopic", containerFactory = "batchFactory")
public void listen14(List<Message<?>> list) {
    ...
}

@KafkaListener(id = "listMsgAck", topics = "myTopic", containerFactory = "batchFactory")
public void listen15(List<Message<?>> list, Acknowledgment ack) {
    ...
}

@KafkaListener(id = "listMsgAckConsumer", topics = "myTopic", containerFactory = "batchFactory")
public void listen16(List<Message<?>> list, Acknowledgment ack, Consumer<?, ?> consumer) {
    ...
}

在这种情况下，不对有效载荷执行转换。

如果BatchMessagingMessageConverter配置了RecordMessageConverter，您还可以向Message参数添加泛型类型并转换有效负载。 有关详细信息，请参阅使用批量侦听器的有效负载转换

您还可以接收ConsumerRecord <？，？>对象的列表，但它必须是该方法上定义的唯一参数（除了可选的确认，当使用手动提交和Consumer <？，？>参数时）。 以下示例显示了如何执行此操作：

@KafkaListener(id = "listCRs", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<Integer, String>> list) {
    ...
}

@KafkaListener(id = "listCRsAck", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<Integer, String>> list, Acknowledgment ack) {
    ...
}

从2.2版开始，侦听器可以接收poll（）方法返回的完整ConsumerRecords <？，？>对象，让侦听器访问其他方法，例如partitions（）（返回列表中的TopicPartition实例）和记录 （TopicPartition）（获取选择性记录）。 同样，这必须是方法上唯一的参数（除了可选的确认，当使用手动提交或消费者<？，？>参数时）。 以下示例显示了如何执行此操作：

@KafkaListener(id = "pollResults", topics = "myTopic", containerFactory = "batchFactory")
public void pollResults(ConsumerRecords<?, ?> records) {
    ...
}

如果容器工厂配置了RecordFilterStrategy，则会忽略ConsumerRecords
<？，？>侦听器，并发出WARN日志消息。 如果使用<List <？>>形式的侦听器，则只能使用批量侦听器过滤记录。

注解属性

从版本2.0开始，id属性（如果存在）用作Kafka使用者group.id属性，覆盖使用者工厂中已配置的属性（如果存在）。 您还可以显式设置groupId或将idIsGroup设置为false以恢复使用使用者工厂group.id的先前行为。

您可以在大多数注释属性中使用属性占位符或SpEL表达式，如以下示例所示：

@KafkaListener(topics = "${some.property}")

@KafkaListener(topics = "#{someBean.someProperty}",
    groupId = "#{someBean.someProperty}.group")

从版本2.1.2开始，SpEL表达式支持一个特殊的令牌：__listener。 它是一个伪bean名称，表示存在此批注的当前bean实例。

请考虑以下示例：

@Bean
public Listener listener1() {
    return new Listener("topic1");
}

@Bean
public Listener listener2() {
    return new Listener("topic2");
}

鉴于上一个示例中的bean，我们可以使用以下内容：

public class Listener {

    private final String topic;

    public Listener(String topic) {
        this.topic = topic;
    }

    @KafkaListener(topics = "#{__listener.topic}",
        groupId = "#{__listener.topic}.group")
    public void listen(...) {
        ...
    }

    public String getTopic() {
        return this.topic;
    }

}

如果您有一个名为__listener的实际bean，则可以使用beanRef属性更改表达式标记。 以下示例显示了如何执行此操作：

@KafkaListener(beanRef = "__x", topics = "#{__x.topic}",
    groupId = "#{__x.topic}.group")

从2.2.4版开始，您可以直接在注释上指定Kafka使用者属性，这些属性将覆盖在使用者工厂中配置的具有相同名称的任何属性。 您不能以这种方式指定group.id和client.id属性; 他们会被忽视; 使用groupId和clientIdPrefix注释属性。

属性被指定为具有普通Java属性文件格式的单个字符串：foo：bar，foo = bar或foo bar。

@KafkaListener(topics = "myTopic", groupId="group", properties= {
    "max.poll.interval.ms:60000",
    ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100"
})

容器线程命名

监听器容器当前使用两个任务执行器，一个用于调用使用者，另一个用于在kafka使用者属性enable.auto.commit为false时调用监听器。您可以通过设置容器的ContainerProperties的consumerExecutor和listenerExecutor属性来提供自定义执行程序。使用池化执行程序时，请确保有足够的线程可用于处理使用它们的所有容器的并发性。使用ConcurrentMessageListenerContainer时，每个消费者使用一个线程（并发）。

如果您未提供使用者执行程序，则使用SimpleAsyncTaskExecutor。此执行程序创建名称类似于 -C-1（使用者线程）的线程。对于ConcurrentMessageListenerContainer，线程名称的部分变为 -m，其中m表示使用者实例。每次启动容器时，n都会递增。因此，使用容器的bean名称，容器第一次启动后，此容器中的线程将被命名为container-0-C-1，container-1-C-1等;容器-0-C-2，容器-1-C-2等，停止并随后启动。

@KafkaListener 作为一个元注解

从2.2版开始，您现在可以使用@KafkaListener作为元注释。 以下示例显示了如何执行此操作：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@KafkaListener
public @interface MyThreeConsumersListener {

    @AliasFor(annotation = KafkaListener.class, attribute = "id")
    String id();

    @AliasFor(annotation = KafkaListener.class, attribute = "topics")
    String[] topics();

    @AliasFor(annotation = KafkaListener.class, attribute = "concurrency")
    String concurrency() default "3";

}

除非已在使用者工厂配置中指定了group.id，否则必须至少为其中一个主题，topicPattern或topicPartitions（以及通常为id或groupId）添加别名。 以下示例显示了如何执行此操作：

@MyThreeConsumersListener(id = "my.group", topics = "my.topic")
public void listen1(String in) {
    ...
}

在一个类上使用@KafkaListener

在类级别使用@KafkaListener时，必须在方法级别指定@KafkaHandler。 传递消息时，转换的消息有效负载类型用于确定要调用的方法。 以下示例显示了如何执行此操作：

@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {

    @KafkaHandler
    public void listen(String foo) {
        ...
    }

    @KafkaHandler
    public void listen(Integer bar) {
        ...
    }

    @KafkaHandler(isDefault = true`)
    public void listenDefault(Object object) {
        ...
    }

}

从版本2.1.3开始，如果与其他方法不匹配，则可以将@KafkaHandler方法指定为调用的默认方法。 最多可以指定一种方法。 使用@KafkaHandler方法时，有效负载必须已经转换为域对象（因此可以执行匹配）。 使用自定义反序列化器，JsonDeserializer或（String | Bytes）JsonMessageConverter，并将其TypePrecedence设置为TYPE_ID。 有关更多信息，请参阅序列化，反序列化和消息转换。

@KafkaListener 生命周期管理

为@KafkaListener注释创建的侦听器容器不是应用程序上下文中的bean。相反，它们是使用KafkaListenerEndpointRegistry类型的基础结构bean注册的。这个bean由框架自动声明并管理容器的生命周期;它将自动启动autoStartup设置为true的任何容器。所有容器工厂创建的所有容器必须处于同一阶段。有关更多信息，请参阅监听器容器自动启动。您可以使用注册表以编程方式管理生命周期。启动或停止注册表将启动或停止所有已注册的容器。或者，您可以使用其id属性获取对单个容器的引用。您可以在注释上设置autoStartup，该注释将覆盖配置到容器工厂中的默认设置。您可以从应用程序上下文中获取对bean的引用，例如自动布线，以管理其已注册的容器。以下示例显示了如何执行此操作：

@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
public void listen(...) { ... }

@Autowired
private KafkaListenerEndpointRegistry registry;

...

    this.registry.getListenerContainer("myContainer").start();

...

@KafkaListener @Payload 校验

从2.2版开始，现在可以更轻松地添加Validator来验证@KafkaListener @Payload参数。 以前，您必须配置自定义DefaultMessageHandlerMethodFactory并将其添加到注册器。 现在，您可以将验证程序添加到注册商本身。 以下代码显示了如何执行此操作：

@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {

    ...

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
      registrar.setValidator(new MyValidator());
    }
}

将Spring Boot与验证启动器一起使用时，会自动配置LocalValidatorFactoryBean，如以下示例所示：

@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {

    @Autowired
    private LocalValidatorFactoryBean validator;
    ...

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
      registrar.setValidator(this.validator);
    }
}

以下示例显示如何验证：

public static class ValidatedClass {

  @Max(10)
  private int bar;

  public int getBar() {
    return this.bar;
  }

  public void setBar(int bar) {
    this.bar = bar;
  }

}

@KafkaListener(id="validated", topics = "annotated35", errorHandler = "validationErrorHandler",
      containerFactory = "kafkaJsonListenerContainerFactory")
public void validatedListener(@Payload @Valid ValidatedClass val) {
    ...
}

@Bean
public KafkaListenerErrorHandler validationErrorHandler() {
    return (m, e) -> {
        ...
    };
}

Rebalancing Listeners

ContainerProperties有一个名为consumerRebalanceListener的属性，它接受Kafka客户端的ConsumerRebalanceListener接口的实现。 如果未提供此属性，则容器将配置记录侦听器，以在INFO级别记录重新平衡事件。 该框架还添加了一个子接口ConsumerAwareRebalanceListener。 以下清单显示了ConsumerAwareRebalanceListener接口定义：

public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {

    void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

    void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

    void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

}

请注意，撤消分区时有两个回调。 第一个是立即调用的。 在提交任何挂起的偏移量之后调用第二个。 如果您希望在某些外部存储库中维护偏移量，这非常有用，如以下示例所示：

containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // acknowledge any pending Acknowledgments (if using manual acks)
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // ...
            store(consumer.position(partition));
        // ...
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // ...
            consumer.seek(partition, offsetTracker.getOffset() + 1);
        // ...
    }
});

Forwarding Listener Results using @SendTo

从2.0版开始，如果您还使用@SendTo批注注释@KafkaListener并且方法调用返回结果，则结果将转发到@SendTo指定的主题。

@SendTo值可以有多种形式：

  • @SendTo（“someTopic”）路由到文字主题

  • @SendTo（“＃{someExpression}”）路由到在应用程序上下文初始化期间通过计算表达式确定的主题。

  • @SendTo（“！{someExpression}”）路由到通过在运行时计算表达式确定的主题。 评估的#root对象有三个属性：

    • request：入站ConsumerRecord（或批处理侦听器的ConsumerRecords对象））

    • source：从请求转换的org.springframework.messaging.Message <？>。

    • result：方法返回结果。

  • @SendTo（无属性）：这被视为！{source.headers [‘kafka_replyTopic’]}（自版本2.1.3起）。

从版本2.1.11和2.2.1开始，属性占位符在@SendTo值内解析。

表达式求值的结果必须是表示主题名称的String。 以下示例显示了使用@SendTo的各种方法：

@KafkaListener(topics = "annotated21")
@SendTo("!{request.value()}") // runtime SpEL
public String replyingListener(String in) {
    ...
}

@KafkaListener(topics = "${some.property:annotated22}")
@SendTo("#{myBean.replyTopic}") // config time SpEL
public Collection<String> replyingBatchListener(List<String> in) {
    ...
}

@KafkaListener(topics = "annotated23", errorHandler = "replyErrorHandler")
@SendTo("annotated23reply") // static reply topic definition
public String replyingListenerWithErrorHandler(String in) {
    ...
}
...
@KafkaListener(topics = "annotated25")
@SendTo("annotated25reply1")
public class MultiListenerSendTo {

    @KafkaHandler
    public String foo(String in) {
        ...
    }

    @KafkaHandler
    @SendTo("!{'annotated25reply2'}")
    public String bar(@Payload(required = false) KafkaNull nul,
            @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {
        ...
    }

}

从2.2版开始，您可以将ReplyHeadersConfigurer添加到侦听器容器工厂。 查阅此信息以确定要在回复消息中设置哪些标头。 以下示例显示如何添加ReplyHeadersConfigurer：

@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer((k, v) -> k.equals("cat"));
    return factory;
}

如果您愿意，还可以添加更多标题。 以下示例显示了如何执行此操作：

@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer(new ReplyHeadersConfigurer() {

      @Override
      public boolean shouldCopy(String headerName, Object headerValue) {
        return false;
      }

      @Override
      public Map<String, Object> additionalHeaders() {
        return Collections.singletonMap("qux", "fiz");
      }

    });
    return factory;
}

使用@SendTo时，必须在其replyTemplate属性中使用KafkaTemplate配置ConcurrentKafkaListenerContainerFactory以执行发送。

除非您使用请求/回复语义，否则仅使用简单的send（topic，value）方法，因此您可能希望创建子类来生成分区或键。
以下示例显示了如何执行此操作：

@Bean
public KafkaTemplate<String, String> myReplyingTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory()) {

        @Override
        public ListenableFuture<SendResult<String, String>> send(String topic, String data) {
            return super.send(topic, partitionForData(data), keyForData(data), data);
        }

        ...

    };
}

如果侦听器方法返回Message <？>或Collection <Message <？>>，则侦听器方法负责设置回复的邮件头。 例如，在处理来自ReplyingKafkaTemplate的请求时，您可能会执行以下操作：

@KafkaListener(id = "messageReturned", topics = "someTopic")
public Message<?> listen(String in, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
        @Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) {
    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.TOPIC, replyTo)
            .setHeader(KafkaHeaders.MESSAGE_KEY, 42)
            .setHeader(KafkaHeaders.CORRELATION_ID, correlation)
            .setHeader("someOtherHeader", "someValue")
            .build();
}

使用请求/回复语义时，发件人可以请求目标分区。

即使没有返回结果，也可以使用@SendTo注释@KafkaListener方法。
这是为了允许配置errorHandler，它可以将有关失败消息传递的信息转发到某个主题。 以下示例显示了如何执行此操作：

KafkaListener(id = "voidListenerWithReplyingErrorHandler", topics = "someTopic",
        errorHandler = "voidSendToErrorHandler")
@SendTo("failures")
public void voidListenerWithReplyingErrorHandler(String in) {
    throw new RuntimeException("fail");
}

@Bean
public KafkaListenerErrorHandler voidSendToErrorHandler() {
    return (m, e) -> {
        return ... // some information about the failure and input data
    };
}

有关更多信息，请参阅处理异常

Filtering Messages

在某些情况下，例如重新平衡，可以重新传递已经处理的消息。框架无法知道是否已处理此类消息。这是一个应用程序级功能。这被称为Idempotent Receiver模式，Spring Integration提供了它的实现。

Spring for Apache Kafka项目还通过FilteringMessageListenerAdapter类提供一些帮助，该类可以包装MessageListener。此类采用RecordFilterStrategy的实现，您可以在其中实现filter方法，以指示消息是重复的并且应该被丢弃。这有一个名为ackDiscarded的附加属性，它指示适配器是否应该确认丢弃的记录。默认情况下为假。

使用@KafkaListener时，在容器工厂上设置RecordFilterStrategy（以及可选的ackDiscarded），以便将侦听器包装在适当的过滤适配器中。

此外，还提供了FilteringBatchMessageListenerAdapter，供您在使用批处理消息侦听器时使用。

如果@KafkaListener收到ConsumerRecords <？，？>而不是List <ConsumerRecord
<？，？>>，则忽略FilteringBatchMessageListenerAdapter，因为ConsumerRecords是不可变的。

Retrying Deliveries

如果侦听器抛出异常，则默认行为是调用ErrorHandler（如果已配置）或以其他方式记录。

提供了两个错误处理程序接口（ErrorHandler和BatchErrorHandler）。 您必须配置适当的类型以匹配消息侦听器。

为了重试传递，提供了一个方便的侦听器适配器RetryingMessageListenerAdapter。

您可以使用RetryTemplate和RecoveryCallback 对其进行配置 - 有关这些组件的信息，请参阅spring-retry项目。如果未提供恢复回调，则在重试耗尽后将向容器抛出异常。在这种情况下，如果已配置，则调用ErrorHandler，否则将记录。

使用@KafkaListener时，可以在容器工厂上设置RetryTemplate（以及可选的recoveryCallback）。执行此操作时，侦听器将包装在适当的重试适配器中。

传递给RecoveryCallback的RetryContext的内容取决于侦听器的类型。上下文始终具有记录属性，该记录属性是发生故障的记录。如果您的侦听器正在确认或消费者知晓，则可以使用其他确认或使用者属性。为方便起见，RetryingMessageListenerAdapter为这些键提供了静态常量。有关更多信息，请参阅其Javadoc。

没有为任何批处理消息侦听器提供重试适配器，因为框架不知道批处理中发生故障的位置。如果在使用批量侦听器时需要重试功能，我们建议您在侦听器本身中使用RetryTemplate。

Stateful Retry

您应该了解上一节中讨论的重试会暂停使用者线程（如果使用BackOffPolicy）。在重试期间没有调用Consumer.poll（）。卡夫卡有两个属性来确定消费者的健康状况。 session.timeout.ms用于确定使用者是否处于活动状态。从版本0.10.1.0开始，心跳在后台线程上发送，因此慢速消费者不再影响它。 max.poll.interval.ms（默认值：五分钟）用于确定消费者是否显示为挂起（从上次轮询处理记录花费的时间太长）。如果poll（）调用之间的时间超过此值，则代理将撤消分配的分区并执行重新平衡。对于冗长的重试序列，退避时，很容易发生这种情况。

从版本2.1.3开始，您可以通过将状态重试与SeekToCurrentErrorHandler结合使用来避免此问题。在这种情况下，每次传递尝试都会将异常抛回到容器中，错误处理程序会重新搜索未处理的偏移量，并且下一次poll（）会重新传递相同的消息。这避免了超出max.poll.interval.ms属性的问题（只要尝试之间的单个延迟不超过它）。因此，在使用ExponentialBackOffPolicy时，必须确保maxInterval小于max.poll.interval.ms属性。要启用有状态重试，可以使用带有状态布尔参数的RetryingMessageListenerAdapter构造函数（将其设置为true）。配置侦听器容器工厂（对于@KafkaListener）时，将工厂的statefulRetry属性设置为true。

Detecting Idle and Non-Responsive Consumers

虽然有效，但异步消费者的一个问题是检测它们何时空闲。 如果在一段时间内没有消息到达，您可能需要采取一些措施。

您可以将侦听器容器配置为在经过一段时间而没有消息传递时发布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<>(...);
    return container;
}

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

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

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

此外，如果代理无法访问，则消费者poll（）方法不会退出，因此不会收到任何消息，也无法生成空闲事件。 要解决此问题，如果轮询未在pollInterval属性的3x内返回，则容器会发布NonResponsiveConsumerEvent。 默认情况下，每个容器中每30秒执行一次此检查。 您可以通过在配置侦听器容器时在ContainerProperties中设置monitorInterval和noPollThreshold属性来修改此行为。 接收此类事件可让您停止容器，从而唤醒消费者以便终止。

Event Consumption

您可以通过实现ApplicationListener来捕获这些事件 - 可以是一般侦听器，也可以是缩小到仅接收此特定事件的侦听器。 您还可以使用Spring Framework 4.2中引入的@EventListener。

下一个示例将@KafkaListener和@EventListener组合到一个类中。 您应该了解应用程序侦听器获取所有容器的事件，因此如果要根据哪个容器空闲采取特定操作，则可能需要检查侦听器ID。 您也可以使用@EventListener条件来实现此目的。

有关事件属性的信息，请参阅事件

该事件通常在使用者线程上发布，因此与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) {
        ...
    }

}

事件侦听器查看所有容器的事件。 因此，在前面的示例中，我们根据侦听器ID缩小接收的事件。
由于为@KafkaListener创建的容器支持并发，因此实际容器名为id-n，其中n是每个实例的唯一值，以支持并发。
这就是我们在条件中使用startsWith的原因。

如果您希望使用idle事件来停止Lister容器，则不应在调用侦听器的线程上调用container.stop（）。
这样做会导致延迟和不必要的日志消息。 相反，您应该将事件移交给另一个可以阻止容器的线程。 此外，如果容器实例是子容器，则不应该停止（）。
您应该停止并发容器。

Current Positions when Idle

请注意，通过在侦听器中实现ConsumerSeekAware，可以在检测到空闲时获取当前位置。 请参阅`寻求特定偏移量中的onIdleContainer（）。

Topic/Partition Initial Offset

几种方法可以为分区设置初始偏移量。

手动分配分区时，可以在配置的TopicPartitionInitialOffset参数中设置初始偏移量（如果需要）（请参阅消息侦听器容器）。 您也可以随时寻找特定的偏移量。

当您使用代理分配分区的组管理时：

  • 对于新的group.id，初始偏移量由auto.offset.reset使用者属性（最早或最新）确定。

  • 对于现有组ID，初始偏移量是该组ID的当前偏移量。 但是，您可以在初始化期间（或之后的任何时间）寻找特定的偏移量。

Seeking to a Specific Offset

为了寻求，您的监听器必须实现ConsumerSeekAware，它具有以下方法：

void registerSeekCallback(ConsumerSeekCallback callback);

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

启动容器时调用第一个方法。 在初始化后的某个任意时间寻找时，您应该使用此回调。 您应该保存对回调的引用。 如果在多个容器（或ConcurrentMessageListenerContainer）中使用相同的侦听器，则应将回调存储在ThreadLocal或由侦听器Thread键入的其他一些结构中。

使用组管理时，在分配更改时调用第二种方法。 例如，您可以通过调用回调来使用此方法来设置分区的初始偏移量。 您必须使用回调参数，而不是传递给registerSeekCallback的参数。 如果您自己显式分配分区，则永远不会调用此方法。 在这种情况下使用TopicPartitionInitialOffset。

回调有以下方法：

void seek(String topic, int partition, long offset);

void seekToBeginning(String topic, int partition);

void seekToEnd(String topic, int partition);

当检测到空闲容器时，您还可以从onIdleContainer（）执行搜索操作。 有关如何启用空闲容器检测，请参阅检测空闲和非响应消费者。

要在运行时任意搜索，请使用registerSeekCallback中的回调引用来获取相应的线程。

Container factory

正如@KafkaListener Annotation中所讨论的，ConcurrentKafkaListenerContainerFactory用于为带注释的方法创建容器。

从2.2版开始，您可以使用同一工厂来创建任何ConcurrentMessageListenerContainer。 如果要创建具有类似属性的多个容器，或者希望使用某些外部配置的工厂（例如Spring Boot自动配置提供的工厂），这可能很有用。 创建容器后，可以进一步修改其属性，其中许多属性是使用container.getContainerProperties（）设置的。 以下示例配置ConcurrentMessageListenerContainer：

@Bean
public ConcurrentMessageListenerContainer<String, String>(
        ConcurrentKafkaListenerContainerFactory<String, String> factory) {

    ConcurrentMessageListenerContainer<String, String> container =
        factory.createContainer("topic1", "topic2");
    container.setMessageListener(m -> { ... } );
    return container;
}

以这种方式创建的容器不会添加到端点注册表中。 它们应该创建为@Bean定义，以便它们在应用程序上下文中注册。

Thread Safety

使用并发消息侦听器容器时，将在所有使用者线程上调用单个侦听器实例。因此，监听器需要是线程安全的，并且最好使用无状态监听器。如果无法使侦听器线程安全或添加同步会显着降低添加并发性的好处，则可以使用以下几种技术之一：

  • 使用并发= 1的n个容器和原型作用域MessageListener bean，以便每个容器都有自己的实例（使用@KafkaListener时这是不可能的）。

  • 将状态保留在ThreadLocal <？>实例中。

  • 让单例侦听器委托给在SimpleThreadScope（或类似范围）中声明的bean。

为了便于清理线程状态（对于前面列表中的第二项和第三项），从2.2版开始，侦听器容器在每个线程退出时发布ConsumerStoppedEvent。您可以使用ApplicationListener或@EventListener方法使用这些事件来从作用域中删除ThreadLocal <？>实例或remove（）线程范围的bean。请注意，SimpleThreadScope不会销毁具有销毁接​​口的bean（例如DisposableBean），因此您应该自己销毁（）实例。

默认情况下，应用程序上下文的事件multicaster在调用线程上调用事件侦听器。 如果更改多播程序以使用异步执行程序，则线程清理无效。

推荐阅读：
3. Spring Boot 2.x最佳实践之Spring for Apache Kafka集成

本节完~

本文标题：第三集 Spring for Apache Kafka 接受消息
本文链接：https://blog.quwenai.cn/post/3006.html
作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网 刊载发布。
版权声明：本文不使用任何协议授权，您可以任何形式自由转载或使用。

本文链接：https://blog.quwenai.cn/post/3006.html 

分享到：
本站所有内容均来自于互联网收集及用户投稿,如果想系统学习相关知识 建议在本博客【自学成才】栏目选择合适的课程进行系统化学习
-- 展开阅读全文 --