编码实现MQ连接池实现JMS消息发送连接管理

今天来说一下在使用到MQ时如果使用MQ的链接池。之前我也是没有注意到MQ也是有连接池的，后来因为系统之前实现每次创建和关闭链接消耗资源、宕机频繁，所以领导要求解决我才接触到。 我在网上看到的关于JMS的讲解还挺多，但是对于MQ连接池的讲解时大家都是讲如何在spring中配置连接池。首先采用spring配置后原系统加密配置的密码就成明文了，另外如果要改成spring发送那改动就大了。如果在不使用spring，不大改动代码的情况下完成采用MQ连接池来获得链接呢？ 现在哥已经养成了好的习惯，每次发博客都会将源码和工程直接发布的，不会让你看了半天连个例子也没有的。 首先要创建工程加入JAR包，要加的包都是工程内，哥不多说了，这些包是必须的。至于activemq-pool-5.2.0.jar，你一定要找到加进去，你懂的。 你还要到官方下载一个MQ服务器，我使用的是5.2，下载后运行起来以备使用。 第二步我们要实现如何获得链接的。这里我做了两个实现，用于更直观的了解MQ连接池。 其中连接池工厂的活跃数我们设置为1，方便测试。这个活跃数实际上是会话的数量，而不是数据库连接池一样的链接数量。 Java代码 package com.mq.service; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.pool.PooledConnectionFactory; /** * 链接工厂管理类 * 自己工厂定义成了单例模式，连接池是静态块进行初始化，具体实现自己看着办 */ public class MQPooledConnectionFactory { private static ActiveMQConnectionFactory connectionFactory; /** * 获得自己创建的链接工厂，这个工厂只初始化一次 */ public static ActiveMQConnectionFactory getMyActiveMQConnectionFactory() { if (null == connectionFactory) { connectionFactory = new ActiveMQConnectionFactory("system","manage", "tcp://127.0.0.1:61616"); } return connectionFactory; } private static PooledConnectionFactory pooledConnectionFactory; static { try { // 需要创建一个链接工厂然后设置到连接池中 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(); activeMQConnectionFactory.setUserName("system"); activeMQConnectionFactory.setPassword("manage"); activeMQConnectionFactory.setBrokerURL("tcp://127.0.0.1:61616"); // 如果将消息工厂作为属性设置则会有类型不匹配的错误，虽然Spring配置文件中是这么配置的，这里必须在初始化的时候设置进去 pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory); // 链接最大活跃数，为了在测试中区别我们使用的到底是不是一个对象和看是否能控制连接数(实际上是会话数)，我们在这里设置为1 int maximumActive = 1; pooledConnectionFactory.setMaximumActive(maximumActive); } catch (Exception e) { e.printStackTrace(); } } /** * 获得链接池工厂 */ public static PooledConnectionFactory getPooledConnectionFactory() { return pooledConnectionFactory; } /** * 对象回收销毁时停止链接 */ @Override protected void finalize() throws Throwable { pooledConnectionFactory.stop(); super.finalize(); } } package com.mq.service; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.pool.PooledConnectionFactory; /** * 链接工厂管理类 * 自己工厂定义成了单例模式，连接池是静态块进行初始化，具体实现自己看着办 */ public class MQPooledConnectionFactory { private static ActiveMQConnectionFactory connectionFactory; /** * 获得自己创建的链接工厂，这个工厂只初始化一次 */ public static ActiveMQConnectionFactory getMyActiveMQConnectionFactory() { if (null == connectionFactory) { connectionFactory = new ActiveMQConnectionFactory("system","manage", "tcp://127.0.0.1:61616"); } return connectionFactory; } private static PooledConnectionFactory pooledConnectionFactory; static { try { // 需要创建一个链接工厂然后设置到连接池中 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(); activeMQConnectionFactory.setUserName("system"); activeMQConnectionFactory.setPassword("manage"); activeMQConnectionFactory.setBrokerURL("tcp://127.0.0.1:61616"); // 如果将消息工厂作为属性设置则会有类型不匹配的错误，虽然Spring配置文件中是这么配置的，这里必须在初始化的时候设置进去 pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory); // 链接最大活跃数，为了在测试中区别我们使用的到底是不是一个对象和看是否能控制连接数(实际上是会话数)，我们在这里设置为1 int maximumActive = 1; pooledConnectionFactory.setMaximumActive(maximumActive); } catch (Exception e) { e.printStackTrace(); } } /** * 获得链接池工厂 */ public static PooledConnectionFactory getPooledConnectionFactory() { return pooledConnectionFactory; } /** * 对象回收销毁时停止链接 */ @Override protected void finalize() throws Throwable { pooledConnectionFactory.stop(); super.finalize(); } } 这个类会提供连接池工厂和MQ链接工厂的创建和获得方式。特别要注意到的是，在初始化连接池时要设置一个链接工厂进去，而且只能在初始化时作为参数传递进去，如果你看过spring的配置你会以为要做为属性传递，而且他确实有这个属性。 然后我们创建发送消息的基础类，在该类中得到链接然后创建消息进行发送，关闭会话等一些列操作。为了防止变量干扰，这些操作是在一个方法里面完成的。 Java代码 package com.mq.service; import java.util.Map; import java.util.Set; import javax.jms.Connection; import javax.jms.MapMessage; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; /** * 数据发送类，用于发送数据 * 如果获得链接，请查看被注释的代码 */ public class MQ_Service { // 发送消息 @SuppressWarnings("static-access") public void send(Map<String, String> map) throws Exception { ActiveMQConnectionFactory connectionFactory = null; Connection connection = null; Session session = null; Queue queue = null; MessageProducer producer = null; MapMessage messagep = null; // ==================================== try { // 获得我们自己初始化的链接工厂然后创建链接 connectionFactory = MQPooledConnectionFactory.getMyActiveMQConnectionFactory(); connection = connectionFactory.createConnection(); // 链接直接从链接池工厂进行获得 //connection = MQPooledConnectionFactory.getPooledConnectionFactory().createConnection(); session = connection.createSession(false, session.AUTO_ACKNOWLEDGE); queue = session.createQueue("CUI_JMS"); producer = session.createProducer(queue); // 链接开始，如果我们使用的是连接池，那么即使你不开始，也是没有问题的 connection.start(); } catch (Exception e) { e.printStackTrace(); } // ==================================== messagep = session.createMapMessage(); Set<String> keySet = map.keySet(); for (String key : keySet) { String value = map.get(key); messagep.setBytes(key, value.getBytes("UTF-8")); System.out.println(key + "-->" + value); } producer.send(messagep); messagep.clearBody(); messagep.clearProperties(); // =================================== // 通过打印会话的内存地址和链接的客户端编号就可以知道我们使用的是不是同一个会话和链接 System.out.println(session.toString()); System.out.println(connection.getClientID()); // 无论使用的自己的工厂还是连接池的，都要将会话关闭 // 如果不关闭，在使用连接池的时可以看到效果，发送两次时只能发送一次，造成堵塞 session.close();