import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Bean(name="connectionFactory")
	public ConnectionFactory connectionFactory(){
		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
//		activeMQConnectionFactory.setUseAsyncSend(true);
		activeMQConnectionFactory.setUserName(jmsUserName);
		activeMQConnectionFactory.setPassword(jmsPassword);
		activeMQConnectionFactory.setBrokerURL(jmsBrokerUrl);

		// 默认重复投递6次将转发到死信队列,改为无限次数
//		RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
//		redeliveryPolicy.setMaximumRedeliveries(-1);
//		
//		activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
		
		PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
		pooledConnectionFactory.setMaxConnections(jmsMaxPooledConnections);
//		CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(pooledConnectionFactory);
//		cachingConnectionFactory.setSessionCacheSize(10);
		return pooledConnectionFactory;
	}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void customPooledConnectionFactoryConfiguration() {
	load(EmptyConfiguration.class, "spring.activemq.pool.enabled:true",
			"spring.activemq.pool.maxConnections:256",
			"spring.activemq.pool.idleTimeout:512",
			"spring.activemq.pool.expiryTimeout:4096",
			"spring.activemq.pool.configuration.maximumActiveSessionPerConnection:1024",
			"spring.activemq.pool.configuration.timeBetweenExpirationCheckMillis:2048");
	ConnectionFactory connectionFactory = this.context
			.getBean(ConnectionFactory.class);
	assertThat(connectionFactory).isInstanceOf(PooledConnectionFactory.class);
	PooledConnectionFactory pooledConnectionFactory = (PooledConnectionFactory) connectionFactory;
	assertThat(pooledConnectionFactory.getMaxConnections()).isEqualTo(256);
	assertThat(pooledConnectionFactory.getIdleTimeout()).isEqualTo(512);
	assertThat(pooledConnectionFactory.getMaximumActiveSessionPerConnection())
			.isEqualTo(1024);
	assertThat(pooledConnectionFactory.getTimeBetweenExpirationCheckMillis())
			.isEqualTo(2048);
	assertThat(pooledConnectionFactory.getExpiryTimeout()).isEqualTo(4096);
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static ConnectionFactory createConnectionFactory(String options) {
    // using a unique broker name improves testing when running the entire test suite in the same JVM
    int id = counter.incrementAndGet();
    String url = "vm://test-broker-" + id + "?broker.persistent=false&broker.useJmx=false";
    if (options != null) {
        url = url + "&" + options;
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    // optimize AMQ to be as fast as possible so unit testing is quicker
    connectionFactory.setCopyMessageOnSend(false);
    connectionFactory.setOptimizeAcknowledge(true);
    connectionFactory.setOptimizedMessageDispatch(true);
    connectionFactory.setTrustAllPackages(true);

    // When using asyncSend, producers will not be guaranteed to send in the order we 
    // have in the tests (which may be confusing for queues) so we need this set to false.
    // Another way of guaranteeing order is to use persistent messages or transactions.
    connectionFactory.setUseAsyncSend(false);

    connectionFactory.setAlwaysSessionAsync(false);
    // use a pooled connection factory
    PooledConnectionFactory pooled = new PooledConnectionFactory(connectionFactory);
    pooled.setMaxConnections(8);
    return pooled;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static ConnectionFactory createConnectionFactory(String options) {
    // using a unique broker name improves testing when running the entire test suite in the same JVM
    int id = counter.incrementAndGet();
    String url = "vm://test-broker-" + id + "?broker.persistent=false&broker.useJmx=false";
    if (options != null) {
        url = url + "&" + options;
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    // optimize AMQ to be as fast as possible so unit testing is quicker
    connectionFactory.setCopyMessageOnSend(false);
    connectionFactory.setOptimizeAcknowledge(true);
    connectionFactory.setOptimizedMessageDispatch(true);

    // When using asyncSend, producers will not be guaranteed to send in the order we
    // have in the tests (which may be confusing for queues) so we need this set to false.
    // Another way of guaranteeing order is to use persistent messages or transactions.
    connectionFactory.setUseAsyncSend(false);

    connectionFactory.setAlwaysSessionAsync(false);
    // use a pooled connection factory
    PooledConnectionFactory pooled = new PooledConnectionFactory(connectionFactory);
    pooled.setMaxConnections(8);
    return pooled;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void testMultiActiveMQ() throws Exception {
    CountDownLatch latch = new CountDownLatch(latchCount());
    LOG.info("Latch count: " + latch.getCount());
    ExecutorService executorService = Executors.newFixedThreadPool(2);
    IntStream.range(0, (int)latch.getCount())
        .forEach(i ->
            executorService.execute(new JmsAtnaMessageConsumer(latch, JMS_BROKER_URL, JMS_QUEUE_NAME))
        );

    ConnectionFactory jmsConnectionFactory = new PooledConnectionFactory(JMS_BROKER_URL);
    ActiveMQQueue jmsQueue = new ActiveMQQueue(JMS_QUEUE_NAME);
    atnaQueue = new JmsAuditMessageQueue(jmsConnectionFactory, jmsQueue, false);
    AuditorModuleContext.getContext().setQueue(atnaQueue);

    IntStream.range(0, (int)latch.getCount())
        .forEach(i ->
            IHEAuditor.getAuditor().auditActorStartEvent(SUCCESS, "actorName" + i, "actorStarter" + i)
        );

    latch.await();
    assertEquals(0, latch.getCount());
    LOG.info("Latch count: " + latch.getCount());
    LOG.info("Shutting down Executor service");
    executorService.shutdown();
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public JMSMsgSrc(String url, Map<String, Object> options) throws JMSException {
	super(options);
	String destName = getOption("destName", "hmp-tmp");
	
	// create/configure factory + pool
	fact = new ActiveMQConnectionFactory(url);
	fact.setUseAsyncSend(true);
	pool = new org.apache.activemq.pool.PooledConnectionFactory(fact);
	pool.setMaxConnections(10);
	pool.setIdleTimeout(0);
	
	// create/configure connection + session
	conn = pool.createConnection();
	conn.start();
	sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
	
	// create/configure destination + producer
	dest = sess.createQueue(destName);
	producer = sess.createProducer(dest);
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static ActiveMQComponent newAMQInstance(){
	 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
	    connectionFactory.setBrokerURL("vm:localhost");
	    // use a pooled connection factory between the module and the queue
	    PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);

	    // how many connections should there be in the session pool?
	    pooledConnectionFactory.setMaxConnections(100);
	    pooledConnectionFactory.setMaximumActiveSessionPerConnection(100);
	    pooledConnectionFactory.setCreateConnectionOnStartup(true);
	    pooledConnectionFactory.setBlockIfSessionPoolIsFull(false);

	    JmsConfiguration jmsConfiguration = new JmsConfiguration(pooledConnectionFactory);
	    jmsConfiguration.setDeliveryPersistent(false);
	    jmsConfiguration.setTimeToLive(1000*10);
	    ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent("vm:localhost");
	    return activeMQComponent;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
private boolean tryToEstablishJMSConnectionImpl() {
	destroy();

	Set<String> jmsBrokerURISet = getJmsBrokerURISet();
	if (BooleanUtils.isNotTrue(isEnabledOAuthAuditnLogging()) || CollectionUtils.isEmpty(jmsBrokerURISet))
		return false;

	this.jmsBrokerURISet = new HashSet<String>(jmsBrokerURISet);
	this.jmsUserName = getJmsUserName();
	this.jmsPassword = getJmsPassword();

	Iterator<String> jmsBrokerURIIterator = jmsBrokerURISet.iterator();

	StringBuilder uriBuilder = new StringBuilder();
	while (jmsBrokerURIIterator.hasNext()) {
		String jmsBrokerURI = jmsBrokerURIIterator.next();
		uriBuilder.append("tcp://");
		uriBuilder.append(jmsBrokerURI);
		if (jmsBrokerURIIterator.hasNext())
			uriBuilder.append(",");
	}

	String brokerUrl = BROKER_URL_PREFIX + uriBuilder + BROKER_URL_SUFFIX;

	ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.jmsUserName, this.jmsPassword,
			brokerUrl);
	this.pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);

	pooledConnectionFactory.setIdleTimeout(5000);
	pooledConnectionFactory.setMaxConnections(10);
	pooledConnectionFactory.start();

	return true;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
/**
 * Inits the.
 *
 * @throws JMSException the jMS exception
 */
public void init() throws JMSException {
    Properties properties = new Properties();
    try {
        properties.load(this.getClass().getResourceAsStream("/sink.properties"));
    } catch (IOException e) {
        logger.error("got: " + e.getMessage());
    }

    user = properties.getProperty("amq.user");
    password = System.getenv("KLOOPZ_AMQ_PASS");


    if (password == null) {
        throw new JMSException("missing KLOOPZ_AMQ_PASS env var");
    }

    AMQConnectorURI connectStringGenerator = new AMQConnectorURI();
    connectStringGenerator.setHost("opsmq");
    connectStringGenerator.setProtocol("tcp");
    connectStringGenerator.setPort(61616);
    connectStringGenerator.setTransport("failover");
    connectStringGenerator.setDnsResolve(true);
    connectStringGenerator.setKeepAlive(true);
    HashMap<String, String> transportOptions = new HashMap<>();
    transportOptions.put("initialReconnectDelay", "1000");
    transportOptions.put("startupMaxReconnectAttempts", mqConnectionStartupRetries);
    transportOptions.put("timeout", mqConnectionTimeout);
    transportOptions.put("useExponentialBackOff", "false");
    connectStringGenerator.setTransportOptions(transportOptions);
    url = connectStringGenerator.build();

    showParameters();

    // Create the connection.
    ActiveMQConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory(user, password, url);
    amqConnectionFactory.setUseAsyncSend(true);
    PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(amqConnectionFactory);
    pooledConnectionFactory.setMaxConnections(amqConnectionPoolSize);
    pooledConnectionFactory.setIdleTimeout(10000);

    for (int i = 0; i < poolsize; i++) {
        JmsTemplate producerTemplate = new JmsTemplate(pooledConnectionFactory);
        producerTemplate.setSessionTransacted(false);
        int shard = i + 1;
        Destination perfin = new org.apache.activemq.command.ActiveMQQueue(queueBase + "-" + shard);
        producerTemplate.setDefaultDestination(perfin);
        producerTemplate.setDeliveryPersistent(false);
        producers[i] = producerTemplate;
    }


}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
/**
 * Close connection.
 */
public void closeConnection() {
    for (JmsTemplate jt : producers) {
        ((PooledConnectionFactory) jt.getConnectionFactory()).stop();
    }
    producers = null;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Bean(destroyMethod = "stop")
@ConditionalOnProperty(prefix = "spring.activemq.pool", name = "enabled", havingValue = "true", matchIfMissing = false)
@ConfigurationProperties("spring.activemq.pool.configuration")
public PooledConnectionFactory pooledJmsConnectionFactory(
		ActiveMQProperties properties) {
	PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(
			new ActiveMQConnectionFactoryFactory(properties)
					.createConnectionFactory(ActiveMQConnectionFactory.class));

	ActiveMQProperties.Pool pool = properties.getPool();
	pooledConnectionFactory.setMaxConnections(pool.getMaxConnections());
	pooledConnectionFactory.setIdleTimeout(pool.getIdleTimeout());
	pooledConnectionFactory.setExpiryTimeout(pool.getExpiryTimeout());
	return pooledConnectionFactory;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void testActiveMQOverriddenPool() {
	load(TestConfiguration.class, "spring.activemq.pool.enabled:true");
	JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
	PooledConnectionFactory pool = this.context
			.getBean(PooledConnectionFactory.class);
	assertThat(jmsTemplate).isNotNull();
	assertThat(pool).isNotNull();
	assertThat(pool).isEqualTo(jmsTemplate.getConnectionFactory());
	ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) pool
			.getConnectionFactory();
	assertThat(factory.getBrokerURL()).isEqualTo(ACTIVEMQ_EMBEDDED_URL);
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void testActiveMQOverriddenPoolAndStandalone() {
	load(TestConfiguration.class, "spring.activemq.pool.enabled:true",
			"spring.activemq.inMemory:false");
	JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
	PooledConnectionFactory pool = this.context
			.getBean(PooledConnectionFactory.class);
	assertThat(jmsTemplate).isNotNull();
	assertThat(pool).isNotNull();
	assertThat(pool).isEqualTo(jmsTemplate.getConnectionFactory());
	ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) pool
			.getConnectionFactory();
	assertThat(factory.getBrokerURL()).isEqualTo(ACTIVEMQ_NETWORK_URL);
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void testActiveMQOverriddenPoolAndRemoteServer() {
	load(TestConfiguration.class, "spring.activemq.pool.enabled:true",
			"spring.activemq.brokerUrl:tcp://remote-host:10000");
	JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
	PooledConnectionFactory pool = this.context
			.getBean(PooledConnectionFactory.class);
	assertThat(jmsTemplate).isNotNull();
	assertThat(pool).isNotNull();
	assertThat(pool).isEqualTo(jmsTemplate.getConnectionFactory());
	ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) pool
			.getConnectionFactory();
	assertThat(factory.getBrokerURL()).isEqualTo("tcp://remote-host:10000");
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Test
public void pooledConnectionFactoryConfiguration() throws JMSException {
	load(EmptyConfiguration.class, "spring.activemq.pool.enabled:true");
	ConnectionFactory connectionFactory = this.context
			.getBean(ConnectionFactory.class);
	assertThat(connectionFactory).isInstanceOf(PooledConnectionFactory.class);
	this.context.close();
	assertThat(connectionFactory.createConnection()).isNull();
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
@Bean
public ConnectionFactory jmsConnectionFactory() {
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, brokerUrl);
    if (pooled) {
        PooledConnectionFactory pool = new PooledConnectionFactory();
        pool.setConnectionFactory(connectionFactory);
        return pool;
    }
    return connectionFactory;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static ConnectionFactory createPersistentConnectionFactory(String options) {
    // using a unique broker name improves testing when running the entire test suite in the same JVM
    int id = counter.incrementAndGet();

    // use an unique data directory in target
    String dir = "target/activemq-data-" + id;

    // remove dir so its empty on startup
    FileUtil.removeDir(new File(dir));

    String url = "vm://test-broker-" + id + "?broker.persistent=true&broker.useJmx=false&broker.dataDirectory=" + dir;
    if (options != null) {
        url = url + "&" + options;
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    // optimize AMQ to be as fast as possible so unit testing is quicker
    connectionFactory.setCopyMessageOnSend(false);
    connectionFactory.setOptimizeAcknowledge(true);
    connectionFactory.setOptimizedMessageDispatch(true);
    connectionFactory.setUseAsyncSend(true);
    connectionFactory.setAlwaysSessionAsync(false);
    connectionFactory.setTrustAllPackages(true);

    // use a pooled connection factory
    PooledConnectionFactory pooled = new PooledConnectionFactory(connectionFactory);
    pooled.setMaxConnections(8);
    return pooled;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static PooledConnectionFactory createPooledConnectionFactory() {
    ConnectionFactory cf = createConnectionFactory(null, null);
    PooledConnectionFactory pooled = new PooledConnectionFactory();
    pooled.setConnectionFactory(cf);
    pooled.setMaxConnections(8);
    return pooled;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
public static ConnectionFactory createPersistentConnectionFactory(String options) {
    // using a unique broker name improves testing when running the entire test suite in the same JVM
    int id = counter.incrementAndGet();

    // use an unique data directory in target
    String dir = "target/activemq-data-" + id;

    // remove dir so its empty on startup
    FileUtil.removeDir(new File(dir));

    String url = "vm://test-broker-" + id + "?broker.persistent=true&broker.useJmx=false&broker.dataDirectory=" + dir;
    if (options != null) {
        url = url + "&" + options;
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    // optimize AMQ to be as fast as possible so unit testing is quicker
    connectionFactory.setCopyMessageOnSend(false);
    connectionFactory.setOptimizeAcknowledge(true);
    connectionFactory.setOptimizedMessageDispatch(true);
    connectionFactory.setUseAsyncSend(true);
    connectionFactory.setAlwaysSessionAsync(false);

    // use a pooled connection factory
    PooledConnectionFactory pooled = new PooledConnectionFactory(connectionFactory);
    pooled.setMaxConnections(8);
    return pooled;
}
 

import org.apache.activemq.pool.PooledConnectionFactory; //导入依赖的package包/类
/**
 * Creates the JMS connection factory without registering it.
 * 
 * @param repo  the component repository, only used to register secondary items like lifecycle, not null
 * @return the JMS connection factory, not null
 */
protected ConnectionFactory createConnectionFactory(ComponentRepository repo) {
  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(getClientBrokerUri());
  connectionFactory.setWatchTopicAdvisories(false);
  PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);
  pooledConnectionFactory.setIdleTimeout(0);
  return pooledConnectionFactory;
}