How to Enable Concurrent Processing of Stream Messages

[sandeep2rawat]

I am using FastStream to process messages from a Redis stream. My expectation is that the subscriber function (handle) should process the messages concurrently.
However, it seems like each message is processed sequentially, and there is no concurrency in the task execution.

import asyncio
from faststream import FastStream, Logger
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)

@broker.subscriber(
    stream=StreamSub("test-stream", group="test-group", consumer="1")
)
async def handle(msg: str, logger: Logger):
    logger.info("start")
    await asyncio.sleep(2)  # Simulating an HTTP request
    logger.info("end")

Logs

2024-10-16 18:45:43,036 INFO     - test-stream | 37f58b04-0 - Received
2024-10-16 18:45:43,039 INFO     - test-stream | 37f58b04-0 - start
2024-10-16 18:45:45,067 INFO     - test-stream | 37f58b04-0 - end
2024-10-16 18:45:45,083 INFO     - test-stream | 37f58b04-0 - Processed
2024-10-16 18:45:45,090 INFO     - test-stream | 4bd65631-b - Received
2024-10-16 18:45:45,091 INFO     - test-stream | 4bd65631-b - start
2024-10-16 18:45:47,093 INFO     - test-stream | 4bd65631-b - end

As you can see from the logs, each message is processed one after the other, waiting for the previous task to finish before the next one starts.
I expected FastStream to handle these tasks concurrently.

Key Points

• Async/Await Behavior: The handle function uses await asyncio.sleep(2) to simulate a network request. Normally, await should allow other coroutines to run concurrently, but that doesn't seem to be happening here.
• Concurrency: I thought the FastStream setup would automatically create tasks for each message in the stream, which should then run concurrently.

Questions

• Is there something I am missing in the configuration to ensure that tasks are executed concurrently?
• Do I need to explicitly create tasks for concurrency, or should FastStream handle this automatically?
• Is there a built-in way to configure concurrency for this type of subscriber function, similar to other task queues where we specify the concurrency level?

Any guidance or suggestions on how to fix this issue would be appreciated!

[kumaranvpl]

Hello @sandeep2rawat,

By default FastStream consumes single message from the Redis Stream before consuming the next message. If you want to process multiple messages at once there are few ways to do it.

1. Use Batch subscription to consume multiple messages at once.
2. While running the faststream application you can use --workers flag for scaling.

You can combine 1 and 2 to consume multiple messages at the same time.

[Lancetnik]

@sandeep2rawat Redis Stream works the same way with Kafka - so it can't be scaled horizontally using the same group and consumer.
Current offset is binded to the group, so if it consumes 1-2-3-... (offsets) messages concurrently, we can't move offset forward correctly in cases, where message 3 was processed correctly, but 2 faults. So, if you are using group and want to consume messages concurently, you should ack them first and use at most once guarantee.