ActiviMQ快速入门

下载ActiviMQ

 

启动mq

 

创建项目

 

写提供者

package com.cxl.activemq;

 

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

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;

 

/**

 * 消息发布者

 * @author Administrator

 *

 */

public class JMSProducer {

  

   private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; //默认的连接用户名

   private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; //默认的连接密码

   private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址

   private static final int SENDNUM = 10;//发送消息数量

  

   @SuppressWarnings("null")

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

     

      ConnectionFactory connectionFactory;//连接工程

      Connection connection; // 连接

      Session session;// 会话

      Destination destination;//消息的目的地

      MessageProducer messageProducer; // 消息生产者

     

      //实例连接工厂

      connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME,JMSProducer.PASSWORD,JMSProducer.BROKEURL);

     

      connection = connectionFactory.createConnection(); //通过连接工厂获取连接

      connection.start(); //启动连接

      session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); //创建session

      //destination = session.createQueue("FirstQueue1"); // 创建消息队列

      destination = session.createTopic("FirstTopic1");

      messageProducer = session.createProducer(destination); // 创建消息生产者

      sendMessage(session, messageProducer); //发送消息

      session.commit();

     

     

      if (connection != null) {

        connection.close();

      }

   }

 

  

   public static void sendMessage(Session session,MessageProducer producer) throws JMSException {

      for(int i = 0;i<JMSProducer.SENDNUM;i++) {

        TextMessage message = session.createTextMessage("发送消息>>> " + i);

        System.err.println("发步消息： " + "ActiveMQ 发送的消息 " + i);

        producer.send(message);

      }

   }

  

  

}

 

 

 

写消费者

package com.cxl.activemq;

 

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MessageConsumer;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

 

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

 

/**

 * 消息订阅者2

 * @author Administrator

 *

 */

public class JMSConsumer2 {

  

   private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; //默认的连接用户名

   private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; //默认的连接密码

   private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址

  

  

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

     

      ConnectionFactory connectionFactory;//连接工程

      Connection connection; // 连接

      Session session;// 会话

      Destination destination;//消息的目的地

      MessageConsumer messageConsumer;

     

      //实例连接工厂

      connectionFactory = new ActiveMQConnectionFactory(JMSConsumer2.USERNAME,JMSConsumer2.PASSWORD,JMSConsumer2.BROKEURL);

     

      connection = connectionFactory.createConnection(); //通过连接工厂获取连接

      connection.start(); //启动连接

      session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); //创建session

      //destination = session.createQueue("FirstQueue1");//创建连接的消息队列

      destination = session.createTopic("FirstTopic1");

      messageConsumer = session.createConsumer(destination);

     

      //注册消息监听

      messageConsumer.setMessageListener(new Listener2());

     

   }

  

  

 

}

 

 

 

写监听器

package com.cxl.activemq;

 

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

 

/**

 * 消息监听-订阅者2

 * @author Administrator

 *

 */

public class Listener2 implements MessageListener{

 

   @Override

   public void onMessage(Message message) {

     

      try {

        System.out.println("订阅者2收到的消息 ： "

        + ((TextMessage)message).getText());

       

      } catch (JMSException e) {

        // TODO Auto-generated catch block

        e.printStackTrace();

      }

     

   }

 

}

 

 

 

 

启动消费者

启动提供者

 

查看消费者

完事