Proper strategy for handling CPU and I/O-bound events

[pySilver]

Hi!

I'm just starting my journey with event sourcing and as a practical exercise, I'm migrating data feed processing from Celery to Nats Jetstream in my Django monolith app. I love how FastStream is structured - it is so cool to experiment with this library! However, I have a few design questions.

In my previous Celery-based approach I've had separate queues for CPU and I/O-bound tasks. The CPU queue was processed by the prefork pool (one process per CPU core) while the I/O-bound task was consumed using a gevent pool of size 500-1000.

Q1: Will it be correct if I:

1. Define different subjects for CPU and IO tasks, so proper consumer app will eventually consume them
2. Setup two separate consumer apps that I will start with different workers number: faststream run cpu:app --workers 12 and faststream run io:app --workers 1

Will I benefit from having multiple workers of my IO app consumer? Given that all handlers are using asyncio to process events.

Q2: How do I throttle a maximum number of processed events of some type? I guess this is not something people usually do when processing events but here is my real-world case: a data processing handler emits thousands of events that require external HTTP requests. How can I make sure not to DDoS an endpoint?

Q3: How to bulk read from a broker? I see there is an option of batch_size but it will simply call a handler function multiple times, right? What if I want to collect a few hundred messages (to bulk-insert them into my database) and manually ack them afterward?

Q4: How do I keep handlers in separate locations of the app? I guess I need to setup my broker and the app in consumer-app.py and keep my handlers in different locations like this:

# io-consumer.py
from faststream import FastStream, Logger
from faststream.nats import JStream, NatsBroker

broker = NatsBroker("nats://localhost:4222")
stream = JStream(name="product_feeds")
app = FastStream(broker)

# project/app1/handlers.py
from serve import broker

@broker.subscriber(
    "product_feeds",
    stream=stream,
    durable="my_app2",
)
async def handler(msg: str, logger: Logger):
    logger.info(msg)

Is this the correct approach?

--
I'll be thankful for any input on this!

[pySilver]

I've read the docs a little bit further. So I guess:

• Q2 can be solved with @broker.subscriber(..., max_workers=10)
• Q3 can be solved with PullSub(batch_size=10, batch=True)

Am I right?

[Lancetnik]

@pySilver sorry for answering so late, I am just missed the question. Next time be free to tag me directly or ask for help in discord.

Q1: it is ok to create two different FastStream instances to process CPU and I/O-bound tasks separately. It is required to don't block your broker at CPU-bounded ones and still process I/O tasks. Workers always gives you scaling benefits, but there is no reason to creat a lot of workers if you have no such strong incoming stream.

Q2: max_workers is an option allows you to consume multiple messages from the same subject in the same time without workers. It doesn't help you to throttle messages and we have no such mechanism in box, sorry

Q3: sure, PullSubscribtion - is a preffered way to consume messages from NATS JS and it also provides you with ability to consume messages in batches (it can works like some kind of throttling, I think)

Q4: please, use Routers https://faststream.airt.ai/latest/getting-started/routers/

[pySilver]

@Lancetnik thanks for such a detailed response!

Q1: Great!
Q2: Well in a way it can at least provide some boundaries! As I see max_worker is used in asyncio semaphore. I'm wondering how it plays with synchronous handlers?
Q3: Great!
Q4: Yes, I was wondering if that's the thing I should be using! Thanks!

[Lancetnik]

Background async task allows you to not block on message processing and consume the next one immediately. Semaphore in this case is just a limiter on procesing messages count.

[pySilver]

Thanks!

[Lancetnik]

Always welcome)