how to stop and start consuming after a few messages

[james jasper]

Hi,

Im currently using pika version 1.3.1, and erlang 22.2.7.  Im currently using blocking connection. My consumer can process only few messages i just wanted to know how  can i stop consuming after a few messages and start again and repeat the process.  

[Luke Bakken]

Hello,

A great way to start this discussion would be to provide a git repository I could clone to see what you're doing now.

One option is to set the channel prefetch to the number of messages you'd like to process at a time. Only acknowledge the batch of messages when all are done, then you would get the next batch. No need to stop / start consuming in that case.

Thanks,

Luke

[james jasper]

I apologize for not adding the repo, here it is https://github.com/Opperessor/rabbitmq_pika/blob/main/consumer.py  . So im trying to consume messages based on size. once the size limit(4000 bytes ) it reached i have stop consuming and process it and start consuming again.

[Luke Bakken]

Hi James,

Please see the following PR:

https://github.com/Opperessor/rabbitmq_pika/pull/2

You were not using getsizeof() correctly. The function is not recursive by default - https://docs.python.org/3/library/sys.html#sys.getsizeof

I made corrections to your use of Pika as well. If you need further assistance let me know.

Thanks,

Luke

[james jasper]

hi luke,i

thanks for the reply and correction. From the modified code i'm not able to process messages for 20 minutes its not able to publish, referring to this(for long running process) document https://github.com/pika/pika/blob/main/examples/basic_consumer_threaded.py , i made few changes to the code in this repo https://github.com/Opperessor/rabbitmq_pika/blob/test_branch/consumer.py

getting error as  raise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer'). Could you pls help me out.

Thanks,

james

[Luke Bakken]

What is logged by RabbitMQ at that point in time?

[james jasper]

hi luke, checked the log, got this missed heartbeats from client, timeout: 5s on server side log.

[Luke Bakken]

Thanks. I see the issue with your code and will open a new PR. Basically you're still doing work on the same thread that Pika is using and thus your work blocks Pika's I/O loop.

[Luke Bakken]

https://github.com/Opperessor/rabbitmq_pika/pull/3

I will comment on the PR itself as to what I changed.

[james jasper]

Hi, Luke thank you so much for the reply, there seems to be no connection timeout error. However when the queue is declared as a quorum queue multiple consumers are produced and few messages are in unackd state(if the messages in the consume queue are in odd number like 9, one message goes to unack state and 2 consumers can be seen), I tried to set it as x-single-active-consumer or using cancel() method didnt seem to work. This is my code https://github.com/Opperessor/rabbitmq_pika/blob/feature_branch/consumer.py made few changes referring from the pr you raised(thank you so much)  changed to consume generator to make use of the timeout option and i tried to avoid recursive way, the consumer is being increased still could you pls help me out.

Thanks,

james

[Luke Bakken]

Hello,

Please re-read what I wrote in my previous pull request: https://github.com/Opperessor/rabbitmq_pika/pull/3#issue-1560439218

I don't think you should use a "for" loop consumer if you plan on cancelling it.

At this point your consuming code is complicated enough that you should consider using SelectConnection rather than BlockingConnection.

I'll see if I have time to assist you further.

Thanks,

Luke

[james jasper]

Sure Luke,

thanks 

[Luke Bakken]

https://github.com/Opperessor/rabbitmq_pika/pull/5

[james jasper]

Thank you so much! Luke