类上的 @KafkaListener

Doubao 中英对照 @KafkaListener on a Class

当你在类级别使用 @KafkaListener 时，必须在方法级别指定 @KafkaHandler。如果该类或其子类的任何方法上都没有 @KafkaHandler，框架将拒绝这样的配置。@KafkaHandler 注解用于明确而简洁地表达方法的用途。否则，在没有额外限制的情况下，很难对这个方法或其他方法做出决定。

当消息被投递时，会使用转换后的消息 payload 类型来决定调用哪个方法。以下示例展示了如何这样做：

@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 方法时，payload 必须已经被转换为领域对象（以便可以进行匹配）。使用自定义反序列化器、JacksonJsonDeserializer，或将其 TypePrecedence 设置为 TYPE_ID 的 JacksonJsonMessageConverter。有关更多信息，请参阅序列化、反序列化与消息转换。

important

由于 Spring 解析方法参数的方式存在一些限制，默认的 @KafkaHandler 不能接收离散的 header；它必须使用Consumer 记录元数据中讨论的 ConsumerRecordMetadata。

例如：

@KafkaHandler(isDefault = true)
public void listenDefault(Object object, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    ...
}

如果 object 是 String，这将不起作用；topic 参数也会得到对 object 的引用。

如果你需要在默认方法中获得关于记录的元数据，请使用如下方式：

@KafkaHandler(isDefault = true)
void listen(Object in, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta) {
    String topic = meta.topic();
    ...
}

另外，这同样也不起作用。topic 被解析为 payload。

@KafkaHandler(isDefault = true)
public void listenDefault(String payload, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
    // payload.equals(topic) is True.
    ...
}

如果存在需要在默认方法中使用离散自定义 header 的用例，请使用如下方式：

@KafkaHandler(isDefault = true)
void listenDefault(String payload, @Headers Map<String, Object> headers) {
    Object myValue = headers.get("MyCustomHeader");
    ...
}