你的位置:首页 > Java教程

[Java教程]ActiveMQ(li)


一、ActiveMQ

  首先,ActiveMQ不是一个框架,它不是struct,webx,netty这种框架,它更像是tomcat服务器,因为你使用它之前必须启动它,activeMQ和JMS的关系有点类似于tomcat和servlet的关系,tomcat实现了servlet规范,同理activeMQ实现了JMS规范,它们都是服务器,在使用其服务之前需要先把服务器启动起来。

//#准备activeMQ的环境和tomcat的环境一样简单。//1.下载:wget http://apache.freelamp.com//activemq/apache-activemq/5.4.2/apache-activemq-5.4.2-bin.tar.gz//2.解压:tar xvf apache-activemq-5.4.2-bin.tar.gz//3.启动:进入bin目录执行activemq脚本,输入 http://localhost:8161,ActiveMQ服务器成功启动之后,我们就可以向这个服务器发送消息,或者从这个服务器接收消息了

  上面是Linux环境下的安装,解压和启动。下面是Windows环境下的ActiveMQ的配置和使用。http://activemq.apache.org/download.html下载稳定版本。

二、ActiveMQ发消息

import javax.jms.Connection;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory; public class SendMessage {   private static final String url     = "tcp://10.20.156.22:61616";  private static final String QUEUE_NAME  = "yunpeng.jiangyp";  protected String      expectedBody = "<hello>world!</hello>";   public void sendMessage() throws JMSException {     ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);    Connection connection = connectionFactory.createConnection();    connection.start();     Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);    Destination desion = session.createQueue(QUEUE_NAME);    MessageProducer producer = session.createProducer(desion);    TextMessage message = session.createTextMessage(expectedBody);    producer.send(message);    connection.close();  }   public static void main(String[] args) throws JMSException {    SendMessage sndMsg = new SendMessage();    sndMsg.sendMessage();  }}

这段代码很简单,就是向activemq服务器发送一条消息,当成功发送给activemq之后,可以在(http://10.20.156.22:8161/admin/queues.jsp)activemq服务器的控制台看到这条消息。控制台显示yunpeng.jiangyp堆积了2条消息。

三、ActiveMQ收消息

 

import javax.jms.Connection;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.Message;import javax.jms.MessageConsumer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory; public class ReceiveMessage {   private static final String url    = "tcp://10.20.156.22:61616";  private static final String QUEUE_NAME = "yunpeng.jiangyp";   public static void main(String[] args) throws JMSException {     ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);    Connection connection = factory.createConnection();    connection.start();     Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);    Destination desion = session.createQueue(QUEUE_NAME);    MessageConsumer consumer = session.createConsumer(desion);    while (true) {       Message message = consumer.receive(1000);      if (message == null) break;      if (message instanceof TextMessage) {        TextMessage textMsg = (TextMessage) message;        String msg = textMsg.getText();        System.out.println(msg);      }    }    connection.close();   }}

 

  这段代码也很简单,就是从activemq服务器接收一条消息,可以发现发送和接收消息都需要创建Connection,Session,Queue,唯一不同的是发送消息是创建MessageProducer,而接收消息是创建MessageConsumer,这也非常形象的证明了activemq的生产者和消费者模式。运行这段代码会打印出之前发送的消息信息,同时控制台会显示两条堆积的消息被消费。

  在JMS中两种发送消息的方式,包含point-to-point 和 publish/subscribe。

四、