amq will respect the message priority order if vmcursor is used. on the other hand it will not completely respect that order as default messagecursor is used and consumers are slower than producer,because message are read from the store and ordered and at last sent to consumers in batches. I donot know whether this is correct and expect Gary show the way.
At 2011-11-15 20:24:01,"Gary Tully" <gary.tu...@gmail.com> wrote: >The fix will be in 5.6, have a peek at >https://issues.apache.org/jira/browse/AMQ-3596, you can validate it in >tonight's 5.6-SNAPSHOT or try one of the workarounds. > >On 14 November 2011 17:33, Mihai Osian <mihai.os...@gmail.com> wrote: >> Hello, >> >> I am trying to enable priority consumption and I can't seem to get it >> working right. I am using ActiveMQ version 5.5.1 and I have followed the >> instructions from >> http://activemq.apache.org/how-can-i-support-priority-queues.html. >> For testing I modified the example code that comes with ActiveMQ itself so >> that the ProducerTool sends 180 messages with cycling priority >> (0,1,..8,0,1,2...). After the producer finishes the job I start the consumer >> and look at the order in which the messages are received. The first few >> messages are sometimes received in an apparently random order, after which >> they start coming in descending order of their priority, as I expect. >> I have also discovered that if I set queuePrefetch="1" in the >> "policyEntry" element from the XML configuration file I get an unsorted >> list. Similarly, setting it to 3 (for example) results in the output list >> being thoroughly randomized (messages are presumably sorted in groups of 3, >> but not quite). >> >> My questions are: >> 1. Can somebody explain the results I see ? Am I missing something ? >> 2. Is there a way to get the all messages fully sorted in descending order, >> assuming one single consumer that is started after all messages are in the >> queue ? >> 3. The ActiveMQ documentation page (written for version 5.4) that I >> mentioned above says that KahaDB supports 3 levels of priority (<5, 5, and >>>5). However: >> 3a. The first few messages that I get are sometimes low-priority (<5). >> 3b. With a larger queuePrefetch most messages are fully sorted in >> descending order, so it may be that the page is out of date and in the >> latest version it is supposed to support the full priority range. Still, I >> don't understand the output. >> >> Other technical information: >> - ActiveMQ 5.5.1 (also tested 5.5.0, same behavior) >> - activemq started from the command line with: >> ./bin/activemq console xbean:conf/activemq-priority.xml >> The activemq-priority.xml is a copy of the conf/activemq.xml, plus the >> following entry: >> <policyEntry queue="TEST.FOO" prioritizedMessages="true" >> queuePrefetch="200" producerFlowControl="true" >> memoryLimit="1mb"></policyEntry> >> - A modified version of the ActiveMQ example (also containing the >> activemq-priority.xml file) can be found here: >> http://www.mediafire.com/?z7u5fhhlfbt3ano , if anybody wants to check the >> code. >> >> ========================================= >> An example output copy-pasted from my console: >> >> mike@dell:~/tmp/apache-activemq-5.5.0/example$ ant producer >> Buildfile: /home/mike/tmp/apache-activemq-5.5.0/example/build.xml >> >> init: >> >> compile: >> [javac] /home/mike/tmp/apache-activemq-5.5.0/example/build.xml:152: >> warning: 'includeantruntime' was not set, defaulting to >> build.sysclasspath=last; set to false for repeatable builds >> >> producer: >> [echo] Running producer against server at $url = tcp://localhost:61616 >> for subject $subject = TEST.FOO >> [java] Connecting to URL: tcp://localhost:61616 >> [java] Publishing a Message with size 1000 to queue: TEST.FOO >> [java] Using non-persistent messages >> [java] Sleeping between publish 0 ms >> [java] Running 1 parallel threads >> [java] [Thread-1] Sending message: 'Message with priority: 0 index: 0 >> sent at: Mon Nov...' with priority 0 >> [java] [Thread-1] Sending message: 'Message with priority: 1 index: 1 >> sent at: Mon Nov...' with priority 1 >> [java] [Thread-1] Sending message: 'Message with priority: 2 index: 2 >> sent at: Mon Nov...' with priority 2 >> [java] [Thread-1] Sending message: 'Message with priority: 3 index: 3 >> sent at: Mon Nov...' with priority 3 >> [java] [Thread-1] Sending message: 'Message with priority: 4 index: 4 >> sent at: Mon Nov...' with priority 4 >> [java] [Thread-1] Sending message: 'Message with priority: 5 index: 5 >> sent at: Mon Nov...' with priority 5 >> [java] [Thread-1] Sending message: 'Message with priority: 6 index: 6 >> sent at: Mon Nov...' with priority 6 >> [java] [Thread-1] Sending message: 'Message with priority: 7 index: 7 >> sent at: Mon Nov...' with priority 7 >> [java] [Thread-1] Sending message: 'Message with priority: 8 index: 8 >> sent at: Mon Nov...' with priority 8 >> [java] [Thread-1] Sending message: 'Message with priority: 0 index: 9 >> sent at: Mon Nov...' with priority 0 >> [java] [Thread-1] Sending message: 'Message with priority: 1 index: 10 >> sent at: Mon No...' with priority 1 >> [java] [Thread-1] Sending message: 'Message with priority: 2 index: 11 >> sent at: Mon No...' with priority 2 >> [java] [Thread-1] Sending message: 'Message with priority: 3 index: 12 >> sent at: Mon No...' with priority 3 >> [java] [Thread-1] Sending message: 'Message with priority: 4 index: 13 >> sent at: Mon No...' with priority 4 >> [java] [Thread-1] Sending message: 'Message with priority: 5 index: 14 >> sent at: Mon No...' with priority 5 >> [java] [Thread-1] Sending message: 'Message with priority: 6 index: 15 >> sent at: Mon No...' with priority 6 >> ... >> >> mike@dell:~/tmp/apache-activemq-5.5.0/example$ ant consumer >> Buildfile: /home/mike/tmp/apache-activemq-5.5.0/example/build.xml >> >> init: >> >> compile: >> [javac] /home/mike/tmp/apache-activemq-5.5.0/example/build.xml:152: >> warning: 'includeantruntime' was not set, defaulting to >> build.sysclasspath=last; set to false for repeatable builds >> >> consumer: >> [echo] Running consumer against server at $url = tcp://localhost:61616 >> for subject $subject = TEST.FOO >> [java] Connecting to URL: tcp://localhost:61616 >> [java] Consuming queue: TEST.FOO >> [java] Using a non-durable subscription >> [java] JMS Selector >> [java] Running 1 parallel threads >> [java] [Thread-1] We are about to wait until we consume: 180 message(s) >> then we will shutdown >> [java] [Thread-1] Received: 'Message with priority: 7 index: 7 sent at: >> Mon Nov...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 8 index: 8 sent at: >> Mon Nov...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 8 index: 17 sent >> at: Mon No...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 7 index: 16 sent >> at: Mon No...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 6 index: 6 sent at: >> Mon Nov...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 6 index: 15 sent >> at: Mon No...' (length 1000) >> [java] [Thread-1] Received: 'Message with priority: 5 index: 5 sent at: >> Mon Nov...' (length 1000) >> .... >> ====================== >> >> Thanks in advance, >> Mihai >> >> http://activemq.2283324.n4.nabble.com/file/n4040260/example.tgz example.tgz >> >> >> -- >> View this message in context: >> http://activemq.2283324.n4.nabble.com/Message-priority-tp4040260p4040260.html >> Sent from the ActiveMQ - User mailing list archive at Nabble.com. >> > > > >-- >http://fusesource.com >http://blog.garytully.com
