How to enable parallel message processing in nodejs app?

[utkarshdeep]

Summary

So I have a NodeJs application that reads off sqs-queue. But since I am using the handleMessage functionality it processes messages one at a time. But I have use cases where I need message processing in parallel as one at a time message processing will cause the queue to build up and reflects poorly on us.

Can anyone help me with how I can I implement this?

Example

No response

[dginzbourg]

@utkarshdeep I'm coming exactly with the same dilemma. The fact that the library waits synchronously only up to 10 messages limits it's usefulness in building event processing when processing takes some time (e.g. webhooks). You can take a look at https://github.com/caiokf/sqs-queue-parallel.git and either use that project or modify/extend this one by not waiting for handleMessages, but queuing it (see https://github.com/caiokf/sqs-queue-parallel/blob/7b8ba01f6c75de4f2c34a6ae4b00a59ef2790fc5/dist/sqs-queue-parallel.js#L126-L127 to queue the messages up to X). I'm not familiar with coffee this project used, but can rework it or merge with sqs-consumer, haven't decided yet.
I see this PR https://github.com/bbc/sqs-consumer/pull/271/files#diff-720b6bfe5e0c4c43dcef59d90017b3517a0cfaa675b8f017ecf9e29fabfa48ef that wasn't merged, but had implemented the concurrency.

[dginzbourg]

I forked the project to https://github.com/dmitrydigi/sqs-consumer and have implement asynchronous message processing while polling in parallel. Polling keeps in memory up to maxInflightMessages and the package supports termination timeout for messages draining. I haven't used fastq as in the PR mentioned earlier. I wish that that it will be supported by the main contributors and then can remove the fork.

[nicholasgriffintn]

This looks quite good actually, we'd be more open to a pr with this sort of syntax and without extra dependencies.

Along the lines of what we were thinking of getting to, just not something we can prioritise on our end.

[dginzbourg]

Do you suggest me to open PR from the fork? I might rollback pino to debug, did it due to vulnerabilities in debug package, but looks like it's better to keep debug instead of pino. I plan to add RateLimitError throw from handleMessage(Batch) to allow visibility timeout override. Working with async tasks (message handlers) requires support of back pressure mechanism.

[nicholasgriffintn]

Hey sorry for the delay, yeah it's probably a decent start that could be expanded for the other functionality.

[jdebarochez]

@dginzbourg @utkarshdeep I come with the same dilemma than both of you, and wonder if we can have n consumers instances on a same Nodejs app. Would there be anything preventing it inside the lib?

We're using Kubernetes, and are already having concurrent consumers on different pods. We don't see (yet) what would go wrong having a concurrent consumers inside the same pod.

[dginzbourg]

I'm not aware of such limitation. Consumer is a class, so every initiation is a different context. I wonder why would you need n consumer instances instead of working with maxInflightMessages (as on the branch I've created)? In that implementation the polling and message execution are decoupled up to maxInflightMessages limit reached. I plan to open PR (just moved recently to Java project, so finding hard to complete that PR)