Spring Интеграция HTTP для Scatter Gather

Я новичок в Spring Integration и пытаюсь использовать корпоративный шаблон scatter-gather, но я борюсь с деталями реализации и борюсь с доступными примерами, которые я могу найти в Интернете.

Короче мой сценарий:

  1. HTTP-запрос отправляется от пользователя в систему A.
  2. Перед ответом (также известным как синхронный) система A асинхронно отправляет N сообщений на N систем X.
  3. Система А сидит и ждет ответов.
  4. После получения ответа от каждой из систем запроса система А объединяет ответы в один более крупный ответ.
  5. Наконец, система А отвечает пользователю большим ответом.

По сути, с точки зрения первоначального потребителя, сингл — это запрос, который отвечает ответом, без необходимости «вернуться позже». Однако на самом деле этот запрос был направлен на фасад, который маскирует скрытую за ним сложность (потенциально затрагивая сотни систем, делая синхронные запросы на серверной части неэффективными и невыполнимыми).

Пока что у меня есть эта реализация (детали очищены, поэтому они могут не быть примером 1: 1 того, с чем я играю, например, корреляционная стратегия, которую я с тех пор разработал, не делает то, что я ожидал):

@Bean
public IntegrationFlow overallRequest(final AmqpTemplate amqpTemplate) {

  return IntegrationFlows.from( // HTTP endpoint to user makes requests on
          Http.inboundChannelAdapter("/request-overall-document")
              .requestMapping(m -> m.methods(HttpMethod.POST))
              .requestPayloadType(String.class))
      .log()
      // Arbitrary header to simplify example, realistically would generate a UUID
      // and attach to some correlating header that works for systems involved
      .enrichHeaders(p -> p.header("someHeader", "someValue"))
      .log()
      .scatterGather(
          recipientListRouterSpec ->
              recipientListRouterSpec
                  .applySequence(true)
                  .recipientFlow(
                      flow ->
                          flow.handle( // Straight pass through of msg received to see in response
                              Amqp.outboundAdapter(amqpTemplate)
                                  .exchangeName( // RabbitMQ fanout exchange to N queues to N systems
                                      "request-overall-document-exchange"))),
          aggregatorSpec ->
              aggregatorSpec
                  // Again for example, arbitrary once two correlated responses
                  .correlationStrategy(msg -> msg.getHeaders().get("someHeader"))
                  .releaseStrategy(gm -> gm.size() == 2)
                  // Simple string concatenation for overall response
                  .outputProcessor(
                      msgrp ->
                          msgrp.getMessages().stream()
                              .map(msg -> msg.getPayload().toString())
                              .reduce("Overall response: ", (nexus, txt) -> nexus + "|" + txt))
                  // Reset group on each response
                  .expireGroupsUponCompletion(true),
          scatterGatherSpec ->
              scatterGatherSpec.gatherChannel(
                  responseChannel())) // The channel to listen for responses to request on
      .log()
      .get();
}

С этим в качестве конфигурации канала ответа:

@Bean
public MessageChannel responseChannel() {
  return new QueueChannel();
}

@Bean
public AmqpInboundChannelAdapter responseChannelAdapter(
    SimpleMessageListenerContainer listenerContainer,
    @Qualifier("responseChannel") MessageChannel channel) {
  AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
  adapter.setOutputChannel(channel);
  return adapter;
}

@Bean
public SimpleMessageListenerContainer responseContainer(ConnectionFactory connectionFactory) {
  SimpleMessageListenerContainer container =
      new SimpleMessageListenerContainer(connectionFactory);
  container.setQueueNames("request-overall-document-responses");
  return container;
}

Все ответы отправляются в отдельное приложение Spring, которое просто перенаправляет полезные данные запроса обратно (то есть для тестирования без необходимости интеграции с реальными системами):

@Bean
public IntegrationFlow systemOneReception(final ConnectionFactory connectionFactory, final AmqpTemplate amqpTemplate) {
  return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "request-overall-document-system-1"))
      .log()
      .handle(Amqp.outboundAdapter(amqpTemplate).routingKey("request-overall-document-responses"))
      .get();
}

@Bean
public IntegrationFlow systemTwoReception(final ConnectionFactory connectionFactory, final AmqpTemplate amqpTemplate) {
  return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "request-overall-document-system-2"))
      .log()
      .handle(Amqp.outboundAdapter(amqpTemplate).routingKey("request-overall-document-responses"))
      .get();
}

И я получаю следующую ошибку в системе A при успешном выпуске в соответствии со стратегией агрегации/выпуска в реализации разброса-сбора:

2020-02-29 20:06:39.255 ERROR 152 --- [ask-scheduler-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: The 'gatherResultChannel' header is required to deliver the gather result., failedMessage=GenericMessage [payload=Overall response: |somerequesttobesent|somerequesttobesent, headers={amqp_receivedDeliveryMode=PERSISTENT, content-length=19, amqp_deliveryTag=2, sequenceSize=1, amqp_redelivered=false, amqp_contentEncoding=UTF-8, host=localhost:18081, someHeader=someValue, connection=keep-alive, correlationId=182ee203-85ab-9ef6-7b19-3a8e2da8f5a7, id=994a0cf5-ad2b-02c3-dc93-74fae2f5092b, cache-control=no-cache, contentType=text/plain, timestamp=1583006799252, http_requestMethod=POST, sequenceNumber=1, amqp_consumerQueue=request-overall-document-responses, accept=*/*, amqp_receivedRoutingKey=request-overall-document-responses, amqp_timestamp=Sat Feb 29 20:06:39 GMT 2020, amqp_messageId=3341deae-7ed0-a042-0bb7-d2d2be871165, http_requestUrl=http://localhost:18081/request-overall-document, amqp_consumerTag=amq.ctag-ULxwuAjp8ZzcopBZYvcbZQ, accept-encoding=gzip, deflate, br, user-agent=PostmanRuntime/7.22.0}]
at org.springframework.integration.scattergather.ScatterGatherHandler.lambda$doInit$2(ScatterGatherHandler.java:160)
at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:77)
at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:71)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:431)
at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:284)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:265)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:223)
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.completeGroup(AbstractCorrelatingMessageHandler.java:823)
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:475)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:143)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:390)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:329)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:277)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:274)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Теперь я понимаю, что у меня есть несколько пробелов, но я изо всех сил пытаюсь понять, как двигаться дальше:

  1. Данная ошибка: нет вывода 'gatherResultChannel'. Я бы подумал, что это будут последующие «дескрипторы» / «журналы» / w.e в результате вызова scatterGather(...), но не так, чтобы поиграть.
  2. Должна быть какая-то форма сопоставления результата агрегации разброса-сбора обратно с исходным запросом Http.XXX.

EDIT: при дальнейшем копании проблема, похоже, заключается в том, что при выходе через AMQP (в моем случае RabbitMQ) рассматриваемый заголовок преднамеренно удален, поскольку это MessageChannel (см. строки с 230 по 257). Не уверен, что здесь подразумевается, что разделение/агрегация не предназначено для пересечения нескольких независимых приложений (я предполагаю, что оно отброшено, потому что это экземпляр объекта Java, который было бы проблематично передать)...

ДАЛЬНЕЙШЕЕ РЕДАКТИРОВАНИЕ: свежим взглядом заметил кое-что, чего раньше не замечал, за исключением того, что я вставил в кавычки сообщение о сбое, и, кажется, это явный результат обработки вывода (пока возился, переключался между DirectChannel и QueueChannel, только DirectChannel не печатает полезную нагрузку, поэтому не искал). Чтобы убедиться, что он не выполняет какое-то клонирование или что-то странное, обновил службу-заглушку, чтобы преобразовать и добавить уникальные постфиксы (как показано ниже), и да, это действительно агрегирование.

   .transform(msg -> MessageFormat.format("{0}_system1response", msg))
   .transform(msg -> MessageFormat.format("{0}_system2response", msg))

The 'gatherResultChannel' header is required to deliver the gather result., failedMessage=GenericMessage [payload=Overall response: |sometext_system2response|sometext_system1response, hea...

Итак, кажется, что рассеяние, сбор и агрегация работают, единственное, чего нет, это то, что данная обработка не знает, куда после этого отправлять сообщения?

ЕЩЕ РАЗ: Согласно ответу Гэри, все адаптеры заменены на шлюзы, однако при этом больше нельзя разветвляться? Поэтому удален аргумент scatterGatherSpec из вызова scatterGather и заменен/добавлен в двух получателях следующим образом:

.recipientFlow(flow -> flow.handle(Amqp.asyncOutboundGateway(asyncTemplate).routingKeyFunction(m -> "request-overall-document-system-1"), e -> e.id("sytemOneOutboundGateway")))
.recipientFlow(flow -> flow.handle(Amqp.asyncOutboundGateway(asyncTemplate).routingKeyFunction(m -> "request-overall-document-system-2"), e -> e.id("sytemTwoOutboundGateway")))

который является самым близким к рабочему примеру, который я могу получить, однако, хотя это выполняет какую-то работу, это приводит к многократной повторной обработке сообщений в очередях / вне очередей, где мой ожидаемый результат для POST с «msgtosend» был бы:

Overall message: |msgtosend_system1response|msgtosend_system2response

Вместо этого я получаю спорадические результаты, такие как:

Overall message: |msgtosend|msgtosend_system1response
Overall message: |msgtosend_system2response|msgtosend_system1response_system1response
Overall message: |msgtosend|msgtosend_system1response_system1response
Overall message: |msgtosend_system2response|msgtosend_system1response_system1response

Я предполагаю, что есть некоторое перекрытие config/bean, но как бы я ни старался, я не могу изолировать, что это такое, то есть фабрика соединений, контейнер прослушивателя, асинхронный шаблон и т. д. и т. д.


http events spring spring-integration enterprise-integration
person It's an account    schedule 29.02.2020    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Используйте исходящий шлюз AMQP вместо адаптеров исходящего и входящего каналов; таким образом заголовок канала будет сохранен. Существует AsyncAmqpOutboundGateway, который, вероятно, лучше всего подходит для ваших целей.

Если по какой-либо причине вам необходимо использовать адаптеры каналов, используйте средство обогащения заголовков вместе с Реестр каналов заголовков, чтобы преобразовать канал в строковое представление, чтобы его можно было сохранить.

ИЗМЕНИТЬ

Вот простой пример:

@SpringBootApplication
public class So60469260Application {

    public static void main(String[] args) {
        SpringApplication.run(So60469260Application.class, args);
    }

    @Bean
    public IntegrationFlow flow(AsyncRabbitTemplate aTemp) {
        return IntegrationFlows.from(Gate.class)
                .enrichHeaders(he -> he.headerExpression("corr", "payload"))
                .scatterGather(rlr -> rlr
                        .applySequence(true)
                        .recipientFlow(f1 -> f1.handle(Amqp.asyncOutboundGateway(aTemp)
                                .routingKey("foo")))
                        .recipientFlow(f2 -> f2.handle(Amqp.asyncOutboundGateway(aTemp)
                                .routingKey("bar"))),
                        agg -> agg.correlationStrategy(msg -> msg.getHeaders().get("corr")))
                .get();
    }

    @Bean
    public AsyncRabbitTemplate aTemp(RabbitTemplate template) {
        return new AsyncRabbitTemplate(template);
    }

    @Bean
    @DependsOn("flow")
    public ApplicationRunner runner(Gate gate) {
        return args -> System.out.println(gate.doIt("foo"));
    }

    @RabbitListener(queues = "foo")
    public String foo(String in) {
        return in.toUpperCase();
    }

    @RabbitListener(queues = "bar")
    public String bar(String in) {
        return in + in;
    }

}

interface Gate {

    List<String> doIt(String in);

}

[foofoo, FOO]
person Gary Russell    schedule 02.03.2020
comment
Я добавил простой пример, который работает так, как ожидалось; Я предлагаю вам включить ведение журнала DEBUG и сравнить журналы, чтобы увидеть, что происходит. - person Gary Russell; 04.03.2020