0

我们正在尝试从 jdbc 源进行轮询，聚合消息，然后将聚合器的输出发送到 kafka 流和 jdbcMessageHandler （以更新我们轮询的行，以便它们不再被轮询。实际上是单独的表格）。我们正在使用 IntegrationFlow DSL。轮询和聚合工作正常。多个输出/源/通道/处理程序的输出不起作用。

我们当前的流程如下所示：

IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(jdbcMessageSource(), new Consumer<SourcePollingChannelAdapterSpec>(){
            @Override
            public void accept(SourcePollingChannelAdapterSpec sourcePollingChannelAdapterSpec) {
                sourcePollingChannelAdapterSpec.poller(poller);
            }
        })

    .split()
    .transform((message)-> {
        // do our stuff; output is a simple POJO representing a single row from the db
    })
    .aggregate(aggregator -> aggregator) // details spared but it works fine; output is a larger POJO containing a collection of the row objects
    .channel(this.source.output()) // this goes to kafka
    .handle(jdbcMessageHandler()); // doesn't get here

这是我们的投票来源：

@Bean
public MessageSource<Object> jdbcMessageSource() {
    JdbcPollingChannelAdapter jdbcPollingChannelAdapter = new JdbcPollingChannelAdapter(this.dataSource, this.properties.getQuery());
      jdbcPollingChannelAdapter.setSelectSqlParameterSource(createSqlParameterSource());

    return jdbcPollingChannelAdapter;
}

jdbcMessageHandler 就是这样：

@Bean
public MessageHandler jdbcMessageHandler() {
    return new JdbcMessageHandler(dataSource, this.properties.getUpdate());
}

this.output 就是这样：

@Autowired
private Source source;

它通过 @EnableBinding(Source.class) 映射到我们的 application.yml 中的 kafka 主题：

spring:
  cloud:
    stream:
      bindings:
        output: our.topic

我们的大部分属性都定义在 application.yml 文件中，我们使用这些属性和注释而不是 XMl 配置。

有了上面的内容，它可以很好地写入 kafka，但没有到达 jdbcMessageHandler。

我这样做的时候工作过（在聚合器之后）：

.publishSubscribeChannel(publishSubscribeSpec -> publishSubscribeSpec
                .subscribe(flow -> flow
                        .handle(jdbcMessageHandler()))
        )
.channel(this.source.output())

但这是错误的顺序；我们要确保首先将消息写入 kafka，然后更新跟踪已成功轮询哪些行的表。

jdbcMessageHandler 简单地包装了一个使用来自消息的值的 INSERT 语句，所以我假设有多种方法可以做到这一点。jdbcOutboundGateway 是一种方式吗？看起来这是为了进行另一个查询并返回一个结果以进行进一步处理，这不适合我们的用例。

还建议使用轮询来更新，因为我们使用轮询作为我们的来源。我调查了这个。我认为这也行不通，因为它似乎在轮询后立即进行更新，甚至在轮询结果被处理之前，所以在最终聚合消息发送到 kafka 之前进行更新的时间也有同样的问题.

编辑：我尝试了下面的答案，所以当前流程是这样的：

@Bean
public IntegrationFlow pollingFlow() {
    IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(jdbcMessageSource(), new Consumer<SourcePollingChannelAdapterSpec>(){
                @Override
                public void accept(SourcePollingChannelAdapterSpec sourcePollingChannelAdapterSpec) {
                    sourcePollingChannelAdapterSpec.poller(poller);
                }
            })

    .split()
    .transform((message)-> {
        // transform ResultSet; output is a simple POJO representing a single row from the db
    })
    .aggregate(aggregator -> aggregator) // details spared but it works fine; output is a larger POJO containing a collection of the row objects
    .publishSubscribeChannel(publishSubscribeSpec -> publishSubscribeSpec
            .subscribe(flow -> flow
                    .handle(this.source.output()))
            .subscribe(flow -> flow
                    .handle(jdbcMessageHandler()))
    );
}

并在启动时收到以下错误：

java.lang.IllegalArgumentException: Found ambiguous parameter type [interface org.springframework.messaging.MessageHandler] for method match: [public boolean org.springframework.integration.channel.AbstractSubscribableChannel.unsubscribe(org.springframework.messaging.MessageHandler), public void org.springframework.integration.channel.AbstractMessageChannel.configureMetrics(org.springframework.integration.support.management.AbstractMessageChannelMetrics), public void org.springframework.integration.channel.AbstractMessageChannel.setDatatypes(java.lang.Class[]), public org.springframework.messaging.support.ChannelInterceptor org.springframework.integration.channel.AbstractMessageChannel.removeInterceptor(int), public void org.springframework.integration.context.IntegrationObjectSupport.setApplicationContext(org.springframework.context.ApplicationContext) throws org.springframework.beans.BeansException, public boolean org.springframework.integration.channel.AbstractMessageChannel.removeInterceptor(org.springframework.messaging.support.ChannelInterceptor), public java.lang.String org.springframework.integration.context.IntegrationObjectSupport.getComponentName(), public void org.springframework.integration.channel.AbstractMessageChannel.setStatsEnabled(boolean), public void org.springframework.integration.channel.AbstractMessageChannel.setMessageConverter(org.springframework.messaging.converter.MessageConverter), public void org.springframework.integration.context.IntegrationObjectSupport.setMessageBuilderFactory(org.springframework.integration.support.MessageBuilderFactory), public void org.springframework.integration.context.IntegrationObjectSupport.setBeanFactory(org.springframework.beans.factory.BeanFactory), public void org.springframework.integration.context.IntegrationObjectSupport.setComponentName(java.lang.String), public void org.springframework.integration.channel.AbstractMessageChannel.setInterceptors(java.util.List), public final void org.springframework.integration.context.IntegrationObjectSupport.setPrimaryExpression(org.springframework.expression.Expression), public void org.springframework.integration.context.IntegrationObjectSupport.setChannelResolver(org.springframework.messaging.core.DestinationResolver)]
at org.springframework.util.Assert.isNull(Assert.java:113)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.findHandlerMethodsForTarget(MessagingMethodInvokerHelper.java:499)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:226)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:149)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:144)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.<init>(MethodInvokingMessageProcessor.java:60)
    at org.springframework.integration.handler.ServiceActivatingHandler.<init>(ServiceActivatingHandler.java:37)
    at org.springframework.integration.dsl.IntegrationFlowDefinition.handle(IntegrationFlowDefinition.java:985)
    at org.springframework.integration.dsl.IntegrationFlowDefinition.handle(IntegrationFlowDefinition.java:964)
    at org.springframework.integration.dsl.IntegrationFlowDefinition.handle(IntegrationFlowDefinition.java:950)

我调试并进入了 IntegrationFlowDefinition 上的 .handle() 方法，发现传递的对象以某种方式转换为 DirectChannel，我不明白。

4

1 回答 1

1

将两个子流订阅到 pub/sub 通道。

.publishSubscribeChannel(publishSubscribeSpec -> publishSubscribeSpec
            .subscribe(flow -> flow
                    .bridge(e -> e.id("bTO"))
                    .channel(this.source.output())
            .subscribe(flow -> flow
                    .handle(jdbcMessageHandler()))
    )

您需要.bridge()将发布/订阅通道桥接到输出通道。

于 2017-09-22T15:58:02.387 回答