【发布时间】：2019-08-06 10:13:23
【问题描述】：
我正在尝试为单线程处理程序定义一个流程。消息数量众多，处理程序很慢（一一处理效率低下）。所以我想让处理程序使用 Java DSL 一次性使用通道中可用的所有消息（或等到一些消息积累）。如果通道中没有消息并且处理程序已经处理了前一个组，它应该等待一段时间（超时“a”），以便在通道中积累一些消息。但是如果消息不断出现，处理程序必须在上一次执行后的一段时间后使用它们（超时“b”）。因此，处理程序执行之间的时间间隔不应超过“b”（除非没有消息到达通道）。
没有理由为这种处理程序创建多个实例：它为接口生成数据。下面的代码描述了一些基本配置。我的问题是，一旦处理程序执行完成，我无法想出去抖动（超时“b”）和释放组。
@Configuration
public class SomeConfig {
private AtomicBoolean someHandlerBusy = new AtomicBoolean(false);
@Bean
StandardIntegrationFlow someFlow() {
return IntegrationFlows
.from("someChannel")
.aggregate(aggregatorSpec -> aggregatorSpec
//The only rule to release a group:
//wait 500ms after last message and have a free someHandler
.groupTimeout(500)
.sendPartialResultOnExpiry(true) //if 500ms expired - send group
.expireGroupsUponCompletion(true) //group should be filled again
.correlationStrategy(message -> true) //one group key, all messages in oe group
.releaseStrategy(message -> false) //never release messages, only with timeout
//Send messages one by one. This is not part of this task.
//I just want to know how to do that. Like splitter.
//.outputProcessor(MessageGroup::getMessages)
)
.handle("someHandler")
.get();
}
}
我有纯 Java (kotlin) 代码的解决方案：https://pastebin.com/mti3Y5tD
更新
下面的配置不会删除组。该小组正在不断壮大，但最终却犯了错误。
错误：
*** java.lang.instrument ASSERTION FAILED ***: "!errorOutstanding" with message transform method call failed at JPLISAgent.c line: 844
配置：
@Configuration
public class InterfaceHandlerConfigJava {
@Bean
MessageChannel interfaceAggregatorFlowChannel() {
return MessageChannels.publishSubscribe("interfaceAggregatorFlowChannel").get();
}
@EventListener(ApplicationReadyEvent.class)
public void initTriggerPacket(ApplicationReadyEvent event) {
MessageChannel channel = event.getApplicationContext().getBean("interfaceAggregatorFlowChannel", MessageChannel.class);
channel.send(MessageBuilder.withPayload(new InterfaceHandler.HandlerReadyMessage()).build());
}
@Bean
StandardIntegrationFlow someFlow(
InterfaceHandler interfaceHandler
) {
long lastMessageTimeout = 10L;
return IntegrationFlows
.from("interfaceAggregatorFlowChannel")
.aggregate(aggregatorSpec -> aggregatorSpec
.groupTimeout(messageGroup -> {
if (haveInstance(messageGroup, InterfaceHandler.HandlerReadyMessage.class)) {
System.out.println("case HandlerReadyMessage");
if (haveInstance(messageGroup, DbChangeStreamConfiguration.InitFromDbMessage.class)) {
System.out.println("case InitFromDbMessage");
return 0L;
} else if (messageGroup.size() > 1) {
long groupCreationTimeout =
messageGroup.getTimestamp() + 500L - System.currentTimeMillis();
long timeout = Math.min(groupCreationTimeout, lastMessageTimeout);
System.out.println("case messageGroup.size() > 1, timeout: " + timeout);
return timeout;
}
}
System.out.println("case Handler NOT ReadyMessage");
return null;
})
.sendPartialResultOnExpiry(true)
.expireGroupsUponCompletion(true)
.expireGroupsUponTimeout(true)
.correlationStrategy(message -> true)
.releaseStrategy(message -> false)
)
.handle(interfaceHandler, "handle")
.channel("interfaceAggregatorFlowChannel")
.get();
}
private boolean haveInstance(MessageGroup messageGroup, Class clazz) {
for (Message<?> message : messageGroup.getMessages()) {
if (clazz.isInstance(message.getPayload())) {
return true;
}
}
return false;
}
}
我想强调：这个流程是循环的。有 IN 没有 OUT。消息进入 IN 但处理程序最后发出 HandlerReadyMessage。 也许应该有一些线程中断通道？
最终变体
由于聚合器和处理程序不应相互阻塞，也不应尝试产生 stackoverflow 异常，它们应该在不同的线程中运行。在上面的配置中，这是通过队列通道实现的。看起来发布-订阅频道没有在不同的线程中运行订阅者（至少对于一个订阅者而言）。
@Configuration
public class InterfaceHandlerConfigJava {
// acts as thread breaker too
@Bean
MessageChannel interfaceAggregatorFlowChannel() {
return MessageChannels.queue("interfaceAggregatorFlowChannel").get();
}
@Bean
MessageChannel threadBreaker() {
return MessageChannels.queue("threadBreaker").get();
}
@EventListener(ApplicationReadyEvent.class)
public void initTriggerPacket(ApplicationReadyEvent event) {
MessageChannel channel = event.getApplicationContext().getBean("interfaceAggregatorFlowChannel", MessageChannel.class);
channel.send(MessageBuilder.withPayload(new InterfaceHandler.HandlerReadyMessage()).build());
}
@Bean
StandardIntegrationFlow someFlow(
InterfaceHandler interfaceHandler
) {
long lastMessageTimeout = 10L;
return IntegrationFlows
.from("interfaceAggregatorFlowChannel")
.aggregate(aggregatorSpec -> aggregatorSpec
.groupTimeout(messageGroup -> {
if (haveInstance(messageGroup, InterfaceHandler.HandlerReadyMessage.class)) {
System.out.println("case HandlerReadyMessage");
if (haveInstance(messageGroup, DbChangeStreamConfiguration.InitFromDbMessage.class)) {
System.out.println("case InitFromDbMessage");
return 0L;
} else if (messageGroup.size() > 1) {
long groupCreationTimeout =
messageGroup.getTimestamp() + 500L - System.currentTimeMillis();
long timeout = Math.min(groupCreationTimeout, lastMessageTimeout);
System.out.println("case messageGroup.size() > 1, timeout: " + timeout);
return timeout;
}
}
System.out.println("case Handler NOT ReadyMessage");
return null;
})
.sendPartialResultOnExpiry(true)
.expireGroupsUponCompletion(true)
.expireGroupsUponTimeout(true)
.correlationStrategy(message -> true)
.releaseStrategy(message -> false)
.poller(pollerFactory -> pollerFactory.fixedRate(1))
)
.channel("threadBreaker")
.handle(interfaceHandler, "handle", spec -> spec.poller(meta -> meta.fixedRate(1)))
.channel("interfaceAggregatorFlowChannel")
.get();
}
private boolean haveInstance(MessageGroup messageGroup, Class clazz) {
for (Message<?> message : messageGroup.getMessages()) {
if (clazz.isInstance(message.getPayload())) {
return true;
}
}
return false;
}
}
【问题讨论】：
标签： spring-integration spring-integration-dsl