ActiveMQ的使用

ActiveMQ使用分为两大块：生产者和消费者

一、准备

项目导入jar包：activemq-all-5.15.3.jar

并buildpath 

 

二、生产者

1. 创建连接工厂

ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL);

注：

userName是ActiveMQ的用户名，默认可以通过：ActiveMQConnection.DEFAULT_USER

password是ActiveMQ的密码，默认可以通过： ActiveMQConnection.DEFAULT_PASSWORD

brokerURL是ActiveMQ的连接，指定格式为：tcp://主机名:61616

 

1. 获取连接

connection = mqf.createConnection();

 

1. 生成会话

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

 

1. 生成对应的topic

Destination destination = session.createTopic("mytopic");

 

1. 创建生产者

MessageProducer producer = session.createProducer(destination);

 

1. 设置发送消息使用的模式

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

默认是：DeliveryMode.PERSISTENT

 

1. 生成消息

TextMessage msg = session.createTextMessage(“message");

 

1. 启动连接

connection.start();

 

1. 发送消息

producer.send(msg);

 

1. 关闭生产者

producer.close();

 

1. 关闭会话

session.close();

 

1. 关闭连接

connection.close();

 

三、消费者

1. 继承接口

MessageListener

ExceptionListener

并实现onException(JMSException exception)和onMessage(Message message)方法

 

1. 创建连接工厂

ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL);

具体参数同上

 

1. 获取连接

Connection connection = mqf.createConnection();

 

1. 生成会话

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

 

1. 生成对应的topic

Destination destination = session.createTopic("mytopic”);

 

1. 创建消费者

MessageConsumer consumer = session.createConsumer(destination);

 

1. 启动连接

connection.start();

 

1. 设置消息监听

consumer.setMessageListener(this);

 

1. 设置异常监听

connection.setExceptionListener(this);

 

1. 实现onMessage方法

改方法有一个参数Message message,这个参数是从ActiveMQ上拿到的消息，可以通过如下方法解析出来：

TextMessage tm = (TextMessage)message;

String result = tm.getText();

 

1. 关闭消费者

consumer.close();

 

1. 关闭会话

session.close();

 

1. 关闭连接

connection.close();

 

四、例程

1. 生产者实现程序

 package activemq_test;

 import javax.jms.Connection;

 import javax.jms.DeliveryMode;

 import javax.jms.Destination;

 import javax.jms.JMSException;

 import javax.jms.MessageProducer;

 import javax.jms.Session;

 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnection;

 import org.apache.activemq.ActiveMQConnectionFactory;

 public class Producer_tool {

     private final static String userName = ActiveMQConnection.DEFAULT_USER;

     private final static String password = ActiveMQConnection.DEFAULT_PASSWORD;

     private final static String brokerURL = "tcp://192.168.0.5:61616";

     private MessageProducer producer = null;

     private Connection connection = null;

     private Session session = null;

     public void initialize() throws JMSException {

         ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL);

         connection = mqf.createConnection();

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

         Destination destination = session.createTopic("mytopic");

         producer = session.createProducer(destination);

         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

     }

     public void send(String message) throws JMSException {

         initialize();

         TextMessage msg = session.createTextMessage(message);

         System.out.println("sending message: " + message);

         connection.start();

         producer.send(msg);

     }

     public void close() throws JMSException {

         if(producer != null) {

             producer.close();

         }

         if(session != null) {

             session.close();

         }

         if(connection != null) {

             connection.close();

         }

         System.out.println("closed");

     }

 }

1. 生产者主程序

 package activemq_test;

 import javax.jms.JMSException;

 public class Producer_test {

     public static void main(String[] args) throws JMSException {

         Producer_tool producer = null;

         for(int i = 0; i < 10; i++) {

             producer = new Producer_tool();

             producer.send("message" + i);

             producer.close();

         }

     }

 }

 

1. 消费者实现程序

 package activemq_test;

 import javax.jms.Connection;

 import javax.jms.Destination;

 import javax.jms.ExceptionListener;

 import javax.jms.JMSException;

 import javax.jms.Message;

 import javax.jms.MessageConsumer;

 import javax.jms.MessageListener;

 import javax.jms.Session;

 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnection;

 import org.apache.activemq.ActiveMQConnectionFactory;

 public class Consumer_tool implements MessageListener,ExceptionListener{

     private final static String userName = ActiveMQConnection.DEFAULT_USER;

     private final static String password = ActiveMQConnection.DEFAULT_PASSWORD;

     private final static String brokerURL = "tcp://192.168.0.5:61616";

     private Connection connection = null;

     private Session session = null;

     private MessageConsumer consumer = null;

     static boolean isConnection = false;

     public void initialize() throws JMSException {

         ActiveMQConnectionFactory mqf = new ActiveMQConnectionFactory(userName, password, brokerURL);

         connection = mqf.createConnection();

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

         Destination destination = session.createTopic("mytopic");

         consumer = session.createConsumer(destination);

     }

     public void consumeMessage() throws JMSException {

         initialize();

         connection.start();

         consumer.setMessageListener(this);

         connection.setExceptionListener(this);

         isConnection = true;

         System.out.println("consumer is listening");

     }

     @Override

     public void onException(JMSException exception) {

         isConnection = false;

     }

     @Override

     public void onMessage(Message message) {

         if(message instanceof TextMessage) {

             TextMessage tm = (TextMessage)message;

             try {

                 System.out.println("consumer received " + tm.getText());

             } catch (JMSException e) {

                 e.printStackTrace();

             }

         }

         else {

             System.out.println(message);

         }

     }

     public void close() throws JMSException {

         if(consumer != null) {

             consumer.close();

         }

         if(session != null) {

             session.close();

         }

         if(connection != null) {

             connection.close();

         }

         System.out.println("consumer has closed");

     }

 }

 

1. 消费者主程序

 package activemq_test;

 import javax.jms.JMSException;

 public class Consumer_test {

     public static void main(String[] args) throws JMSException {

         Consumer_tool consumer = new Consumer_tool();

         consumer.consumeMessage();

         while(Consumer_tool.isConnection) {

         }

         consumer.close();

     }

 }