activemq

1activeMq是apache的一个子项目，支持jms1.1和j2ee1.4，主要实现了一个队列，用于网络间消息传输

 

2下载apache actionmq  

  http://www.apache.org/dyn/closer.cgi?path=%2Factivemq%2Fapache-activemq%2F5.5.0%2Fapache-activemq-5.5.0-bin.zip

 

3下载后的目录

 

 activemq-all-5.5.1.jar

 bin

 conf

 data

 docs

 example

 lib

 LICENSE

 NOTICE

 README.txt

 user-guide.html

 webapps

 WebConsole-README.txt

 

启动 mq,进入 bin目录，运行activemq.bat 文件，可以看到绑定的端口和侦听的端口，启动后访问地址

 

http://localhost:8161/

 

可以查看一些例子和查看控制台。这个时候，MQ的队列服务功能已经开启了。接下来需要编码实现

 

package com.mchz.mq.home;

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://localhost:61616";;
	private static final String QUEUE_NAME = "send.queue";
	protected String expectedBody = "hello,word,172.16.4.106,hello,word,"
			+ "172.16.4.106,hello,word,172.16.4.106,hello,word,172.16.4.106,"
			+ "hello,word,172.16.4.106,hello,word,172.16.4.106,hello,word,"
			+ "172.16.4.106,hello,word,172.16.4.106,hello,word,172.16.4.106,"
			+ "hello,word,172.16.4.106,hello,word,172.16.4.106,hello,word,"
			+ "172.16.4.106,hello,word,172.16.4.106,hello,word,172.16.4.106,"
			+ "hello,word,172.16.4.106,hello,word,172.16.4.106,hello,word,"
			+ "172.16.4.106,hello,word,172.16.4.106,hello,word,172.16.4.106,"
			+ "hello,word,172.16.4.106,hello,word,172.16.4.106,";

	public void sendMessage() throws JMSException {

		Connection connection = null;
		Session session = null;
		try {
			ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
					url);
			connection = connectionFactory.createConnection();

			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(QUEUE_NAME);
			MessageProducer producer = session.createProducer(destination);
			TextMessage message = session.createTextMessage(expectedBody);
			message.setStringProperty("ipaddress", "172.16.4.106");
			message.setStringProperty("address", "zj.hz.mc.219");
			producer.send(message);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			session.close();
			connection.close();
		}
	}

	public static void main(String[] args) {

		SendMessage send = new SendMessage();
		long startTime=System.currentTimeMillis();
				
		try {
			for (int i = 0; i < 100000; i++) {
				send.sendMessage();
//				System.out.println(i);
			}
		} catch (JMSException e) {
			e.printStackTrace();
			long endTime=System.currentTimeMillis();
			System.out.println(endTime-startTime);
		}
		long endTime=System.currentTimeMillis();
		System.out.println(endTime-startTime);
	}

}

 

 

 

package com.mchz.mq.home;

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

import org.apache.activemq.ActiveMQConnectionFactory;

public class ReceiveMessage {
	private static final String url = "tcp://localhost:61616";
	private static final String QUEUE_NAME = "send.queue";

	public void receiveMessage() {
		Connection connection = null;
		try {
			try {
				ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
						url);
				connection = connectionFactory.createConnection();
			} catch (Exception e) {
				e.printStackTrace();
			}
			connection.start();
			Session session = connection.createSession(false,
					Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(QUEUE_NAME);
			MessageConsumer consumer = session.createConsumer(destination);
			consumeMessagesAndClose(connection, session, consumer);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	protected void consumeMessagesAndClose(Connection connection,
			Session session, MessageConsumer consumer) throws JMSException {
		for (int i = 0; i < 16349;i++) {
			Message message = consumer.receive(1000);
			if (message != null) {
				onMessage(message);
			}
		}
		System.out.println("Closing connection");
		consumer.close();
		session.close();
		connection.close();
	}

	public void onMessage(Message message) {
		System.out.println("come in onMessage.....");

		try {
			if (message instanceof TextMessage) {
				TextMessage txtMsg = (TextMessage) message;
				String msg = txtMsg.getText();
				System.out.println("Received: " + msg);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

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

 

 

一个用与发送消息，一个用于接收消息，先运行发送消息的类SendMessage ，然后查看MQ的控制 http://localhost:8161/admin/queues.jsp

可以看到自己往队列里面放的消息，接下来运行接收的类ReceiveMessage ，在刷新控制台，可以看到消息被接收。

 

另外，MQ还可以把一个队列里面的消息发送到另一个队列。