ActiveMQ异步发送JMS消息实例应用

服务端:

package cn.com.vnvtrip.apachemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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;

/**
 * 消息的生产者（发送者）
 *
 * @author longgangbai
 *
 */
public class JmsSender {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：连接工厂，JMS 用它创建连接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "vm://localhost");

  // JMS 客户端到JMS Provider 的连接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一个发送或接收消息的线程
  Session session = connection.createSession(Boolean.TRUE,
    Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息发送给谁.
  // 获取session注意参数值my-queue是Query的名字
  Destination destination = session.createQueue("my-queue");
  // MessageProducer：消息生产者
  MessageProducer producer = session.createProducer(destination);
  // 设置不持久化
  producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  // 发送一条消息
  sendMsg(session, producer);
  session.commit();
  connection.close();
 }

 /**
  * 在指定的会话上，通过指定的消息生产者发出一条消息
  *
  * @param session
  *            消息会话
  * @param producer
  *            消息生产者
  */
 public static void sendMsg(Session session, MessageProducer producer)
   throws JMSException {
  // 创建一条文本消息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！");
  // 通过消息生产者发出消息
  producer.send(message);
  System.out.println("message=" + message);
 }
}

客户端:

package cn.com.vnvtrip.apachemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息的消费者（接受者）
 *
 * @author longgangbai
 *
 */
public class JmsReceiver {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：连接工厂，JMS 用它创建连接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "vm://localhost");

  // JMS 客户端到JMS Provider 的连接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一个发送或接收消息的线程
  Session session = connection.createSession(Boolean.TRUE,
    Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息发送给谁.
  // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
  Destination destination = session.createQueue("my-queue");
  // 消费者，消息接收者
  MessageConsumer consumer = session.createConsumer(destination);
  while (true) {
   TextMessage message = (TextMessage) consumer.receive(1000);
   if (null != message)
    System.out.println("收到消息：" + message.getText());
   else
    break;
  }
  session.close();
  connection.close();
 }
}

评论

1 楼 marswon 2013-05-17  

您好，JMS消息，异步消费，实际应用场景是用户登录上来创建消费者，然后用户下线的时候，销毁消费者，因而是不同的方法，如何去实现？而不是异步收一次就关闭了？