如何将此弹簧集成配置从 XML 转换为 Java？

Question

这个特定的部分在应用程序而不是 XML 中实现是有意义的，因为它在整个集群中是一个常量，而不是本地化到单个作业。

通过剖析 XSD，在我看来，int-kafka:outbound-channel-adapter 的 xml 构造了一个 KafkaProducerMessageHandler。

没有可见的方法来设置频道、主题或大多数其他属性。

注意潜在的反对者 - （咆哮）我已经使用 RTFM 一周了，并且比我开始时更加困惑。我对语言的选择已经从形容词到副词毕业，我开始从其他语言借词。答案可能就在里面。但如果是这样，它就不是普通人能找到的。 （吐槽）

XML 配置：

<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
                                    kafka-template="kafkaTemplate"
                                    auto-startup="false"
                                    channel="outbound-staging"
                                    topic="foo"
                                    sync="false"
                                    message-key-expression="'bar'"
                                    send-failure-channel="failures"
                                    send-success-channel="successes"
                                    partition-id-expression="2">
</int-kafka:outbound-channel-adapter>

如果是这样，那么我希望 java 配置看起来像这样：

@Bean
public KafkaProducerMessageHandler kafkaOutboundChannelAdapter () {
    KafkaProducerMessageHandler result = new KafkaProducerMessageHandler(kafkaTemplate());

    result.set????? ();    // WTH?? No methods for most of the attributes?!!!

    return result;
}

编辑：关于正在解决的高级问题的附加信息

作为一个更大项目的一部分，我正在尝试实现来自 https://docs.spring.io/spring-batch/4.0.x/reference/html/spring-batch-integration.html#remote-partitioning 的教科书示例，使用 Kafka 支持而不是 JMS 支持。

我相信最终的集成流程应该是这样的：

partitionHandler -> messingTemplate -> outbound-requests (DirectChannel) -> outbound-staging (KafkaProducerMessageHandler) -> kafka

kafka -> executionContainer (KafkaMessageListenerContainer) -> inboundKafkaRequests (KafkaMessageDrivenChannelAdapter) -> inbound-requests (DirectChannel) -> serviceActivator (StepExecutionRequestHandler)

serviceActivator (StepExecutionRequestHandler) -> 回复-staging (KafkaProducerMessageHandler) -> kafka

kafka -> replyContainer (KafkaMessageListenerContainer) -> inboundKafkaReplies (KafkaMessageDrivenChannelAdapter) -> inbound-replies (DirectChannel) -> partitionhandler

Answer 1

不知道你的意思是他们错过了，但这是我在KafkaProducerMessageHandler的源代码中看到的：

public void setTopicExpression(Expression topicExpression) {
    this.topicExpression = topicExpression;
}

public void setMessageKeyExpression(Expression messageKeyExpression) {
    this.messageKeyExpression = messageKeyExpression;
}

public void setPartitionIdExpression(Expression partitionIdExpression) {
    this.partitionIdExpression = partitionIdExpression;
}

/**
 * Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record.
 * The resulting value should be a {@link Long} type representing epoch time in milliseconds.
 * @param timestampExpression the {@link Expression} for timestamp to wait for result
 * fo send operation.
 * @since 2.3
 */
public void setTimestampExpression(Expression timestampExpression) {
    this.timestampExpression = timestampExpression;
}

等等。

您还可以访问超类设置器，例如用于您的 XML 变体的 setSync()。

input-channel 不是MessageHandler 的责任。它转到Endpoint，可以通过@ServiceActivator 和@Bean 进行配置。

在 Core Spring 集成参考手册中查看更多信息：https://docs.spring.io/spring-integration/reference/html/#annotations_on_beans

开头还有很重要的一章：https://docs.spring.io/spring-integration/reference/html/#programming-tips

此外，最好考虑使用 Java DSL 而不是直接使用MessageHandler：

             Kafka
                .outboundChannelAdapter(producerFactory)
                .sync(true)
                .messageKey(m -> m
                        .getHeaders()
                        .get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
                .headerMapper(mapper())
                .partitionId(m -> 0)
                .topicExpression("headers[kafka_topic] ?: '" + topic + "'")
                .configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic))
                .get();

在提到的 Spring 集成文档中查看有关 Java DSL 的更多信息：https://docs.spring.io/spring-integration/reference/html/#java-dsl