消息队列之ActiveMQ简单环境搭建

2022-11-09 10:51:26

准备:

　　　环境:win7,Eclipse,jdk1.8

　　　ActiveMQ版本:ActiveMQ 5.9.0 Release下载地址:http://activemq.apache.org/download-archives.html

开始:

　　1.下载完后进行解压,解压完找到其中的activemq-all-5.9.0.jar包.

　　2.Eclipse创建一个java项目,导入上面这个jar包

　　3.创建生产者和消费者类

　　生产者

package com.sinosoft.activemq;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

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;

public class JMSProducer {

	//默认连接用户名

	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;

	//默认连接密码

	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;

	//默认连接地址

	private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL;

	//发送的消息数量

	private static final int SENDNUM = 10;

	public static void main(String[] args) {

		//连接工厂

		ConnectionFactory connectionFactory;

		//连接

		Connection connection = null;

		//会话,接收或者发送消息的线程

		Session session;

		//消息的目的地

		Destination destination;

		//消息生产者

		MessageProducer messageProducer;

		//实例化连接工厂

		connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKERURL);

		try {

			//通过连接工厂获取连接

			connection = connectionFactory.createConnection();

			//启动连接

			connection.start();

			//创建session

			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

			//创建一个名称为Hello World!的消息队列

			destination = session.createQueue("Hello World");

			//创建消息生产者

			messageProducer = session.createProducer(destination);

			//发送消息

			sendMessage(session,messageProducer);

			session.commit();

		} catch (Exception e) {

			e.printStackTrace();

		} finally{

			if(connection != null){

				try {

					connection.close();

				} catch (JMSException e) {

					e.printStackTrace();

				}

			}

		}

	}

	/**

	 * 发送消息

	 * @param session

	 * @param messageProducer 消息生产者

	 * @throws Exception

	 */

	public static void sendMessage(Session session,MessageProducer messageProducer) throws Exception{

		for (int i = 0; i < JMSProducer.SENDNUM; i++) {

			//创建一条文本消息

			TextMessage message = session.createTextMessage("activemq 发送消息:" + i);

			System.err.println("发送消息:activemq 发送消息:" + i);

			//通过消息生产者发出消息

			messageProducer.send(message);

		}

	}

}

　　消费者

package com.sinosoft.activemq;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.MessageConsumer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JMSConsumer {

	//默认连接用户名

	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;

	//默认连接密码

	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;

	//默认连接地址

	private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL;

	public static void main(String[] args) {

		//连接工厂

		ConnectionFactory connectionFactory;

		//连接

		Connection connection = null;

		//会话,接收或者发送消息的线程

		Session session;

		//消息目的地

		Destination destination;

		//消息的消费者

		MessageConsumer messageConsumer;

		//实例化连接工厂

		connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKERURL);

		try {

			//通过工厂获取连接

			connection = connectionFactory.createConnection();

			//启动连接

			connection.start();

			//创建会话

			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			//创建一个连接Hello World!的消息队列

			destination = session.createQueue("Hello World");

			//创建消息的消费者

			messageConsumer = session.createConsumer(destination);

			while(true){

				TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);

				if (textMessage != null) {

					System.err.println("收到的消息:" + textMessage.getText());

				} else {

					break;

				}

			}

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

}