jms.txt

利用ActiveMQ中间件发送JMS消息的例子

在这里利用MQ作为消息中间件来发送点对点(PTP)的文本消息，以便对JMS有入门级的认识，需要下载activemq并进入${active-mq}\bin目录运行activemq.bat。需要activemq-all对应版本的jar和jms.jar。
消息发送方QSender通过键盘输入文本信息来发送消息，接收方QReceiver实现MessageListener接口异步接受消息并打印。代码如下

发送方

import java.io.BufferedReader;
import java.io.InputStreamReader;

import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;


public class QSender
{
public static void main(String[] args)
{
   new QSender().send();
}

public void send()
{
   BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
   try
   {
    //init connection factory with activemq
    QueueConnectionFactory factory=new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    //specify the destination
    Queue queue=new ActiveMQQueue("queue.name.sample");
    //create connection,session,producer and deliver message
    QueueConnection conn=factory.createQueueConnection();
    QueueSession session=conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    QueueSender sender=session.createSender(queue);
    String msgText=null;
    while(true)
    {
     System.out.println("enter message to send or 'quit'");
     msgText=br.readLine();
     if("quit".equals(msgText))
      break;
     TextMessage msg=session.createTextMessage(msgText);
     sender.send(msg);
    }
    System.out.println("exiting..");
    //close opened resources 
    br.close();
    conn.close();
   }
   catch(Exception e)
   {
    e.printStackTrace();
    System.exit(1);
   }
}
}


接收方

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;

// implements MessageListener in order to receiving asynchronous message 
public class QReceiver implements MessageListener
{
//indicate whether stoping receiving 
private boolean stop=false;
public static void main(String[] args)
{
   new QReceiver().receive();
}

public void receive()
{
   try
   {
    //the same as QSender
    QueueConnection conn=new ActiveMQConnectionFactory("tcp://127.0.0.1:61616").createQueueConnection();
    Queue queue=new ActiveMQQueue("queue.name.sample");
    QueueSession session=conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    QueueReceiver receiver=session.createReceiver(queue);
    //Sets the message consumer's MessageListener.
    receiver.setMessageListener(this);
    //ready to setup consumer,begin to receive message
    conn.start();
    while(!stop)
    {
     Thread.sleep(1000);
    }
    System.out.println("exiting..");
    conn.close();
   }
   catch(Exception e)
   {
    e.printStackTrace();
    System.exit(1);
   }
}
//called while receiving message
@Override
public void onMessage(Message msg)
{
   try
   {
    String text=((TextMessage)msg).getText();
    System.out.println(text);
    if(text.equals("stop"))
     stop=true;
   }
   catch(JMSException e)
   {
    e.printStackTrace();
    stop=true;
   }
}
}