activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Neil Pritchard <Neil.Pritch...@securetrading.com>
Subject RE: Individual acknowledge in pyactivemq
Date Tue, 14 Sep 2010 16:00:01 GMT
On Tue, 2010-09-14 at 14:17 +0100, Neil Pritchard wrote:
> Hi All,
> 
> I'm using pyactivemq as both a producer and consumer of messages which are being brokered
by ActiveMQ 5.3.2.  In the past I used python stompy to produce messages and java ActiveMQ
to consume them.  I need to set a prefetch policy of 1 messgae at a time and more importantly
use individual acknowledge (or client acknowledge) but can't find any example of how to acknowledge
the message in pyactivemq, also the activeMQ log (set to debug) indicates that the messages
are being consumed regardless.  Can anyone tell me how to set the acknowledgement mode properly
and how to acknowledge a message.  When I try to acknowledge, Python can't find the Acknowledge()
method.
> 
> Many thanks,
> 
> Neil
> 
> Here's my consumer:....
> 
> #!/usr/bin/env python
> 
> import cpickle as pickle
> import Queue
> import pyactivemq
> from pyactivemq import ActiveMQConnectionFactory
> from pyactivemq import AcknowledgeMode
> 
> class MessageListener(pyactivemq.MessageListener):
>     def __init__(self, name, queue):
>         pyactivemq.MessageListener.__init__(self)
>         self.name = name
>         self.queue = queue
> 
>     def onMessage(self, message):
>         self.queue.put('%s got: %s' % (self.name, message.text))
> 
> f = ActiveMQConnectionFactory('failover:(tcp://localhost:61616)?wireFormat=openwire')
> conn = f.createConnection()
> consumer_session = conn.createSession(AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE)
> myqueue = consumer_session.createQueue('NOTIFICATIONS.QUEUE')
> queue = Queue.Queue(0)
> 
> session = conn.createSession(AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE)
> consumer = session.createConsumer(myqueue)
> listener = MessageListener('consumer', queue)
> consumer.messageListener = listener
> 
> conn.start()
> while queue:
>     message = queue.get(block=True)
>     message = message[14:]
>     notificationDict = pickle.loads(message)
>     print notificationDict
>     acknowledge()
> 
> conn.close()
> 

It looks as if the acknowledge method is exposed on each Message object,
so you should be able to ack them as they are received.  Whether or not
it works is another question.

Regards


-- 
Tim Bish



That's what I thaught, but I've had to move the message.acknowledge() to the 'onMessgae' method
definition on the MessageListener class to get it to acknowledge.  The problem that I now
have is that the consumer takes all the messages off the queue, processes them and then acknowledges
them.  I need the messages to stay on the queue so that they're available to other consumers
while this one is blocked (for load balancing).  I the Java consumer that I have running,
I point the consumer at a brokerURL of 
"failover:(tcp://localhost:61616)?jms.prefetchPolicy.all=1"
to set a prefetch policy of 1.  I've tried adding it to the connection URL in this script
" ActiveMQConnectionFactory('failover:(tcp://localhost:61616)?wireFormat=openire&jms.prefetchPolicy.all=1')"
but it's obviously still collecting as many messages as it can.  How should I sent the prefetch
policy ?

Mime
View raw message