0
Ich bin Wolke zu springen und suchen unsere Monostruktur zu Mikro Dienste zu ändern, dies zunächst gesagt, was ich versuche, jetzt zu tun, wird die folgendeAggregate Nachrichten im Frühjahr Wolke Stream-
- Erhalten Sie Anfragen aufzurufen ein Web-Service (externes System) aus verschiedenen Quellen. Zu jeder bestimmten Zeit kann dies 1 Anfrage oder bis zu 100K Anfrage sein.
- Das externe System unterstützt Bulk, so ist es besser, wenn ich die Nachrichten aggregieren und sie in großen Mengen senden kann. Zum Beispiel so lange aggregieren, bis entweder der Nummernschwellenwert erreicht ist (100 Nachrichten) oder der Zeitgrenzwert von 2 Sekunden erreicht ist.
- Auch wenn ich einen Fehler empfangen mag ich exponentiell
Meine erste Idee zurückzuziehen ist einen Prozessor, bevor mein Sink zu schaffen, die die obene Aggregation tun.
Ist das die richtige Art zu denken im Cloud Computing oder ist das ein anderer Weg?
Arbeitslösung
@EnableBinding(Processor.class)
class Configuration {
@Autowired
Processor processor;
@ServiceActivator(inputChannel = Processor.INPUT)
@Bean
public MessageHandler aggregator() {
AggregatingMessageHandler aggregatingMessageHandler =
new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
new SimpleMessageStore(10));
//AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
//aggregatorFactoryBean.setMessageStore();
aggregatingMessageHandler.setOutputChannel(processor.output());
//aggregatorFactoryBean.setDiscardChannel(processor.output());
aggregatingMessageHandler.setSendPartialResultOnExpiry(true);
aggregatingMessageHandler.setSendTimeout(1000L);
aggregatingMessageHandler.setCorrelationStrategy(new ExpressionEvaluatingCorrelationStrategy("'FOO'"));
aggregatingMessageHandler.setReleaseStrategy(new MessageCountReleaseStrategy(3)); //ExpressionEvaluatingReleaseStrategy("size() == 5")
aggregatingMessageHandler.setExpireGroupsUponCompletion(true);
aggregatingMessageHandler.setGroupTimeoutExpression(new ValueExpression<>(3000L)); //size() ge 2 ? 5000 : -1
aggregatingMessageHandler.setExpireGroupsUponTimeout(true);
return aggregatingMessageHandler;
}
}