Camel - Проектирование масштабируемого динамического потока данных

Question

Вариант использования: я получаю сообщение от конечной точки источника. В зависимости от типа сообщения оно обрабатывается совершенно другой бизнес-логикой (например, HTTP-запрос, tcp-сообщение, вызов db). После этого результат и передается в нисходящем направлении (предполагается единый стандарт для выходного сообщения)

Маршрут будет выглядеть примерно так:

from(_source_)
...
// handle data (this is dynamic)
...
process(_logger_)
to(_receiver_)

Прямое решение будет использовать выбор ():

... // Upstream
.choice()
.when(someCondition).process(sendHTTP)
.when(anotherCondition).process(getToken).process(sendTCP)
.otherwise().process(sendToDB)
... // Downstream

Но это не очень масштабируемо.

Другим решением было бы просто поместить все в один процессор, который, в свою очередь, вызывает клиента с полиморфным поведением:

... // Upstream
.process(messageSwitch)
... // Downstream

public class MessageSwitch implements Processor {

    public void process(Exchange ex) {
         RequestClient client = this.resolveClient(ex);
         client.sendRequest(ex.getIn());
     }
}

Однако это также заставляет нас упускать из виду маршрут. В этом случае мой вопрос заключается в том, является ли обычной практикой выполнение нового маршрута, который происходит внутри процессора. Например, я могу захотеть выполнить вызов http через библиотеку http4 (http://camel.apache.org/http4.html).

А может, я неправильно подхожу к этой проблеме.

Answer 1

Один из способов, которым вы могли бы подойти к этому, - это установить процессором заголовок в полученном сообщении с маршрутом назначения для каждого типа, а затем использовать список получателей верблюда, чтобы отправить сообщение по этим маршрутам назначения.

from(_source_)
.process(_processorcodebelow_)
.recipientList(simple("direct:${header.MessageRouterByType}"));

Процессор будет выглядеть примерно так:

public void doProcees(Message message){
Message message = exchange.getIn();
... //logic to check type 
message.setHeader("MessageRouterByType", "messagetype1");
...//etc for different types

Тогда у вас могут быть разные маршруты для выполнения бизнес-логики для каждого из типов сообщений.

from("direct:messagetype1").routeId("messagetype1")
.process(_messagetype1processor_)
.to(_receiver_)

Вы можете использовать идентификатор маршрута или журнал, чтобы отслеживать, по какому маршруту прошло сообщение.