Help with basic_consumer_threaded.py example

[albertkohl-monotek]

Hello everyone, i'm fairly new to python and i'm trying to setup a windows machine as a worker for a rabbitmq queue. i got everything working with with heartbeats and everything, and now i'm trying to get this same script to work with multiple worker threads.

in a nut-shell, i would like to run 4 tasks (where 4 is configurable)

i'm wondering if anyone can help me adapt my script to work. this has been adapted from the sample script: /examples/basic_consumer_threaded.py written by @mprinc

main.py attached.

any help would be appreciated, and be gentle, i'm new to python! :)

main.py.txt

[lukebakken]

Hi @albertkohl-monotek, I maintain Pika and I can help you out. Would you please create a git repository, containing all of your code, that I can clone to assist you?

[albertkohl-monotek]

Thanks @lukebakken here's the repo: albertkohl-monotek/pika_worker
basically i want to set a constant at the top that tells it how many worker threads to run. this way i can run more threads when needed for specific workers.

i know there is the prefetch_count option you can pass the channel, but my impression is that's just how many max jobs it would ack/pull initially, i guess i'm assuming that after that first initial pull, it would eventally dwindle down to only 1 until there were no more jobs. then whenever it goes to pull more (when available) it would pull up to 4 again (for example)

maybe that's really all i have to change and it's the key.

[lukebakken]

Yep, you just need to set the prefetch and that will set a maximum number of un-acked messages sent to your client. Once a thread acks a message, RabbitMQ will send a new one and a new thread will be started.

If you're interested, I ran some formatting on your code in my fork - https://github.com/lukebakken/pika-1429/tree/lukebakken/pika-1429. I also checked in the Pipfile.lock file.

Anyway, the thread handling code is pretty basic in the example as the list of threads isn't cleaned up when a thread completes. That could be replaced with a dict, for instance.

Thanks for using Pika and RabbitMQ.

[albertkohl-monotek]

Thanks @lukebakken i'll run through it when i get a sec later today. appreicate the advice/help!

[albertkohl-monotek]

answer was to use the prefork option on the channel.

[albertkohl-monotek]

@lukebakken quick question, if i had 3 nodes running processes, and each node was setup to take 5 tasks, but i only had 10 tasks available, would the tasks round-robin so each the dispursement is 3/3/4, or wil the first node that checks in get 5, second gets 5, and 3rd gets 0?

if the later, is there a way to configure it so the tasks are distributed evenly?

[lukebakken]

That's a good question. My guess is that you will get perfect round-robin distribution if you use a qos of 1, and the scenario you describe (5, 5, 0) will happen when you use a qos of 5.

[albertkohl-monotek]

any way to get it to do the 3/3/4 method while still allowing each machine to run multiple processes simultaneously? i would like to have my cake, and eat it too :)

[lukebakken]

Unfortunately not without additional coding -

• Consumers have two main threads - one consumes from the task queue with a QoS of 1, the other consumes from a dedicated work queue (exclusive queue would be a good option) without a QoS
• The task queue consumer gets task messages, publishes them to the exclusive work queue, and checks the main worker thread to see how many tasks it's currently working on. If 5 or greater, stop consuming, else ack the work message to keep getting work.
• The task worker starts threads to do the work any time it has messages in its work queue. When a task is complete, ack it to remove it from the exclusive work queue, and decrement the "in progress" count.