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.

[sandeep2rawat]

Hi @kumaranvpl ,thanks it makes sense, is there any documention around batch configuration, few configuration I'm looking for

• max batch size
• and how message acknowledgement works, is it happens after all messages gets processed and the handler returns successfully ?
  Pls share if there's any other doc than batching

[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.

[vvanglro]

Should each process use a different consumer name?

According to the Redis documentation, each consumer maintains its own PEL.

If I use UUID to represent consumer names, then each time the service restarts, there will be consumers that cannot be found.

Redis streams make it a heavy burden to use.😮‍💨

[sandeep2rawat]

Consumer name shouldn't be uuid, it's the name of consumer what it's doing
Eg. Consumer 1 calculating billing around events , so you can name it( billing), another generating some report (reporting)and another sending emails etc.

[vvanglro]

So is it normal for multiple processes to share the same consumer name? Or should it be better for each process to have a unique name?

https://stackoverflow.com/questions/69507376/why-redis-stream-consumers-should-specify-an-individual-name-in-the-same-consume

https://stackoverflow.com/questions/65702366/redis-streams-groups-consumers-naming-and-cleanup

[sandeep2rawat]

My bad, yes consumer name should be unique , it's the group name which is unique , u're right you may need write cleanup.
As per your requirement (running periodically) u can name consumers based on some pattern eg. {Hostname}-{count} to prevent growing consumer list.

[vvanglro]

Can add a feature to clean up consumers? 🤔

consumers_info = redis_conn.xinfo_consumers(REDIS_DOWNLOAD_STREAM, REDIS_CONSUMER_GROUP)
idle_threshold = 60 * 60 * 1000  # Over an hour free, clean up

for consumer in consumers_info:
    name = consumer['name']
    pending = consumer['pending']
    idle = consumer['idle']  # ms

    if pending == 0 and idle > idle_threshold and name != CONSUMER_NAME:
        # Delete idle old consumer
        try:
            redis_conn.xgroup_delconsumer(REDIS_DOWNLOAD_STREAM, REDIS_CONSUMER_GROUP, name)
            print(f"🗑️ Deleted idle old consumer: {name}")
        except Exception as e:
            import traceback
            traceback.print_exc()