ActiveMQ KahaDB always locking and waiting

markgiaconia :

I am trying to use ActiveMQ in a relatively simple work queue use case. I have one queue, and have a simple Producer and Consumer. My question is what am I doing wrong that continuously makes the DB lock? here is the message I get continously:

14/04/05 18:14:13 INFO store.SharedFileLocker: Database activemq-data\localhost\KahaDB\lock is locked... waiting 10 seconds for the database to be unlocked. Reason: java.io.IOException: File 'activemq-data\localhost\KahaDB\lock' could not be locked.

I am running the producer and consumer in separate threads at the same time. Initially, I had the connection at the class level, so I thought that was the problem, but even creating the connection from scratch every call to put and get still causes locking.

I have done research but have failed to find a solution. I am running ActiveMQ 5.9.0 on windows 7.

Here is what prints out when I start it via cmd prompt:

C:\activemq\apache-activemq-5.9.0\bin>activemq
Java Runtime: Oracle Corporation 1.7.0_40 C:\Program Files\Java\jre7
  Heap sizes: current=1005568k  free=995061k  max=1005568k
    JVM args: -Dcom.sun.management.jmxremote -Xms1G -Xmx1G -Djava.util.logging.config.file=logging.properties -Dhawtio.realm=activemq -Dhawtio.role=admins -Dhawtio.rolePrincipalCla
vemq.jaas.GroupPrincipal -Djava.security.auth.login.config=C:\activemq\apache-activemq-5.9.0\bin\..\conf\login.config -Dactivemq.classpath=C:\activemq\apache-activemq-5.9.0\bin\..\
che-activemq-5.9.0\bin\../conf;C:\activemq\apache-activemq-5.9.0\bin\../conf; -Dactivemq.home=C:\activemq\apache-activemq-5.9.0\bin\.. -Dactivemq.base=C:\activemq\apache-activemq-5
mq.conf=C:\activemq\apache-activemq-5.9.0\bin\..\conf -Dactivemq.data=C:\activemq\apache-activemq-5.9.0\bin\..\data -Djava.io.tmpdir=C:\activemq\apache-activemq-5.9.0\bin\..\data\t
Extensions classpath:
  [C:\activemq\apache-activemq-5.9.0\bin\..\lib,C:\activemq\apache-activemq-5.9.0\bin\..\lib\camel,C:\activemq\apache-activemq-5.9.0\bin\..\lib\optional,C:\activemq\apache-activemq
b,C:\activemq\apache-activemq-5.9.0\bin\..\lib\extra]
ACTIVEMQ_HOME: C:\activemq\apache-activemq-5.9.0\bin\..
ACTIVEMQ_BASE: C:\activemq\apache-activemq-5.9.0\bin\..
ACTIVEMQ_CONF: C:\activemq\apache-activemq-5.9.0\bin\..\conf
ACTIVEMQ_DATA: C:\activemq\apache-activemq-5.9.0\bin\..\data
Loading message broker from: xbean:activemq.xml
 INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@5bf2a8f5: startup date [Sat Apr 05 17:42:42 EDT 2014]; root of context hierarchy
 INFO | PListStore:[C:\activemq\apache-activemq-5.9.0\bin\..\data\localhost\tmp_storage] started
 INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[C:\activemq\apache-activemq-5.9.0\bin\..\data\kahadb]
 INFO | KahaDB is version 5
 INFO | Recovering from the journal ...
 INFO | Recovery replayed 6935 operations from the journal in 0.416 seconds.
 INFO | Apache ActiveMQ 5.9.0 (localhost, ID:Owner-PC-49614-1396734165637-0:1) is starting
 INFO | Listening for connections at: tcp://Owner-PC:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
 INFO | Connector openwire started
 INFO | Listening for connections at: amqp://Owner-PC:5673?maximumConnections=1000&wireFormat.maxFrameSize=104857600
 INFO | Connector amqp started
 INFO | Listening for connections at: stomp://Owner-PC:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600
 INFO | Connector stomp started
 INFO | Listening for connections at: mqtt://Owner-PC:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600
 INFO | Connector mqtt started
 INFO | Listening for connections at ws://Owner-PC:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
 INFO | Connector ws started
 INFO | Apache ActiveMQ 5.9.0 (localhost, ID:Owner-PC-49614-1396734165637-0:1) started
 INFO | For help or more information please see: http://activemq.apache.org
 INFO | Welcome to hawtio 1.2-M23 : http://hawt.io/ : Don't cha wish your console was hawt like me? ;-)
 INFO | Starting hawtio authentication filter, JAAS realm: "activemq" authorized role: "admins" role principal classes: "org.apache.activemq.jaas.GroupPrincipal"
 INFO | Using file upload directory: C:\activemq\apache-activemq-5.9.0\bin\..\data\tmp\uploads
 INFO | jolokia-agent: Using access restrictor classpath:/jolokia-access.xml
 INFO | ActiveMQ WebConsole available at http://localhost:8161/
 INFO | Initializing Spring FrameworkServlet 'dispatcher'

Here is my java to implement the Queue as ActiveMQ (I have a Queue interface and this is an Impl)

import com.google.gson.Gson;
import java.util.Set;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQImpl implements Queue, ExceptionListener {

  private String host;
  private String user;
  private String pw;

  public void init() {
  }

  public void close() { 
  }

  public Message get() {
    Message outMessage = null;
    ActiveMQConnectionFactory connectionFactory = null;
    Connection connection = null;  
    try {
      if (connection == null) {
        connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

        // Create a Connection
        connection = connectionFactory.createConnection();
      }
      connection.start();

      // Create a Session
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      // Create the destination (Topic or Queue)
      Destination destination = session.createQueue("work");
      // Create a MessageConsumer from the Session to the Topic or Queue
      MessageConsumer consumer = session.createConsumer(destination);

      // Wait for a message
      javax.jms.Message message = consumer.receive(1000);

      if (message instanceof TextMessage) {
        TextMessage textMessage = (TextMessage) message;
        String text = textMessage.getText();
        outMessage = new Gson().fromJson(text, Message.class);
        //   System.out.println("Received: " + text);
      } else {
        // System.out.println("Received: " + message);
      }

      consumer.close();
      //  session.commit();
      session.close();
      //connection.close();
    } catch (Exception e) {
      System.out.println("Caught: " + e);
      e.printStackTrace();
    }
    return outMessage;
  }

  public void put(Message inMessage) {
    try {
      ActiveMQConnectionFactory connectionFactory = null;


      Connection prodConnection = null;
      connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

      prodConnection = connectionFactory.createConnection();

      prodConnection.start();

      Session session = prodConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      Destination destination = session.createQueue("work");

      MessageProducer producer = session.createProducer(destination);
      producer.setDeliveryMode(DeliveryMode.PERSISTENT);
      producer.setTimeToLive(60000);
      // Create a messages
      String text = inMessage.toString();
      TextMessage message = session.createTextMessage(text);

      // Tell the producer to send the message
      System.out.println("Sent message: " + text);
      producer.send(message);
      producer.close();
      // session.commit();
      session.close();
      prodConnection.close();
    } catch (Exception e) {
      System.out.println("Caught: " + e);
      onException(null);
      e.printStackTrace();
    }
  }

  public void onException(JMSException jmse) {
    //send this to the error channel object... 
    System.out.println(jmse);
  }

  public void put(Set<Message> messages) {
    try {
  ActiveMQConnectionFactory connectionFactory=null;
  Connection connection=null;

  Connection prodConnection=null;
      if (connection == null) {
        connectionFactory = new ActiveMQConnectionFactory("vm://localhost?jms.useAsyncSend=true");

        // Create a Connection
        connection = connectionFactory.createConnection("admin", "admin");
      }

      connection.start();

      // Create a Session
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      // Create the destination (Topic or Queue)
      Destination destination = session.createQueue("work");
      // Create a MessageProducer from the Session to the Topic or Queue
      MessageProducer producer = session.createProducer(destination);
      producer.setDeliveryMode(DeliveryMode.PERSISTENT);

      // Create a messages
      for (Message inMessage : messages) {
        String text = inMessage.toString();
        TextMessage message = session.createTextMessage(text);

        // Tell the producer to send the message
        System.out.println("Sent message: " + text);
        producer.send(message);
      }
      producer.close();
      // session.commit();
      session.close();
      //connection.close();
    } catch (Exception e) {
      System.out.println("Caught: " + e);
      onException(null);
      e.printStackTrace();
    }
  }

}

Here are the producers and consumers (simple debugging classes)

public class Producer {

  public static void main(String[] args) {
    Queue q = QueueFactory.create(QueueType.ACTIVEMQ);
    try {
      for (int i = 0; i < 10; i++) {

        q.put(new Message("testimpl" + i, "whatever", i));
        Thread.sleep(1000);

      }
    } catch (InterruptedException ex) {
      Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
    }

  }
}


public class Consumer {

  public static void main(String[] args) {

    Queue q = QueueFactory.create(QueueType.ACTIVEMQ);
    try {
      while (true) {

        Message get = q.get();
        System.out.println(get);
        Thread.sleep(1000);

      }
    } catch (InterruptedException ex) {
      Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
    }
  }
}

any help is appreciated.

dev2d :

do following

kill java process for active mq, open task manager -> go to process -> check for java -> and right click and say end process

then go to <activemq_install_directory>/data/kahadb

and delete lock file which gets generated

try starting activemq again

for me this works every time I face this issue

Collected from the Internet

Please contact [email protected] to delete if infringement.

edited at
0

Comments

0 comments
Login to comment

Related

ActiveMQ won't restart because KahaDB is locked

What is the maximum number of xxx of db-xxx in kahaDB (ActiveMQ)?

ConnectionState always waiting

Waiting Results not always the same

Does ActiveMQ store non-persistent message for inactive durable message subscriber in kahadb?

Why ActiveMQ doesn't and console still shows messages after deleting db-*.log files from KahaDB

snapshot.ConnectionState always waiting

node is always ""waiting for localhost" in browser

Rails - Optimistic locking always fires StaleObjectError exception

ActiveMQ JDBC persistence trouble: ACTIVEMQ_MSGS is always empty

Akka actors always times out waiting for future

TCP Socket send/receive always waiting

Async & await always returns "Waiting for completion"

Selenium always times out waiting for server to start

App review status(always "waiting for upload") not change

KahaDb properties in Grails

istio-ingressgateway always Waiting for Istio Pilot information

Can't start MongoDB, mongod is always stop/waiting

Task always on status Waiting for Activation (finding all serial devices)

fgets() always waiting for input, can't find how to fix it

Can't start MongoDB , mongod is always stop/waiting

Shutdown is slow, always waiting 2 minutes for gvfsd-fuse to terminate

What is the best way to handle waiting for elements that are not always present with Selenium?

Migrate data from Kahadb to MKahadb

How to use always newest cookbook while some other keep locking with Berkshelf

java: If notify() is always called before the lock release, how could a waiting threads acquire this same lock?

Why python debugger always get this timeout waiting for response on 113 when using Pycharm?

Is there way to see kahaDB logs in AWS using amazonMQ?

ConcurrentHashMap locking