ActiveMQ入门

1.安装ActiveMQ
 首先去官方网站下载http://activemq.apache.org/download.html  apache-activemq-4.1-incubator.zip 进行解压
    结构如下：
       +bin       (windows下面的bat和unix/linux下面的sh)
       +conf      (activeMQ配置目录,包含最基本的activeMQ配置文件)
       +data      (默认是空的)
       +docs      (index,replease版本里面没有文档,-.-b不知道为啥不带)
       +example   (几个例子
       +lib       (activemMQ使用到的lib)
       -apache-activemq-4.1-incubator.jar  (ActiveMQ的binary)
       -LICENSE.txt      
       -NOTICE.txt       
       -README.txt
       -user-guide.html

2.配置jdk和ActiveMQ环境变量

 

3.运行%ACTIVEMQ_HOME% /bin 下的 activemq.bat启动文件出现如下信息：

如果看到该信息，表示启动成功。

 

JMS源于企业应用对于消息中间件的需求，使应用程序可以通过消息进行异步处理而互不影响。Sun公司和它的合作伙伴设计的JMS API定义了一组公共的应用程序接口和相应语法，使得Java程序能够和其他消息组件进行通信。JMS有四个组成部分：JMS服务提供者、消息管理对象、消息的生产者消费者和消息本身。
1)JMS服务提供者实现消息队列和通知，同时实现消息管理的API。JMS已经是J2EE API的一部分，J2EE服务器都提供JMS服务。
2) 消息管理对象提供对消息进行操作的API。JMS API中有两个消息管理对象：创建jms连接使用的工厂（ConnectionFactory）和目的地（Destination），根据消息的消费方式的不同ConnectionFactory可以分为QueueConnectionFactory和TopicConnectionFactory，目的地（Destination）可以分为队列（Queue）和主题（Topic）两种。
3)消息的生产者和消费者。消息的产生由JMS的客户端完成，JMS服务提供者负责管理这些消息，消息的消费者可以接收消息。消息的生产者可以分为――点对点消息发布者（P2P）和主题消息发布者（TopicPublisher）。所以，消息的消费者分为两类：主题消息的订阅者（TopicSubscriber)和点对点消息的接收者（queue receiver）
4)消息。消息是服务提供者和客户端之间传递信息所使用的信息单元。JMS消息由以下三部分组成：
　　消息头（header）――JMS消息头包含了许多字段，它们是消息发送后由JMS提供者或消息发送者产生，用来表示消息、设置优先权和失效时间等等，并且为消息确定路由。
　　属性（property）――用来添加删除消息头以外的附加信息。
　　消息体（body）――JMS中定义了5种消息体：ByteMessage、MapMessage、ObjectMessage、StreamMessage和TextMessage。

2.Messages 通信方式
上面提到JMS通信方式分为点对点通信和发布/订阅方式

 

进行相关准备工作后，分别编写服务端和客户端。

 

服务端

public class SendMessage {
 private static final String URL_STRING = "tcp://localhost:61616";
 private static final String QUEUE_NAME = "choice.queue";
 protected String expectedBody = "<hello>hello world!</hello>";
 protected Connection connection = null;

 public void sendMessage() throws JMSException {
  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    URL_STRING);
  connection = (Connection) connectionFactory.createConnection();
  try {
   connection.start();
   Session session = connection.createSession(false,
     Session.AUTO_ACKNOWLEDGE);
   Destination destination = session.createQueue(QUEUE_NAME);
   MessageProducer messageProducer = session
     .createProducer(destination);
   TextMessage textMessage = session.createTextMessage(expectedBody);
   textMessage.setStringProperty("headname", "remoteB");
   messageProducer.send(textMessage);
  } catch (Exception e) {
   e.printStackTrace();
   close();
  }
 }

 public void close() {
  try {
   if (null != connection) {
    connection.close();
   }
  } catch (JMSException e) {
   e.printStackTrace();
  }

 }

 public static void main(String[] args) {
  SendMessage sendMessage = new SendMessage();
  try {
   sendMessage.sendMessage();
  } catch (JMSException e) {
   e.printStackTrace();
  }
 }
}

 

客户端

public class ReceiveMessage {
 private static final String url = "tcp://localhost:61616";
 private static final String QUEUE_NAME = "choice.queue";
 private void receiveMessage() {
  Connection connection = null;
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    url);
  try {
   connection = activeMQConnectionFactory.createConnection();
   connection.start();
   Session session = connection.createSession(false,
     Session.AUTO_ACKNOWLEDGE);
   Destination destination = session.createQueue(QUEUE_NAME);
   MessageConsumer messageConsumer = session
     .createConsumer(destination);
   consumeMessagesAndClose(connection, session, messageConsumer);
  } catch (JMSException e) {
   e.printStackTrace();
  }
 }

 protected void consumeMessagesAndClose(Connection connection,
   Session session, MessageConsumer messageConsumer)
   throws JMSException {
  for (int i = 0; i < 100000;) {
   Message message = messageConsumer.receive(1000);
   if (null != message) {
    i++;
    onMessage(message);
   }
  }
  messageConsumer.close();
  session.close();
  connection.close();

 }

 public void onMessage(Message message) throws JMSException {
  if (message instanceof TextMessage) {
   TextMessage textMessage = (TextMessage) message;
   String msg = textMessage.getText();
   System.out.println("Received : " + msg);
  }
 }

 public static void main(String[] args) {
  ReceiveMessage receiveMessage = new ReceiveMessage();
  receiveMessage.receiveMessage();
 }
}

 

首先启动ActiveMQ的服务器，分别运行SendMessage和ReceiveMessage中的main方法即可输出

 

Received : <hello>hello world!</hello>