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