ActiveMq池

有两种连接方式

1.Spring  引用

<!-- 配置JMS连接工厂 -->
    <bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
        destroy-method="stop">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL">
                    <value>failover:(tcp://192.168.2.211:62617,tcp://192.168.2.211:62618,tcp://192.168.2.211:62619)?maxReconnectAttempts=10</value>
                </property>
                <property name="useAsyncSend">
                    <value>true</value>
                </property>
            </bean>
        </property>
    </bean>

2.配置文件

MqUserName=admin
MqPassword=admin
MqUrl=failover:(tcp://192.168.2.211:62617,tcp://192.168.2.211:62618,tcp://192.168.2.211:62619)
  

3.代码

import java.util.Properties;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;


public class ActiveMqUtils {
    
    private static PooledConnectionFactory factory;
    
    static{
            //方式一
//             Properties pps = new Properties();
//             String url="",name="",pwd="";
//            try {
//                pps.load(ActiveMqUtils.class.getClassLoader().getResourceAsStream("MqProperty.properties"));
//                url = pps.getProperty("MqUrl");
//                name = pps.getProperty("MqUserName");
//                pwd = pps.getProperty("MqPassword");
//                ActiveMQConnectionFactory   activeMQConnectionFactory = new ActiveMQConnectionFactory();  
//                activeMQConnectionFactory.setUserName(name);  
//                activeMQConnectionFactory.setPassword(pwd);  
//                activeMQConnectionFactory.setBrokerURL(url);  
//                   
//                factory = new PooledConnectionFactory( activeMQConnectionFactory);  
                   
                // session数  
//                int maximumActive = 5;
//                factory.setMaximumActiveSessionPerConnection(maximumActive);  
//                factory.setIdleTimeout(120);  
//                factory.setMaxConnections(5);  
//                factory.setBlockIfSessionPoolIsFull(true);  
//            } catch (Exception e) {
//                e.printStackTrace();
//            } 
            
          //方式二
            String[] springConfigFiles = {"mybatis-spring-config.xml","module-service-config.xml" };
            ApplicationContext ctx = new ClassPathXmlApplicationContext( springConfigFiles );
            factory = (PooledConnectionFactory) ctx.getBean( "connectionFactory" );
    }
    
    public static void doSend(String  key,String message){
        Connection connection =null;
        Session session;
        Destination destination;
        MessageProducer producer;
        try {
            connection =factory.createConnection();
            connection.start();
            session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
            destination = session.createQueue(key);
            producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            TextMessage msgObj = session .createTextMessage(message);
            producer.send(msgObj);
            session.commit();
        }catch (Exception e) {
            e.printStackTrace();
            }    
        finally {
            close(connection);
        }
    }
    
    public static String doReceive(String  key){
        String rs="";
        Connection connection =null;
        Session session;
        Destination destination;
        MessageConsumer consumer;
        try {
            connection =factory.createConnection();
            connection.start();
            session = connection.createSession(Boolean.FALSE,
                    Session.AUTO_ACKNOWLEDGE);
            destination = session.createQueue("sales");
            consumer = session.createConsumer(destination);
            TextMessage message = (TextMessage) consumer.receive(10);
            if (null != message) {
                rs = message.getText();
            } 
        }catch (Exception e) {
            e.printStackTrace();
        }    
        finally {
            close(connection);
        }
        return rs;
    }

       /** 
  * 关闭连接 
  */ 
 public static void close(Connection connection) {  
     try {  
         if (connection != null) {  
             connection.close();  
         }  
     } catch (JMSException e) {  

         e.printStackTrace();  
     }  
 }  
    
    public static void main(String[]  agrs){
        
        for(int i= 0;i<20;i++){
            long begin = System.currentTimeMillis();
            ActiveMqUtils.doSend("1111", "sss");
            long end = System.currentTimeMillis();
            System.out.println("耗时："+(end-begin));
        }
        
    }
}