activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jamie McCrindle <jamiemccrin...@gmail.com>
Subject Re: Spring && DefaultMessageListenerContainer && JmsTemplate && maxMessagesPerTask == 1 && Network of Brokers && maxConcurrentConsumer > 1 == Crumple
Date Fri, 07 May 2010 21:20:32 GMT
Oh, this is using:

ActiveMQ 5.3.1
Spring 2.5.6

On Fri, May 7, 2010 at 10:19 PM, Jamie McCrindle
<jamiemccrindle@gmail.com> wrote:
> Greetings all,
>
> After some weeks of scratching my head, I _believe_ I have found the
> magic combination that appears to be causing messages to become stuck
> in our network of brokers. It's so convoluted that it's entirely
> likely that the error isn't what I think it is but I have managed to
> create a test case that mirrors the behaviour we're seeing live. And
> it goes something like this:
>
> We have two brokers in a network of brokers. Producers were publishing
> to a queue on one of the brokers and consumers reading off the queue
> in the other broker. After a short while, messages would suddenly pile
> up on the 'producer' broker and not get read off on the 'consumer'
> broker.
>
> Pause for lots of random testing...
>
> It appeared that the network bridge subscription from the 'consumer'
> broker was disappearing from the 'producer' broker, causing the pile
> up.
>
> More testing later...
>
> And it looks like if we have a DefaultMessageListenerContainer with
> the following configuration:
>
> maxMessagesPerTask: 1
> cacheLevel: CONSUMER
> maxConcurrentConsumer: 3 (more than 1, basically)
> concurrentConsumers: 1
> sessionAcknowledgeMode: Session.AUTO_ACKNOWLEDGE
>
> When Spring scales back the dynamic amount of consumers, the network
> subscription appears to get lost and messages pile up on the producer
> side.
>
> Workaround:
>
> Use concurrentConsumers instead of maxConcurrentConsumers so that
> there are a static number of consumers (not setting maxMessagesPerTask
> also seems to work).
>
> I'll add it this to Jira if you'd like but the test case to replicate
> the behaviour is as follows:
>
>
> package org.example.activemq;
>
> import java.util.concurrent.Callable;
> import java.util.concurrent.CountDownLatch;
> import java.util.concurrent.ExecutorService;
> import java.util.concurrent.Executors;
> import java.util.concurrent.TimeUnit;
>
> import javax.jms.JMSException;
> import javax.jms.Message;
> import javax.jms.MessageListener;
> import javax.jms.Session;
> import javax.jms.TextMessage;
>
> import junit.framework.TestCase;
>
> import org.apache.activemq.ActiveMQConnectionFactory;
> import org.apache.activemq.broker.BrokerService;
> import org.apache.activemq.command.ActiveMQQueue;
> import org.apache.activemq.network.NetworkConnector;
> import org.apache.activemq.pool.PooledConnectionFactory;
> import org.apache.activemq.store.memory.MemoryPersistenceAdapter;
> import org.springframework.jms.core.JmsTemplate;
> import org.springframework.jms.core.MessageCreator;
> import org.springframework.jms.listener.DefaultMessageListenerContainer;
>
> public class NetworkTest extends TestCase {
>
>    public void testNetworkOfBrokers() throws Exception {
>        BrokerService brokerService1 = null;
>        BrokerService brokerService2 = null;
>
>        final int total = 100;
>        final CountDownLatch latch = new CountDownLatch(total);
>
>        try {
>
>        {
>            brokerService1 = new BrokerService();
>            brokerService1.setBrokerName("one");
>            brokerService1.setUseJmx(false);
>            brokerService1.setPersistenceAdapter(new
> MemoryPersistenceAdapter());
>            brokerService1.addConnector("tcp://0.0.0.0:61616");
>            NetworkConnector network1 =
> brokerService1.addNetworkConnector("static:(tcp://localhost:51515)");
>            network1.setName("network1");
>            network1.setDynamicOnly(true);
>            network1.setNetworkTTL(3);
>            network1.setPrefetchSize(1);
>            brokerService1.start();
>        }
>
>        {
>            brokerService2 = new BrokerService();
>            brokerService2.setBrokerName("two");
>            brokerService2.setUseJmx(false);
>            brokerService2.setPersistenceAdapter(new
> MemoryPersistenceAdapter());
>            brokerService2.addConnector("tcp://0.0.0.0:51515");
>            NetworkConnector network2 =
> brokerService2.addNetworkConnector("static:(tcp://localhost:61616)");
>            network2.setName("network1");
>            network2.setDynamicOnly(true);
>            network2.setNetworkTTL(3);
>            network2.setPrefetchSize(1);
>            brokerService2.start();
>        }
>
>        ExecutorService pool = Executors.newSingleThreadExecutor();
>
>        ActiveMQConnectionFactory connectionFactory1 = new
> ActiveMQConnectionFactory("failover:(tcp://localhost:61616,tcp://localhost:51515)?randomize=false");
>
>
>        final DefaultMessageListenerContainer container = new
> DefaultMessageListenerContainer();
>        container.setConnectionFactory(connectionFactory1);
>        container.setMaxConcurrentConsumers(10);
>        container.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
>        container.setCacheLevel(DefaultMessageListenerContainer.CACHE_CONSUMER);
>        container.setDestination(new ActiveMQQueue("testingqueue"));
>        container.setMessageListener(new MessageListener() {
>            public void onMessage(Message message) {
>                latch.countDown();
>            }
>        });
>        container.setMaxMessagesPerTask(1);
>        container.afterPropertiesSet();
>        container.start();
>
>        pool.submit(new Callable<Object>() {
>            public Object call() throws Exception {
>                try {
>                    final int batch = 10;
>                    ActiveMQConnectionFactory connectionFactory2 = new
> ActiveMQConnectionFactory("failover:(tcp://localhost:51515,tcp://localhost:61616)?randomize=false");
>                    PooledConnectionFactory pooledConnectionFactory =
> new PooledConnectionFactory(connectionFactory2);
>                    JmsTemplate template = new
> JmsTemplate(pooledConnectionFactory);
>                    ActiveMQQueue queue = new ActiveMQQueue("testingqueue");
>                    for(int b = 0; b < batch; b++) {
>                        for(int i = 0; i < (total / batch); i++) {
>                            template.send(queue, new MessageCreator() {
>                                public Message createMessage(Session
> session) throws JMSException {
>                                    TextMessage message =
> session.createTextMessage();
>                                    message.setText("Hello World!");
>                                    return message;
>                                }
>                            });
>                        }
>                        // give spring time to scale back again
>                        while(container.getActiveConsumerCount() > 1) {
>                            System.out.println("active consumer count:
> " + container.getActiveConsumerCount());
>                            System.out.println("concurrent consumer
> count: " + container.getConcurrentConsumers());
>                            Thread.sleep(1000);
>                        }
>                    }
>                    pooledConnectionFactory.stop();
>                } catch(Throwable t) {
>                    t.printStackTrace();
>                }
>                return null;
>            }
>        });
>
>        pool.shutdown();
>        pool.awaitTermination(10, TimeUnit.SECONDS);
>
>        int count = 0;
>
>        // give it 20 seconds
>        while(!latch.await(1, TimeUnit.SECONDS) && count++ < 20) {
>            System.out.println("count " + latch.getCount());
>        }
>
>
>        container.destroy();
>
>        } finally {
>            try { if(brokerService1 != null) { brokerService1.stop();
> }} catch(Throwable t) { t.printStackTrace(); }
>            try { if(brokerService2 != null) { brokerService2.stop();
> }} catch(Throwable t) { t.printStackTrace(); }
>        }
>
>        if(latch.getCount() > 0) {
>            fail("latch should have gone down to 0 but was " +
> latch.getCount());
>        }
>
>    }
>
> }
>

Mime
View raw message