一、安装activemq
下载地址：https://archive.apache.org/dist/activemq/5.13.0/apache-activemq-5.13.0-bin.zip
下载完后如果是64位操作系统打开~\apache-activemq-5.13.0\bin\win64目录下的activemq.bat即可，如果是32位的打开~\apache-activemq-5.13.0\bin\win32目录下的activemq.bat就可以了。
打开activemq的管理地址http://localhost:8161/admin/，用户名和密码都是admin,出现以下界面说明activemq已经正常运行。
二、创建maven项目，pom.xml中添加activemq的jar包依赖：
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.13.0</version>
</dependency>
三、activemq连接池的实现
1、创建连接池接口ConnectionPool
package com.guods.mymq.activemq; import javax.jms.Connection;
import javax.jms.JMSException; /**
* jms连接池接口
* @author guods
*
*/
public interface ConnectionPool { Connection getConnection() throws JMSException;
void releaseConnection(Connection connection);
}
2、连接池实现ConnectionPoolImpl
创建初始连接，并把连接存到链表，使用的时候从链表中获取连接。
package com.guods.mymq.activemq; import java.util.LinkedList; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException; import org.apache.activemq.ActiveMQConnectionFactory; /**
* jms连接池实现
* @author guods
*
*/
public class ConnectionPoolImpl implements ConnectionPool { //连接用户名
private static final String USERNAME = "admin";
//连接密码
private static final String PASSWORD = "admin";
//连接地址
private static final String BROKEURL = "tcp://localhost:61616";
//初始连接数
private static final int INIT_SIZE = 2;
//最大连接数
private static final int MAX_SIZE = 5;
//当前存在的连接总数，包括：正在使用的连接 + connections列表中的连接
private int poolSize = 0; //创建连接工厂
private ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
//存放空闲connection的链表
private LinkedList<Connection> connections = new LinkedList<Connection>(); public ConnectionPoolImpl() {
initConnections();
}
/**
* 初始化连接，生成初始连接数
*/
private void initConnections() {
for (int i = 0; i < INIT_SIZE; i++) {
try {
connections.add(createConnection());
} catch (JMSException e) {
e.printStackTrace();
}
}
} /**
* 添加连接，每增加一个connection，poolSize加1
* @throws JMSException
*/
private synchronized Connection createConnection() throws JMSException {
Connection newConnection = connectionFactory.createConnection();
newConnection.start();
poolSize++;
return newConnection;
}
/**
* 删除连接，每删除一个connection，poolSize减1
* @param connection
* @throws JMSException
*/
private synchronized void delConnection() throws JMSException {
Connection connection = connections.removeFirst();
if (connection != null) {
connection.close();
poolSize--;
}
}
/**
* 从连接池获取connection
* @throws JMSException
*/
public synchronized Connection getConnection() throws JMSException {
int i = 0;
while (true) {
//3次获取连接失败，抛超时异常
if (i == 3) {
throw new JMSException("Get connection timeout !");
}
//连接池有连接，直接取一个返回
if (connections.size() > 0) {
return connections.removeFirst();
}
//连接池空，如果连接数没到最大，创建一个连接返回
if (poolSize < MAX_SIZE) {
try {
return createConnection();
} catch (JMSException e) {
}
}
//连接池空，并且连接达到最大数，等1秒再试尝试获取连接
try {
Thread.sleep(1000);
i++;
} catch (InterruptedException e) {
}
}
}
/**
* 释放连接，连接用完后放回队列
*/
public synchronized void releaseConnection(Connection connection) {
connections.add(connection);
} }
四、创建jms模板
beginSession方法从连接池获取连接，创建session。
endSession方法提交session，并把连接交还给连接池。
execute方法整合了beginSession和endSession两个模板方法，中间执行抽象方法action中的内容。
action在子类中实现业务操作，如创建队列、生产消息、消费消息等。
package com.guods.mymq.activemq; import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Session; /**
* 模板模式，创建JMS模板
* @author guods
*
*/
public abstract class JmsTemplate { private ConnectionPool connectionPool;
private Connection connection;
private Session session; public JmsTemplate(ConnectionPool connectionPool) {
super();
this.connectionPool = connectionPool;
} public abstract Object action(Session session) throws JMSException; public Object execute() throws JMSException{
//获取连接，开启session
beginSession();
//处理事务
Object object = action(session);
//关闭session，释放连接
endSession();
return object;
} /**
* 从连接池获取connection，创建session
* @throws JMSException
*/
private void beginSession() throws JMSException{
connection = connectionPool.getConnection();
session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
} /**
* 提交事务，关闭session，释放connection
* @throws JMSException
*/
private void endSession() throws JMSException{
session.commit();
session.close();
connectionPool.releaseConnection(connection);
}
}
五、实现activemq的各种业务操作
package com.guods.mymq.activemq; import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage; /**
* activemq的各种业务操作
* @author guods
*
*/
public class JmsAction { private ConnectionPool connectionPool; public JmsAction(ConnectionPool connectionPool) {
super();
this.connectionPool = connectionPool;
} /**
* 创建消息队列
* @param queueName
* @return
* @throws JMSException
*/
public Queue createQueue(final String queueName) throws JMSException { return (Queue) new JmsTemplate(connectionPool) { @Override
public Object action(Session session) {
try {
return session.createQueue(queueName);
} catch (JMSException e) {
return null;
}
}
}.execute();
}; /**
* 发送p2p消息
* @param queue
* @param message
* @throws JMSException
*/
public void produceP2pMessage(final Queue queue, final String message) throws JMSException{
new JmsTemplate(connectionPool) { @Override
public Object action(Session session) throws JMSException {
MessageProducer messageProducer = session.createProducer(queue);
TextMessage textMessage = session.createTextMessage(message);
messageProducer.send(queue, textMessage);
return null;
}
}.execute();
} /**
* 消费一条p2p消息
* @param queue
* @return
* @throws JMSException
*/
public String consumP2pMessage(final Queue queue) throws JMSException{
return (String) new JmsTemplate(connectionPool) { @Override
public Object action(Session session) throws JMSException {
MessageConsumer createConsumer = session.createConsumer(queue);
TextMessage textMessage = (TextMessage) createConsumer.receive(10000);
if (textMessage == null) {
return null;
}
return textMessage.getText();
}
}.execute();
} }
六、测试
1、生产者生产消息
package com.guods.mymq.activemq; import javax.jms.JMSException;
import javax.jms.Queue; public class JmsProducer { private static ConnectionPool connectionPool = new ConnectionPoolImpl();
private static JmsAction jmsAction = new JmsAction(connectionPool); public static void main(String[] args) throws JMSException {
//创建队列
Queue queue = jmsAction.createQueue("MyQueue");
//生产消息
for (int i = 0; i < 5; i++) {
String message = "第" + i + "条消息！";
jmsAction.produceP2pMessage(queue, message);
System.out.println("生产消息：" + message);
}
}
}
2、消费者消费消息
package com.guods.mymq.activemq; import javax.jms.JMSException;
import javax.jms.Queue; public class JmsConsumer { private static ConnectionPool connectionPool = new ConnectionPoolImpl();
private static JmsAction jmsAction = new JmsAction(connectionPool); public static void main(String[] args) throws JMSException {
//创建队列
Queue queue = jmsAction.createQueue("MyQueue");
//消费消息
while (true) {
String message = jmsAction.consumP2pMessage(queue);
if (message != null) {
System.out.println("消费消息：" + message);
}
}
}
}
git源码：https://github.com/dongsheng824/mymq.git