activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Vadim Chekan" <kot.bege...@gmail.com>
Subject Re: CLIENT_ACKNOWLEDGE mode for a session misunderstanding?
Date Wed, 29 Oct 2008 03:16:50 GMT
Sorry, I think I got it wrong. If message remain intact when consumer
disconnect then CLIENT_ACKNOWLEDGE is honored.


On Tue, Oct 28, 2008 at 7:13 PM, Vadim Chekan <kot.begemot@gmail.com> wrote:
> James,
>
> Are you saying that CLIENT_ACKNOWLEDGE option can not be trusted?
>
> Vadim.
>
> On Tue, Oct 28, 2008 at 10:44 AM, James Strachan
> <james.strachan@gmail.com> wrote:
>> 2008/10/28 Eugeny N Dzhurinsky <bofh@redwerk.com>:
>>> Hello!
>>>
>>> I am facing some strange issue with acknowledge mode in ActiveMQ. After
>>> reading the specs, I realized the broker will never deliver a message to the
>>> consumer unless the consumer acknowledges the previous message. So I thought
>>> it is enough to not send the acknowledgement message to the server in the
>>> onMessage method of the consumer and sent it later from another thread, which
>>> does it's job. This means I spawn a thread when onMessage is called and forget
>>> about it in this method.
>>
>>
>> You should only be using a JMS session from one thread at once - so
>> you should call message.acknowlege() from the consumer thread.
>>
>>
>>
>>>
>>> However I've found this is not true. Below is my simple test case - I am
>>> starting the single consumer and several producers.
>>>
>>> ===================================================================================================================
>>> import javax.jms.Connection;
>>>
>>> public class TestAcknowledgeMode {
>>>
>>>    private static final String DATA = "data";
>>>
>>>    private static final String SERVER_QUEUE = "server_queue";
>>>
>>>    private static final String CONNECTION_URL = "vm://localhost?broker.persistent=false&jms.prefetchPolicy.all=1";
>>>
>>>    private static final int CLIENT_THREADS = 5;
>>>
>>>    static ActiveMQConnectionFactory factory;
>>>
>>>    Semaphore semaphore = new Semaphore(CLIENT_THREADS, true);
>>>
>>>    AtomicInteger successCount = new AtomicInteger(0);
>>>
>>>    @BeforeClass
>>>    public static void initFactory() throws Exception {
>>>        factory = new ActiveMQConnectionFactory(CONNECTION_URL);
>>>        BasicConfigurator.configure(new ConsoleAppender(new PatternLayout(
>>>                "%5p %d{hh:mm:ss} [%t] (%C{1}:%L) - %m%n")));
>>>        Logger.getRootLogger().setLevel(Level.DEBUG);
>>>    }
>>>
>>>    @Test(timeout = 10000)
>>>    public void testAckDelivery() throws Exception {
>>>        final Connection clientConnection = factory.createConnection();
>>>        clientConnection.start();
>>>        final Connection serverConnection = factory.createConnection();
>>>        serverConnection.start();
>>>
>>>        final Session serverSession = serverConnection.createSession(false,
>>>                Session.CLIENT_ACKNOWLEDGE);
>>>        final Queue serverQueue = serverSession.createQueue(SERVER_QUEUE);
>>>        final MessageConsumer consumer = serverSession
>>>                .createConsumer(serverQueue);
>>>        consumer.setMessageListener(new MessageListener() {
>>>
>>>            public void onMessage(final Message message) {
>>>                System.out.println("Got message");
>>>                try {
>>>                    System.out.println(message.getStringProperty(DATA));
>>>                    // message.acknowledge();
>>>                    successCount.incrementAndGet();
>>>                } catch (final Exception e) {
>>>                    e.printStackTrace();
>>>                }
>>>            }
>>>
>>>        });
>>>        for (int i = 0; i < CLIENT_THREADS; i++) {
>>>            final Thread t = new Thread(new Runnable() {
>>>
>>>                public void run() {
>>>                    try {
>>>                        final Session clientSession = clientConnection
>>>                                .createSession(false,
>>>                                        Session.CLIENT_ACKNOWLEDGE);
>>>                        final Queue clientQueue = clientSession
>>>                                .createQueue(SERVER_QUEUE);
>>>                        final MessageProducer producer = clientSession
>>>                                .createProducer(clientQueue);
>>>                        final Message msg = clientSession.createMessage();
>>>                        msg.setStringProperty(DATA, Thread.currentThread()
>>>                                .getName());
>>>                        System.err.println("Sending data "
>>>                                + Thread.currentThread().getName());
>>>                        producer.send(msg);
>>>                    } catch (final Exception e) {
>>>                        e.printStackTrace();
>>>                    }
>>>                }
>>>            });
>>>            t.start();
>>>        }
>>>        for (;;) {
>>>            if (successCount.intValue() == CLIENT_THREADS)
>>>                break;
>>>            Thread.sleep(1000);
>>>        }
>>>        Assert.assertEquals(CLIENT_THREADS, successCount.intValue());
>>>    }
>>> }
>>> ===================================================================================================================
>>>
>>> I expected the only one message will be processed by the consumer, so the rest
>>> messages will never be delivered to it, but this test shows the messages are
>>> delivered to the consumer and onMessage method is called exactly CLIENT_THREADS
times,
>>> which seems to be wrong.
>>>
>>> Does it mean the onMessage method will be executed with no matter of
>>> acknowledgement is sent back, as soon as previous execution finishes?
>>>
>>> And what is the purpose of message.acknowledge() method?
>>
>> If you don't want the message broker to send another message to the
>> consumer until it is acknowledged, set prefetch to 1
>> http://activemq.apache.org/what-is-the-prefetch-limit-for.html
>>
>> --
>> James
>> -------
>> http://macstrac.blogspot.com/
>>
>> Open Source Integration
>> http://fusesource.com/
>>
>
>
>
> --
> From RFC 2631: In ASN.1, EXPLICIT tagging is implicit unless IMPLICIT
> is explicitly specified
>



-- 
>From RFC 2631: In ASN.1, EXPLICIT tagging is implicit unless IMPLICIT
is explicitly specified

Mime
View raw message