ActiveMQ 和 JMS : Basic steps for novice

Question

大家好，请为新手提供一些关于带有 JMS 的 ActiveMQ 的基础知识。而且配置步骤也一样。

Answer 1

我们将使用多线程创建一个基于控制台的应用程序。所以为控制台应用程序创建一个java项目。

现在按照这些步骤......

将 javax.jms.jar、activemq-all-5.3.0.jar、log4j-1.2.15.jar 添加到您的项目库中。
(您可以从 http://www.jarfinder.com/ 下载上述所有 jar 文件。

创建一个名为 jndi.properties 的文件并粘贴以下这些文本..(jndi.properties 的详细信息只需谷歌一下)

# START SNIPPET: jndi

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory

# use the following property to configure the default connector
java.naming.provider.url = tcp://localhost:61616

# use the following property to specify the JNDI name the connection factory
# should appear as.
#connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry
connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry

# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.MyQueue = example.MyQueue


# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
topic.MyTopic = example.MyTopic

# END SNIPPET: jndi

添加 JMSConsumer.java

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class JMSConsumer implements Runnable{
    private static final Log LOG = LogFactory.getLog(JMSConsumer.class);

    public void run() {
        Context jndiContext = null;
        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        Destination destination = null;
        String sourceName = null;
        final int numMsgs; 
        sourceName= "MyQueue";
        numMsgs = 1;
        LOG.info("Source name is " + sourceName);
        /*
         * Create a JNDI API InitialContext object
         */
        try {
            jndiContext = new InitialContext();
        } catch (NamingException e) {
            LOG.info("Could not create JNDI API context: " + e.toString());
            System.exit(1);
        }

        /*
         * Look up connection factory and destination.
         */
        try {
            connectionFactory = (ConnectionFactory)jndiContext.lookup("queueConnectionFactory");
            destination = (Destination)jndiContext.lookup(sourceName);
        } catch (NamingException e) {
            LOG.info("JNDI API lookup failed: " + e);
            System.exit(1);
        }


        try {
            connection = connectionFactory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
            consumer = session.createConsumer(destination);
            connection.start();
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            MessageListener listener = new MyQueueMessageListener();
            consumer.setMessageListener(listener ); 
            //Let the thread run for some time so that the Consumer has suffcient time to consume the message
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        } catch (JMSException e) {
            LOG.info("Exception occurred: " + e);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }

    }

添加 JMSProducer.java

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class JMSProducer implements Runnable{
private static final Log LOG = LogFactory.getLog(JMSProducer.class);

public JMSProducer() {
}

//Run method implemented to run this as a thread.
public void run(){
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageProducer producer = null;
String destinationName = null;
final int numMsgs; 
destinationName = "MyQueue";
numMsgs = 5;
LOG.info("Destination name is " + destinationName);

/*
* Create a JNDI API InitialContext object
*/
try {
    jndiContext = new InitialContext();
} catch (NamingException e) {
    LOG.info("Could not create JNDI API context: " + e.toString());
    System.exit(1);
}

/*
* Look up connection factory and destination.
*/
try {
    connectionFactory = (ConnectionFactory)jndiContext.lookup("queueConnectionFactory");
    destination = (Destination)jndiContext.lookup(destinationName);
} catch (NamingException e) {
    LOG.info("JNDI API lookup failed: " + e);
    System.exit(1);
}

/*
* Create connection. Create session from connection; false means
* session is not transacted.create producer, set the text message, set the co-relation id and send the message.
*/
try {
    connection = connectionFactory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    producer = session.createProducer(destination);
    TextMessage message = session.createTextMessage();
    for (int i = 0; i 

---

Add MyQueueMessageListener.java

import java.io.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.*;


public class MyQueueMessageListener implements MessageListener {
    private static final Log LOG = LogFactory.getLog(MyQueueMessageListener.class);
    /**
    *
    */
    public MyQueueMessageListener() {
    // TODO Auto-generated constructor stub
    }

    /** (non-Javadoc)
    * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
    * This is called on receving of a text message.
    */
    public void onMessage(Message arg0) {
        LOG.info("onMessage() called!");
        if(arg0 instanceof TextMessage){
            try {
                //Print it out
                System.out.println("Recieved message in listener: " + ((TextMessage)arg0).getText());

                System.out.println("Co-Rel Id: " + ((TextMessage)arg0).getJMSCorrelationID());
                try {
                    //Log it to a file
                    BufferedWriter outFile = new BufferedWriter(new FileWriter("MyQueueConsumer.txt"));
                    outFile.write("Recieved message in listener: " + ((TextMessage)arg0).getText());
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            } catch (JMSException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }else{
            System.out.println("~~~~Listener : Error in message format~~~~");
        }

    }

    }

---

Add SimpleApp.java

public class SimpleApp {

    //Run the producer first, then the consumer
    public static void main(String[] args) throws Exception {
        runInNewthread(new JMSProducer());
        runInNewthread(new JMSConsumer()); 
    }

    public static void runInNewthread(Runnable runnable) {
        Thread brokerThread = new Thread(runnable);
        brokerThread.setDaemon(false);
        brokerThread.start();
    }

}

现在运行 SimpleApp.java 类。

一切都好。快乐编码。