Как правильно использовать шаблон Scatter-Gather? - PullRequest
Новая
       8

Как правильно использовать шаблон Scatter-Gather?

0 голосов
/ 04 октября 2018

Я все еще довольно нов, когда дело доходит до Spring Integration, я пытаюсь использовать IntetrationFlowDefinition.scatterGather() безрезультатно.Общая идея состоит в том, чтобы:

  • принять String в качестве входных данных
  • использовать сплиттер для его токенизации, используя символ пробела
  • , отправив каждый токеннабор услуг, каждый из которых возвращает значение
  • агрегированные ответы от каждого сервиса
  • агрегированные ранее агрегированные ответы в List
  • , преобразующие список в один String

Я использую шлюз обмена сообщениями в качестве ввода / вывода: 

@MessagingGateway(name = "gateway")
public interface TestGateway {

  @Gateway(requestChannel = "input")
  String process(String input);

}

и вызываю его через: 

final TestGateway gateway = (TestGateway) ctx.getBean("gateway");
System.out.println(gateway.process("this is a test"));

Мои IntegrationFlow s:

@Bean
public IntegrationFlow soutFlow() {
  return f -> f.handle(m -> System.out.println(m.toString()));
}

@Bean
public IntegrationFlow flow() {
  return IntegrationFlows.from("input")
    .split(s -> s.delimiters(" "))
    .wireTap(soutFlow())
    .scatterGather(
      sc -> sc.recipientFlow(m -> true, f1 -> f1.handle((p, h) -> p + " - flow 1").wireTap(soutFlow()))
        .recipientFlow(m -> true, f2 -> f2.handle((p, h) -> p + " - flow 2").wireTap(soutFlow()))
        .applySequence(true),
      ga -> ga.outputProcessor(mg -> mg.getMessages()
        .stream()
        .map(m -> m.getPayload().toString())
        .collect(Collectors.joining(", "))),
      sg -> sg.gatherTimeout(1_000))
    .wireTap(soutFlow())
    .aggregate()
    .wireTap(soutFlow())
    .transform((List<String> source) -> source.stream()
      .map(s -> "- " + s)
      .collect(Collectors.joining("\n")))
    .get();
}

Я ожидаю увидеть вывод, отформатированный примерно так: 

- this - flow 1, this - flow 2
- is - flow 1, is - flow 2
- a - flow 1, a - flow 2
- test - flow 1, test - flow 2

К сожалению, это не работает, и сборщик всегда истекает через 1 секунду.Отладочные выходные данные с прослушиванием по проводам показывают следующие сообщения: 

GenericMessage [payload=this, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceNumber=1, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=4, correlationId=7c9653a7-55eb-72e4-ff74-62b10782cc3b, id=6c81a1ce-9a7c-83c3-3b69-7fb2f8b3112c, timestamp=1538668173435}]
GenericMessage [payload=this - flow 1, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=1, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:1, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 1, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=8d29aea7-3ec7-2c15-5b92-fbb0af002a3f, id=7ca4de30-f801-ad14-6452-249c85e9ab36, timestamp=1538668173446}]
GenericMessage [payload=this - flow 2, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=2, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:1, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 1, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=8d29aea7-3ec7-2c15-5b92-fbb0af002a3f, id=71b22112-8e7b-c2c5-057f-2a91c49af8c8, timestamp=1538668173446}]
GenericMessage [payload=is, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceNumber=2, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=4, correlationId=7c9653a7-55eb-72e4-ff74-62b10782cc3b, id=9da2f181-82de-d50e-51bb-229a3e32b998, timestamp=1538668174448}]
GenericMessage [payload=is - flow 1, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=1, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:2, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 2, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=0abfd6a8-713d-e0cb-bc59-a111a85429df, id=d5102417-f6c2-cbac-197a-b9c4f3f51a74, timestamp=1538668174448}]
GenericMessage [payload=is - flow 2, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=2, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:2, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 2, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=0abfd6a8-713d-e0cb-bc59-a111a85429df, id=d9b31390-d3d1-9b61-43fc-45c06ac86931, timestamp=1538668174449}]
GenericMessage [payload=a, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceNumber=3, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=4, correlationId=7c9653a7-55eb-72e4-ff74-62b10782cc3b, id=f109f368-6859-ac8d-0e0c-3beb563f9594, timestamp=1538668175451}]
GenericMessage [payload=a - flow 1, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=1, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:3, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 3, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=0e347381-a6e2-891a-72ff-c862565d6a1e, id=e0270063-6e19-5827-7674-21fa047ab11d, timestamp=1538668175451}]
GenericMessage [payload=a - flow 2, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=2, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:3, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 3, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=0e347381-a6e2-891a-72ff-c862565d6a1e, id=b08aab99-10ce-4dac-2f3c-b48b4514fd35, timestamp=1538668175451}]
GenericMessage [payload=test, headers={replyChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceNumber=4, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=4, correlationId=7c9653a7-55eb-72e4-ff74-62b10782cc3b, id=52feeb27-8da4-70c4-889e-f482d2922e40, timestamp=1538668176453}]
GenericMessage [payload=test - flow 1, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=1, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:4, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 4, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=53e7fd2c-d3a4-0c0a-1296-2a14ef5e90ab, id=9152e603-e436-c0a4-abde-1f7e5ae654bf, timestamp=1538668176454}]
GenericMessage [payload=test - flow 2, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@78aea4b9, sequenceNumber=2, gatherResultChannel=ac418949-6ade-4db5-bd4c-3cbbbb7e9a49:4, sequenceDetails=[[7c9653a7-55eb-72e4-ff74-62b10782cc3b, 4, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@649725e3, sequenceSize=2, correlationId=53e7fd2c-d3a4-0c0a-1296-2a14ef5e90ab, id=a7357efd-d34a-6a05-159e-5860e3bc5281, timestamp=1538668176454}]

Я, должно быть, упускаю что-то очевидное, но документации Spring очень мало, когда дело доходит до хороших примеров Java DSL.

UPDATE:

Использование wireTap() фактически влияет на поток.Я удалил его и установил несколько точек останова.Похоже, что сборщик правильно обрабатывает разбросанные и обработанные сообщения, проблема заключается в том, что они отправляются дальше по течению.Исключение составляет: 

org.springframework.messaging.MessagingException: Failed to handle Message; nested exception is org.springframework.messaging.core.DestinationResolutionException:
 failed to look up MessageChannel with name '27cf1bec-7e06-424e-80f8-9f2f67ebaf87:2' in the BeanFactory.;
 nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No bean named '27cf1bec-7e06-424e-80f8-9f2f67ebaf87:2' available,
 failedMessage=GenericMessage [payload=is - flow 1, is - flow 2, headers={replyChannel=org.springframework.integration.channel.FixedSubscriberChannel@4bafe935, sequenceNumber=2, gatherResultChannel=27cf1bec-7e06-424e-80f8-9f2f67ebaf87:2, sequenceDetails=[[2baa7135-5c02-932f-6b5d-cfd4052a965b, 2, 4]], errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@87b5b49, sequenceSize=2, correlationId=c04f280d-83b7-74e9-a0a6-0f89efdf795c, id=ffec13cf-531e-0721-1484-e8b26b705330, timestamp=1538735548924}]

Я бы предположил, что Spring будет автоматически создавать эти промежуточные каналы при использовании IntegrationFlow.Разве это не так?

1 Ответ

0 голосов
/ 05 октября 2018

Проблема в неверном / неправильно настроенном сборщике.Мое предположение состояло в том, что если вход разброса представляет собой одиночное String (токенизированное слово), то вывод также должен быть единственным String.К сожалению, созданный таким образом агрегатор неправильно восстанавливал заголовки последовательности / корреляции, что приводило к тому, что следующий по очереди агрегатор ждал завершения группы бесконечно долго.Предоставление сборщика null восстанавливает поведение по умолчанию.Недостатком является то, что вывод рассеяния теперь равен List.Это можно легко смягчить, поместив преобразователь, который выравнивает список перед следующим агрегатором.Рабочий код ниже: 

@Bean
public IntegrationFlow flow() {
  return IntegrationFlows.from("input")
    .split(s -> s.delimiters(" "))
    .scatterGather(
      sc -> sc
        .applySequence(true)
        .recipientFlow(m -> true, f1 -> f1.handle((p, h) -> p + " - flow 1"))
        .recipientFlow(m -> true, f2 -> f2.handle((p, h) -> p + " - flow 2")),
      null
    )
    .transform((List<String> l) -> l.stream().collect(Collectors.joining(", ")))
    .aggregate()
    .transform((List<String> source) -> source.stream()
      .map(s -> "- " + s)
      .collect(Collectors.joining("\n")))
    .get();
}

И вывод теперь точно такой, как ожидалось: 

- this - flow 1, this - flow 2
- is - flow 1, is - flow 2
- a - flow 1, a - flow 2
- test - flow 1, test - flow 2
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...