activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravimbhatt <r...@qubitdigital.com>
Subject Stops Message Delivery to consumer: QueueConnection Stop Waits forever
Date Wed, 21 Sep 2011 17:00:00 GMT
[Posting again as my previous post did not get requiredd help]
Hi All, 

I am facing a very strange problem with ActiveMQ. 

I have a java class that attaches itself to a queue. This class holds a
threadExecutor and once it gets a message, it uses the threadexecutor to run
a task with just arrived message. 

It keeps track of how many threads are currently performing tasks, once it
reaches a MAX, say 4, it stops the connection so that it does not get more
messages from the queue. As and when any one of the tasks complete it starts
connection again. 

This setup works for few messages and then suddenly my class waits forever
in connection's stop method. it never returns!! 

Here is my onMessage method: 

public void onMessage(Message objectMessage) { 
    if (objectMessage instanceof ObjectMessage) { 
      try { 
        Serializable serializableObject = ((ObjectMessage) objectMessage) 
            .getObject(); 

        // we look for MyRequest object, we ignore 
        // everything else that was sent by mistake! 
        // TODO: check if its possible to reject a message delivery. reject
the 
        // ones we are not looking for. 
        if (serializableObject instanceof MyRequest) { 
          Log.getLogger().log(Level.INFO, 
              Thread.currentThread().getName() + " received an object"); 
          MyRequest myRequest = (MyRequest) serializableObject; 
          MyWorker myWorker = new MyWorker( 
              myRequest); 
          // make this object an observer of this categorisor 
          MyWorker.addObserver(this); 
          threadPoolExecutor.execute(myWorker); 
          // increment running thread count, its inside synchronized as some 
          // other thread may call update and try to decrement the count at
the 
          // same time when this thread is incrementing it. 
          Log.getLogger().log( 
              Level.INFO, 
              Thread.currentThread().getName() 
                  + " Entering synchronised block..."); 
          synchronized (this) { 
            runningThreads++; 

            Log.getLogger().log( 
                Level.INFO, 
                Thread.currentThread().getName() + " Running: " 
                    + runningThreads + " Max:" + MAX_THREADS); 

            // now check if we are running max allowed thread for this
machine. 
            if (runningThreads == MAX_THREADS) { 
              Log.getLogger() 
                  .log( 
                      Level.INFO, 
                      Thread.currentThread().getName() 
                          + " Reached max threads... stoping feedback
message consumer!!"); 
              // stop receiving more messages. 
              stopConsumption(); 

            } 
          } 

          Log.getLogger().log( 
              Level.INFO, 
              Thread.currentThread().getName() 
                  + " out of synchronised block...."); 

        } 
      } catch (JMSException e) { 
        e.printStackTrace(); 
      } 
    } 
  } 


Please note that this class adds itself as a listner to MyWorker object. 
  // make this object an observer of this categorisor 
   MyWorker.addObserver(this);

MyWorker once complete, updates this class as below: 
public void update(Observable observableSource, 
      MyStatus status) { 
    // TODO: work with status object later. 
    if (observableSource instanceof MyWorker) { 
      MyWorker myWorker = (MyWorker) observableSource; 
      // notify observers about categorization being over. 
      notifyObservers((MyResult) myWorker.getResult()); 

      Log.getLogger().log(Level.INFO, 
          Thread.currentThread().getName() + ": Notified observers..."); 

      synchronized (this) { 
        Log.getLogger() 
            .log( 
                Level.INFO, 
                Thread.currentThread().getName() 
                    + ": One of the threads comepleted... starting message
consumer if max was reached!!"); 
        runningThreads--; 
        Log.getLogger().log( 
            Level.INFO, 
            Thread.currentThread().getName() + " Running: " + runningThreads 
                + " Max:" + MAX_THREADS); 
        // start consuming only when we were at max. need not call 
        // startConsumption on each thread completion. 
        if (runningThreads == MAX_THREADS - 1) { 
          startConsumption(); 
        } 
      } 
    } 
  } 

Since there are multiple threads running and they all can call update
together, i have used synchronized block to decrement runningThreads and
call startConsumption() method. 

It works for random N messages and then it stops at the call to 
stopConsumption(); 

Here is my start and stop consumption methods. 

  /** 
   * stops listing to the queue 
   */ 
  private void stopConsumption() { 
    try { 
      conn.stop(); 
    } catch (JMSException e) { 
      e.printStackTrace(); 
    } 
  } 

  /** 
   * start listing to the queue again. 
   */ 
  private void startConsumption() { 
    try { 
      conn.start(); 
    } catch (JMSException e) { 
      e.printStackTrace(); 
    } 
  } 

Looking at jconsole threads, i can see that it waits in the call to
conn.start() and keeps on waiting forever. 

Here is how i connect to the queue in one my run() method: 

  @Override 
  public void run() { 
    // create a activeMQ Queue consumer and register this 
    // object as a JMS message listner. 
    // try connecting to ActiveMQ broker 
    try { 
      javax.naming.Context ctx = ActiveMQInitialContext.getInstance(); 
      // lookup the connection factory 
      javax.jms.QueueConnectionFactory factory =
(javax.jms.QueueConnectionFactory) ctx 
         
.lookup(PropertyManager.getProperty("JMS_CONNECTION_FACTORY_NAME")); 
      
      conn = factory.createQueueConnection(); 
      conn.start(); 
      // lookup an existing queue 
      javax.jms.Queue myqueue = (Queue) ctx.lookup(PropertyManager 
          .getProperty("JMS_REQUEST_QUEUE_NAME")); 
      // create a new queueSession for the client 
      QueueSession session = conn.createQueueSession(false, 
          QueueSession.AUTO_ACKNOWLEDGE); 

      // create a new subscriber to receive messages 
      MessageConsumer subscriber = session.createConsumer(myqueue); 
      subscriber.setMessageListener(this); 

      Log.getLogger().log( 
          Level.INFO, 
          Thread.currentThread().getName() + ": Attached to: " 
              + PropertyManager.getProperty("JMS_REQUEST_QUEUE_NAME")); 

    } catch (NamingException e) { 
      e.printStackTrace(); 
    } catch (JMSException e) { 
      e.printStackTrace(); 
    } 
  } 

As you can see i am using Auto ACK. I have tested this on activemq 5.4.2 and
5.5.0 and it behaves the same on both versions. I am using java 6.

Here is a thread dump...for the waiting thread. 


Name: ActiveMQ Session Task 
State: WAITING on org.apache.activemq.thread.PooledTaskRunner$1@35e59bda 
Total blocked: 3  Total waited: 4 

Stack trace: 
java.lang.Object.wait(Native Method) 
org.apache.activemq.thread.PooledTaskRunner.shutdown(PooledTaskRunner.java:95) 
org.apache.activemq.thread.PooledTaskRunner.shutdown(PooledTaskRunner.java:102) 
org.apache.activemq.ActiveMQSessionExecutor.stop(ActiveMQSessionExecutor.java:155) 
   - locked org.apache.activemq.ActiveMQSessionExecutor@55f6bd6a 
org.apache.activemq.ActiveMQSession.stop(ActiveMQSession.java:1653) 
org.apache.activemq.ActiveMQConnection.stop(ActiveMQConnection.java:550) 
   - locked java.util.concurrent.CopyOnWriteArrayList@561d0c91 
com.qubit.command.DefaultQueueListenerCommand.stopConsumption(DefaultQueueListenerCommand.java:239)

com.qubit.command.DefaultQueueListenerCommand.onMessage(DefaultQueueListenerCommand.java:173)

   - locked com.qubit.command.UrlCategorisationReqConsumerCmd@fed453 
org.apache.activemq.ActiveMQMessageConsumer.dispatch(ActiveMQMessageConsumer.java:1229) 
   - locked java.lang.Object@37cd238a 
org.apache.activemq.ActiveMQSessionExecutor.dispatch(ActiveMQSessionExecutor.java:134) 
org.apache.activemq.ActiveMQSessionExecutor.iterate(ActiveMQSessionExecutor.java:205) 
org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:122) 
org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:43) 
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886) 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908) 
java.lang.Thread.run(Thread.java:662) 


And here is one other thread thats blocked for the same object.. 


Name: Thread-23 
State: BLOCKED on com.qubit.command.UrlCategorisationReqConsumerCmd@fed453
owned by: ActiveMQ Session Task 
Total blocked: 1,961  Total waited: 84,130 

Stack trace: 
com.qubit.command.DefaultQueueListenerCommand.update(DefaultQueueListenerCommand.java:215)

com.qubit.categoriser.URLCategoriser.notifyObservers(URLCategoriser.java:263) 
com.qubit.categoriser.URLCategoriser.categorise(URLCategoriser.java:194) 
com.qubit.categoriser.URLCategoriser.run(URLCategoriser.java:270) 
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886) 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908) 
java.lang.Thread.run(Thread.java:662)

--
View this message in context: http://activemq.2283324.n4.nabble.com/Stops-Message-Delivery-to-consumer-QueueConnection-Stop-Waits-forever-tp3830716p3830716.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.

Mime
View raw message