ActiveMQ Hello Word 簡單應用例子編寫

下面是個非常簡單的 JMS應用程式,它含有多個消費者和生産者,在初始化配置引導裡檢視如何正确的設定你得classpath

運作玩這個例子後,你可能需要做得事情:

1.配置一個broker,進而代替直接使用org.activemq.borker.impl.Main 這個類

2.是用JNDI去查找一個javax.jms.ConnectionFactory ,而不是直接建立一個 ActiveMQConnectionFactory .

3.實作javax.jms.MessageListener 接口 而不是 直接調用 consumer 的reveive()方法

4.使用一個帶事物的session

5.使用一個主題,而不是使用一個序列

這個例子的重點在于 向你展示直接使用JMS程式設計時必須得基本代碼.這些消費者和生産者可以很容易的完全部署到不同的機器上,或者是不同的流程中.

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

/**
 * Hello world!
 */
public class App {

    public static void main(String[] args) throws Exception {
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        Thread.sleep(1000);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        Thread.sleep(1000);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldProducer(), false);
        Thread.sleep(1000);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldConsumer(), false);
        thread(new HelloWorldProducer(), false);
    }

    public static void thread(Runnable runnable, boolean daemon) {
        Thread brokerThread = new Thread(runnable);
        brokerThread.setDaemon(daemon);
        brokerThread.start();
    }

    public static class HelloWorldProducer implements Runnable {
        public void run() {
            try {
                // 建立一個連接配接工廠
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

                // 建立一個連接配接
                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();
            }
        }
    }

    public static class HelloWorldConsumer implements Runnable, ExceptionListener {
        public void run() {
            try {

                // 建立一個連接配接工廠
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

                // 建立一個連接配接
                Connection connection = connectionFactory.createConnection();
                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: " + message);
                }

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

        public synchronized void onException(JMSException ex) {
            System.out.println("JMS Exception occured.  Shutting down client.");
        }
    }
}
           

以上内容翻譯自: http://activemq.apache.org/version-5-hello-world.html