Советы по проектированию интеграции

Question

Я планирую реализовать интеграционный поток, как показано ниже:

IntegrationFlows.from(httpInboundGateway)
    .transform(transformer-rest-api-1)
    .transform(transformer-rest-api-2)
    .handle(jdbc-outbound)
    .handle(http-outbound-gateway-1)
    .get(); 

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

1. , чтобы этот процесс выполнялся в параллельных потоках как можно большенасколько возможно
2. постоянное сообщение в каждой конечной точке
3. сделать саму конечную точку как rest-api (чтобы сделать поток масштабируемым)

Имеет ли смыслсделать поток реактивным?Если так, как идти об этом?Как мне войти на каждом шагу?(Поможет ли прослушивание телефонных разговоров?), И, наконец, не могли бы вы привести конкретный пример простой реализации выше в java dsl?

Answer 1

1. Чтобы сделать его параллельным, вы должны рассмотреть возможность использования ExecutorChannel между конечными точками:

   IntegrationFlows.from(httpInboundGateway)
      .channel(c -> c.executor(myTaskExecutor()))
      .transform(transformer-rest-api-1)
   

и т. Д.

Однако, поскольку вы хотели бы иметь опцию персистентности, вам следует рассмотреть возможность использования QueueChannel с персистентным MessageStore: https://docs.spring.io/spring-integration/docs/current/reference/html/system-management-chapter.html#message-store. Затем каждая конечная точка должна быть снабжена *Опции 1016 * poller , включая tasExecutor для параллельной обработки:

.channel(c -> c.queue(jdbcMessageStore(), "queue1Channel"))
       .transform(transformer-rest-api-1, 
                     e -> e.poller(p -> p.fixedDelay(100).taskExecutor(myTaskExecutor())))

Для создания конечных точек в качестве REST API вам просто нужно использовать Http.outboundGateway().Или для реактивного варианта - WebFlux.outboundGateway() в .handle() вместо transform().Или просто продолжайте в существующем .handle(http-outbound-gateway-1) в вашем потоке

Чтобы сделать поток реактивным, вам нужно использовать .channel(c -> c.flux())), но вы потеряете настойчивость в этом вопросе.

Для регистрации каждого шага между конечными точками должен быть расположен оператор .log().

Ваши требования не ясны, чтобы поделиться некоторымиобразец ...