无法在 Spring Integration 中将订户连接到 BroadCastingDispatcher

Question

按照Bootiful GCP：与Google Cloud Pub/Sub集成(4/8(中的这个例子，我试图构建一个流程，从Google Pubsub订阅中读取并写入另一个主题。

在DEBUG模式下启动我的应用程序后，我可以看到消息来自Google PubSub但因此没有被"消耗"

o.s.i.dispatcher.BroadcastingDispatcher ： 没有订阅者，默认行为被忽略

非常感谢对此的任何帮助。

以下是我的主要代码的样子——

public class PubsubRouteBuilderService {
private final PubSubTemplate pubSubTemplate; // injected via Spring
public PubsubRouteBuilderService(PubSubTemplate pubSubTemplate) {
this.pubSubTemplate = pubSubTemplate;
}
public synchronized boolean buildRoute(PubsubRouteModel pubsubRouteModel) {
log.info("Building route for: {}", pubsubRouteModel);
buildPubsubRoute(pubsubRouteModel);
// some unrelated logic
return true;
}
private void buildPubsubRoute(PubsubRouteModel pubsubRouteModel) {
final StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows.from(
RouteBuilderFactory
.messageChannelAdapter(
RouteBuilderFactory.getMessageChannel(),
pubSubTemplate,
pubsubRouteModel.getFromSub()))
.handle(
message -> {
log.info("consumed new message: [" + message.getPayload() + "]");
AckReplyConsumer consumer = message.getHeaders()
.get(GcpPubSubHeaders.ORIGINAL_MESSAGE, AckReplyConsumer.class);
consumer.ack();
})
.get();
standardIntegrationFlow.start();
}
}

以下是RouteBuilderFactory中的其他方法，如下所示-

public static MessageChannel getMessageChannel() {
return MessageChannels.publishSubscribe().get();
}
public static PubSubInboundChannelAdapter messageChannelAdapter(MessageChannel inputChannel, PubSubTemplate pubSubTemplate, String channelName) {
PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, channelName);
adapter.setOutputChannel(inputChannel);
adapter.setAckMode(AckMode.MANUAL);
return adapter;
}

Answer 1

你的代码似乎根本不是基于那篇博客文章......

private void buildPubsubRoute(PubsubRouteModel pubsubRouteModel) {
final StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows.from(
RouteBuilderFactory
.messageChannelAdapter(
RouteBuilderFactory.getMessageChannel(),
pubSubTemplate,
pubsubRouteModel.getFromSub()))
.handle(
message -> {
log.info("consumed new message: [" + message.getPayload() + "]");
AckReplyConsumer consumer = message.getHeaders()
.get(GcpPubSubHeaders.ORIGINAL_MESSAGE, AckReplyConsumer.class);
consumer.ack();
})
.get();
standardIntegrationFlow.start();
}

你不能只是"启动"一些任意IntegrationFlow对象 - 它必须由 Spring 管理(声明为@Bean(。

框架在幕后构建了一些基础设施，以使所有这些工作。