Commands queue for sequential and parallel execution

Mansour 0
Viewed 28

I need to queue commands in such a way to avoid conflict, during commands execution. Commands are received in order. A command operate on specific entity, so I want to queue the commands based on the entities they are intended to operate on. For example, commands c1 and c3 operate on entity-id 5, so they need to go in the order they are received into q5. At the same time, commands c2, and c4 operate on entity-id 6, so they need to go on the queue specific to that entity (q-6). I am not sure sure if spring-integration is best option for this use case, but here's what I have so far:


@Bean
public IntegrationFlow commands() {

    return f -> f   .<Command>split()
                    .enrichHeaders(h -> {
                        h.headerExpression("objectId", "payload.objectId.toString() ");
                    })
                    .route("@channelResolver.resolve(headers['objectId'])") // how to create channels dynamically
                    .handle(mySingletonBeanHandler); // how to use singleton bean here
}

Problems I am facing is using Dynamic Router. Since I don't know the entities Ids in advance, I would like those queues to be created dynamically. The exception I am getting is:

The 'currentComponent' (org.springframework.integration.router.ExpressionEvaluatingRouter@3e134896) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow.

So, my questions are:

  • How to construct queues dynamically (with auto cleaning) for each entity/object-Id.
  • How to pass the messages after routing, to a singleton handler, ensuring the processing is parallel for all queues, but sequential per queue.

I am new to spring-integration, and hoping to continue using java DSL if possible to avoid confusion.

spring-integration java
1 Answers

It is really not recommended to create channels at runtime: more over you are asking probably about removal for them eventually. And you also don't know how many of them are going to be there.

I'd suggest a bit different pattern which may work for you.

  1. Use an aggregator to group command per entity as a single message for the output.
  2. The output channel of this aggregator could be an ExecutorChannel to perform per entity groups in parallel.
  3. The next endpoint could be a splitter to perform commands per entity one by one.

The resequencer also could be used since it does not let the next command to be released until the previous has release, but its purpose is going to be neglected if we use an ExecutorChannel for parallelism.

The problem with an aggregator that you have to wait for all the commands for entity to be able to release them for processing.

Or... You just can have several QueueChannel for some parallelism and respective router to send messages with the same key to the same queue channel. All of these queues can have the same consumer. I guess that's where you mention your mySingletonBeanHandler. You indeed cannot continue the flow after the route(), but you can use a @BridgeTo("myHandlerChannel") on those QueueChannel beans to dispatch their messages to your handler with the @ServiceActivator(inputChannel = "myHandlerChannel") or an IntegrationFlow starting with this channel and then handle().

In the you cannot reach more parallelism as your CPU allows, so there is no point to have "queue per entity" either way.

Artem Bilan0
Related