12

EDIT: Rephrased the question:

I want to use ActiveMQ as a messenger service between my server and client applications.

I am trying to set up an embedded broker (i.e. not a separate process) within the server to handle the produced messages for my clients to consume. This queue is persisted.

The broker initialisation as follows:

BrokerService broker = new BrokerService();
KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter();
adaptor.setDirectory(new File("activemq"));
broker.setPersistenceAdapter(adaptor);
broker.setUseJmx(true);
broker.addConnector("tcp://localhost:61616");
broker.start();

After tinkering, I ended up with the server part being:

public static class HelloWorldProducer implements Runnable {
    public void run() {
        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost"); // apparently the vm part is all i need
            Connection connection = connectionFactory.createConnection(); 
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("TEST.FOO");
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            String text = "Hello world! From: " + Thread.currentThread().getName() + " : " + this.hashCode();
            TextMessage message = session.createTextMessage(text);
            System.out.println("Sent message: "+ message.hashCode() + " : " + Thread.currentThread().getName());
            producer.send(message);
            session.close();
            connection.close();
        }
        catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        }
    }
}

The client is very similar and looks like this:

public static class HelloWorldConsumer implements Runnable, ExceptionListener {
    public void run() {
        try {
          ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost");
            Connection connection = connectionFactory.createConnection(); // exception happens here...
            connection.start();
            connection.setExceptionListener(this);
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("TEST.FOO");
            MessageConsumer consumer = session.createConsumer(destination);
            Message message = consumer.receive(1000);
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                System.out.println("*****Received: " + text);
            } else {
                System.out.println("*****Received obj: " + message);
            }
            consumer.close();
            session.close();
            connection.close();
        } catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        }
    }

The main method simply starts each of these in a thread to start producing/receiving messages.

...but I am running into the following with the start of each thread:

2013-01-24 07:54:31,271 INFO  [org.apache.activemq.broker.BrokerService] Using Persistence Adapter: AMQPersistenceAdapter(activemq-data/localhost)
2013-01-24 07:54:31,281 INFO  [org.apache.activemq.store.amq.AMQPersistenceAdapter] AMQStore starting using directory: activemq-data/localhost
2013-01-24 07:54:31,302 INFO  [org.apache.activemq.kaha.impl.KahaStore] Kaha Store using data directory activemq-data/localhost/kr-store/state
2013-01-24 07:54:31,339 INFO  [org.apache.activemq.store.amq.AMQPersistenceAdapter] Active data files: []
2013-01-24 07:54:31,445 DEBUG [org.apache.activemq.broker.jmx.ManagementContext] Probably not using JRE 1.4: mx4j.tools.naming.NamingService
2013-01-24 07:54:31,450 DEBUG [org.apache.activemq.broker.jmx.ManagementContext] Failed to create local registry
    java.rmi.server.ExportException: internal error: ObjID already in use
    at sun.rmi.transport.ObjectTable.putTarget(ObjectTable.java:186)
    at sun.rmi.transport.Transport.exportObject(Transport.java:92)
    at sun.rmi.transport.tcp.TCPTransport.exportObject(TCPTransport.java:247)
    at sun.rmi.transport.tcp.TCPEndpoint.exportObject(TCPEndpoint.java:411)
    at sun.rmi.transport.LiveRef.exportObject(LiveRef.java:147)
        <snip....>

It seems like the messages are produced and consumed successfully (the other issues I previously posted about was resolved), but the above exception is worrying me.

EDIT: During broker shutdown, I am now also greeted by the following:

2013-01-25 08:40:17,486 DEBUG [org.apache.activemq.transport.failover.FailoverTransport] Transport failed with the following exception:
    java.io.EOFException
    at java.io.DataInputStream.readInt(DataInputStream.java:392)
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:269)
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:210)
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:202)
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
    at java.lang.Thread.run(Thread.java:722)
6
  • You should include all the broker creation code so we can see what all you've configured. You can disable JMX if you don't need it via broker.setUseJmx(false);
    – Tim Bish
    Commented Jan 23, 2013 at 14:46
  • To clarify, the message displayed is of level DEBUG. This is not necessarily an ERROR. It could just be informative. Are you experiencing an error in actually producing/consuming messages? The question is not clear on this point.
    – cmonkey
    Commented Jan 23, 2013 at 15:29
  • I've rephrased the question completely. Essentially I'm asking 3 sub-questions. (1) The exception, (2) Lost messages and (3) Persistence. Thanks for looking at my question. Commented Jan 24, 2013 at 5:44
  • It looks like you are using an extremely old version of ActiveMQ as its defaulting to AMQ persistence adapter, advise you switch to version 5.8.0.
    – Tim Bish
    Commented Jan 24, 2013 at 11:48
  • Do you want to run an embedded broker (i.e. ActiveMQ itself) inside your server or just a JMS producer/consumer? In the former case, please also post details on how you tried running that broker.
    – SirRichie
    Commented Jan 24, 2013 at 12:33

2 Answers 2

15

You can embed a broker into your code in a number of ways, much of which is documented here. You may want to try upgrading you version since what you are using appears to be quite old as it defaulting to the now deprecated AMQ Store instead of the newer KahaDB store. You might be having issues because of a race between the client threads as they use the different connection factories which could race to create in VM brokers. If you set the create=false option on the producer and ensure the consumer thread starts after that could address the issue, or you could create the VM broker ahead of time and the add create=false to both thread's and that might do the trick.

BrokerService broker = new BrokerService();
// configure the broker
broker.setBrokerName("localhost");
broker.setUseJmx(false);
broker.start();

And then in the client code just attach via this connection factory configuration.

ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?create=false");
3
  • Thanks, I have it as such. However, I am still worried about the exception above. Do you know whether it can be safely ignored? Commented Jan 25, 2013 at 6:36
  • 1
    The exception looks to be telling you that the client connection failed because you shutdown the broker while the client was still connected. Not a huge problem if you are shutting down the app. It looks like your client is using TCP instead of VM in its connection factory, though its not entirely clear from the question.
    – Tim Bish
    Commented Jan 25, 2013 at 11:35
  • Yes... that is indeed the issue and the exception can be safely ignored. To avoid it, the clients needs to be shut down before the server (which include the broker). Commented Oct 15, 2013 at 10:31
4

When I run your code, I got the below exception:

javax.jms.JMSException: Could not connect to broker URL: tcp://localhost. 
Reason java.lang.IllegalArgumentException: port out of range:-1

Your broker is running and listening to port 61616, so any client which tries to connect to broker need to have the port in its URL.

The client code tries to connect to localhost but doesn't mention the port to which it has to connect. Both the producer and consumer code needs to be fixed.

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost");

To

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

After fixing the port, I was able to run your code.

1
  • Thanks Satish. I was able to get this working. Apologies, I should have closed the question. Commented Feb 18, 2013 at 12:11

Not the answer you're looking for? Browse other questions tagged or ask your own question.