Spring Cloud Stream (Kafka) 参数化指定的错误通道 {destination}.{group}.errors

Question

我正在尝试查看我传递给@ServiceActivator 的错误通道是否可以bounded/parameterized 引用 YAML 中指定的值，而不是在代码本身中硬编码实际目标和使用者组。

@ServiceActivator(
        // I do not want to hardcode destination and consumer group here
        inputChannel = "stream-test-topic.my-consumer-group.errors"
    )
    public void handleError(ErrorMessage errorMessage) {
        // Getting exception objects
        Throwable errorMessagePayload = errorMessage.getPayload();
        log.error("exception occurred", errorMessagePayload);

        // Get message body
        Message<?> originalMessage = errorMessage.getOriginalMessage();
        if (originalMessage != null) {
            log.error("Message Body: {}", originalMessage.getPayload());
        } else {
            log.error("The message body is empty");
        }
    }

Answer 1

你不能用 @ServiceActivator 做到这一点；请改用 Java DSL：

@Value("${error.channel}")
String errors;

@Bean
public IntegrationFlow flow() {
    return IntegrationFlows.from(this.errors)
            .handle(msg -> {
                System.out.println(msg);
            })
            .get();
}

并设置

error:
  channel: stream-test-topic.my-consumer-group.errors

Spring Cloud Stream (Kafka) 参数化指定的错误通道 {destination}.{group}.errors

Spring Cloud Stream (Kafka) parameterize specified error channel {destination}.{group}.errors

java

spring

kafka-consumer-api

spring-cloud-stream

spring-kafka