001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.jms.pool; 018 019import java.util.Properties; 020import java.util.concurrent.atomic.AtomicBoolean; 021import java.util.concurrent.atomic.AtomicReference; 022 023import javax.jms.Connection; 024import javax.jms.ConnectionFactory; 025import javax.jms.JMSException; 026import javax.jms.QueueConnection; 027import javax.jms.QueueConnectionFactory; 028import javax.jms.TopicConnection; 029import javax.jms.TopicConnectionFactory; 030 031import org.apache.commons.pool.KeyedPoolableObjectFactory; 032import org.apache.commons.pool.impl.GenericKeyedObjectPool; 033import org.slf4j.Logger; 034import org.slf4j.LoggerFactory; 035 036/** 037 * A JMS provider which pools Connection, Session and MessageProducer instances 038 * so it can be used with tools like <a href="http://camel.apache.org/activemq.html">Camel</a> and Spring's 039 * <a href="http://activemq.apache.org/spring-support.html">JmsTemplate and MessagListenerContainer</a>. 040 * Connections, sessions and producers are returned to a pool after use so that they can be reused later 041 * without having to undergo the cost of creating them again. 042 * 043 * b>NOTE:</b> while this implementation does allow the creation of a collection of active consumers, 044 * it does not 'pool' consumers. Pooling makes sense for connections, sessions and producers, which 045 * are expensive to create and can remain idle a minimal cost. Consumers, on the other hand, are usually 046 * just created at startup and left active, handling incoming messages as they come. When a consumer is 047 * complete, it is best to close it rather than return it to a pool for later reuse: this is because, 048 * even if a consumer is idle, ActiveMQ will keep delivering messages to the consumer's prefetch buffer, 049 * where they'll get held until the consumer is active again. 050 * 051 * If you are creating a collection of consumers (for example, for multi-threaded message consumption), you 052 * might want to consider using a lower prefetch value for each consumer (e.g. 10 or 20), to ensure that 053 * all messages don't end up going to just one of the consumers. See this FAQ entry for more detail: 054 * http://activemq.apache.org/i-do-not-receive-messages-in-my-second-consumer.html 055 * 056 * Optionally, one may configure the pool to examine and possibly evict objects as they sit idle in the 057 * pool. This is performed by an "idle object eviction" thread, which runs asynchronously. Caution should 058 * be used when configuring this optional feature. Eviction runs contend with client threads for access 059 * to objects in the pool, so if they run too frequently performance issues may result. The idle object 060 * eviction thread may be configured using the {@link org.apache.activemq.jms.pool.PooledConnectionFactory#setTimeBetweenExpirationCheckMillis} method. By 061 * default the value is -1 which means no eviction thread will be run. Set to a non-negative value to 062 * configure the idle eviction thread to run. 063 * 064 */ 065public class PooledConnectionFactory implements ConnectionFactory, QueueConnectionFactory, TopicConnectionFactory { 066 private static final transient Logger LOG = LoggerFactory.getLogger(PooledConnectionFactory.class); 067 068 protected final AtomicBoolean stopped = new AtomicBoolean(false); 069 private GenericKeyedObjectPool<ConnectionKey, ConnectionPool> connectionsPool; 070 071 protected Object connectionFactory; 072 073 private int maximumActiveSessionPerConnection = 500; 074 private int idleTimeout = 30 * 1000; 075 private boolean blockIfSessionPoolIsFull = true; 076 private long blockIfSessionPoolIsFullTimeout = -1L; 077 private long expiryTimeout = 0l; 078 private boolean createConnectionOnStartup = true; 079 private boolean useAnonymousProducers = true; 080 private boolean reconnectOnException = true; 081 082 // Temporary value used to always fetch the result of makeObject. 083 private final AtomicReference<ConnectionPool> mostRecentlyCreated = new AtomicReference<ConnectionPool>(null); 084 085 public void initConnectionsPool() { 086 if (this.connectionsPool == null) { 087 this.connectionsPool = new GenericKeyedObjectPool<ConnectionKey, ConnectionPool>( 088 new KeyedPoolableObjectFactory<ConnectionKey, ConnectionPool>() { 089 090 @Override 091 public void activateObject(ConnectionKey key, ConnectionPool connection) throws Exception { 092 } 093 094 @Override 095 public void destroyObject(ConnectionKey key, ConnectionPool connection) throws Exception { 096 try { 097 if (LOG.isTraceEnabled()) { 098 LOG.trace("Destroying connection: {}", connection); 099 } 100 connection.close(); 101 } catch (Exception e) { 102 LOG.warn("Close connection failed for connection: " + connection + ". This exception will be ignored.",e); 103 } 104 } 105 106 @Override 107 public ConnectionPool makeObject(ConnectionKey key) throws Exception { 108 Connection delegate = createConnection(key); 109 110 ConnectionPool connection = createConnectionPool(delegate); 111 connection.setIdleTimeout(getIdleTimeout()); 112 connection.setExpiryTimeout(getExpiryTimeout()); 113 connection.setMaximumActiveSessionPerConnection(getMaximumActiveSessionPerConnection()); 114 connection.setBlockIfSessionPoolIsFull(isBlockIfSessionPoolIsFull()); 115 if (isBlockIfSessionPoolIsFull() && getBlockIfSessionPoolIsFullTimeout() > 0) { 116 connection.setBlockIfSessionPoolIsFullTimeout(getBlockIfSessionPoolIsFullTimeout()); 117 } 118 connection.setUseAnonymousProducers(isUseAnonymousProducers()); 119 connection.setReconnectOnException(isReconnectOnException()); 120 121 if (LOG.isTraceEnabled()) { 122 LOG.trace("Created new connection: {}", connection); 123 } 124 125 PooledConnectionFactory.this.mostRecentlyCreated.set(connection); 126 127 return connection; 128 } 129 130 @Override 131 public void passivateObject(ConnectionKey key, ConnectionPool connection) throws Exception { 132 } 133 134 @Override 135 public boolean validateObject(ConnectionKey key, ConnectionPool connection) { 136 if (connection != null && connection.expiredCheck()) { 137 if (LOG.isTraceEnabled()) { 138 LOG.trace("Connection has expired: {} and will be destroyed", connection); 139 } 140 141 return false; 142 } 143 144 return true; 145 } 146 }); 147 148 // Set max idle (not max active) since our connections always idle in the pool. 149 this.connectionsPool.setMaxIdle(1); 150 this.connectionsPool.setLifo(false); 151 152 // We always want our validate method to control when idle objects are evicted. 153 this.connectionsPool.setTestOnBorrow(true); 154 this.connectionsPool.setTestWhileIdle(true); 155 } 156 } 157 158 /** 159 * @return the currently configured ConnectionFactory used to create the pooled Connections. 160 */ 161 public Object getConnectionFactory() { 162 return connectionFactory; 163 } 164 165 /** 166 * Sets the ConnectionFactory used to create new pooled Connections. 167 * <p/> 168 * Updates to this value do not affect Connections that were previously created and placed 169 * into the pool. In order to allocate new Connections based off this new ConnectionFactory 170 * it is first necessary to {@link #clear} the pooled Connections. 171 * 172 * @param toUse 173 * The factory to use to create pooled Connections. 174 */ 175 public void setConnectionFactory(final Object toUse) { 176 if (toUse instanceof ConnectionFactory) { 177 this.connectionFactory = toUse; 178 } else { 179 throw new IllegalArgumentException("connectionFactory should implement javax.jmx.ConnectionFactory"); 180 } 181 } 182 183 @Override 184 public QueueConnection createQueueConnection() throws JMSException { 185 return (QueueConnection) createConnection(); 186 } 187 188 @Override 189 public QueueConnection createQueueConnection(String userName, String password) throws JMSException { 190 return (QueueConnection) createConnection(userName, password); 191 } 192 193 @Override 194 public TopicConnection createTopicConnection() throws JMSException { 195 return (TopicConnection) createConnection(); 196 } 197 198 @Override 199 public TopicConnection createTopicConnection(String userName, String password) throws JMSException { 200 return (TopicConnection) createConnection(userName, password); 201 } 202 203 @Override 204 public Connection createConnection() throws JMSException { 205 return createConnection(null, null); 206 } 207 208 @Override 209 public synchronized Connection createConnection(String userName, String password) throws JMSException { 210 if (stopped.get()) { 211 LOG.debug("PooledConnectionFactory is stopped, skip create new connection."); 212 return null; 213 } 214 215 ConnectionPool connection = null; 216 ConnectionKey key = new ConnectionKey(userName, password); 217 218 // This will either return an existing non-expired ConnectionPool or it 219 // will create a new one to meet the demand. 220 if (getConnectionsPool().getNumIdle(key) < getMaxConnections()) { 221 try { 222 connectionsPool.addObject(key); 223 connection = mostRecentlyCreated.getAndSet(null); 224 connection.incrementReferenceCount(); 225 } catch (Exception e) { 226 throw createJmsException("Error while attempting to add new Connection to the pool", e); 227 } 228 } else { 229 try { 230 // We can race against other threads returning the connection when there is an 231 // expiration or idle timeout. We keep pulling out ConnectionPool instances until 232 // we win and get a non-closed instance and then increment the reference count 233 // under lock to prevent another thread from triggering an expiration check and 234 // pulling the rug out from under us. 235 while (connection == null) { 236 connection = connectionsPool.borrowObject(key); 237 synchronized (connection) { 238 if (connection.getConnection() != null) { 239 connection.incrementReferenceCount(); 240 break; 241 } 242 243 // Return the bad one to the pool and let if get destroyed as normal. 244 connectionsPool.returnObject(key, connection); 245 connection = null; 246 } 247 } 248 } catch (Exception e) { 249 throw createJmsException("Error while attempting to retrieve a connection from the pool", e); 250 } 251 252 try { 253 connectionsPool.returnObject(key, connection); 254 } catch (Exception e) { 255 throw createJmsException("Error when returning connection to the pool", e); 256 } 257 } 258 259 return newPooledConnection(connection); 260 } 261 262 protected Connection newPooledConnection(ConnectionPool connection) { 263 return new PooledConnection(connection); 264 } 265 266 private JMSException createJmsException(String msg, Exception cause) { 267 JMSException exception = new JMSException(msg); 268 exception.setLinkedException(cause); 269 exception.initCause(cause); 270 return exception; 271 } 272 273 protected Connection createConnection(ConnectionKey key) throws JMSException { 274 if (connectionFactory instanceof ConnectionFactory) { 275 if (key.getUserName() == null && key.getPassword() == null) { 276 return ((ConnectionFactory) connectionFactory).createConnection(); 277 } else { 278 return ((ConnectionFactory) connectionFactory).createConnection(key.getUserName(), key.getPassword()); 279 } 280 } else { 281 throw new IllegalStateException("connectionFactory should implement javax.jms.ConnectionFactory"); 282 } 283 } 284 285 public void start() { 286 LOG.debug("Staring the PooledConnectionFactory: create on start = {}", isCreateConnectionOnStartup()); 287 stopped.set(false); 288 if (isCreateConnectionOnStartup()) { 289 try { 290 // warm the pool by creating a connection during startup 291 createConnection().close(); 292 } catch (JMSException e) { 293 LOG.warn("Create pooled connection during start failed. This exception will be ignored.", e); 294 } 295 } 296 } 297 298 public void stop() { 299 if (stopped.compareAndSet(false, true)) { 300 LOG.debug("Stopping the PooledConnectionFactory, number of connections in cache: {}", 301 connectionsPool != null ? connectionsPool.getNumActive() : 0); 302 try { 303 if (connectionsPool != null) { 304 connectionsPool.close(); 305 } 306 } catch (Exception e) { 307 } 308 } 309 } 310 311 /** 312 * Clears all connections from the pool. Each connection that is currently in the pool is 313 * closed and removed from the pool. A new connection will be created on the next call to 314 * {@link #createConnection}. Care should be taken when using this method as Connections that 315 * are in use be client's will be closed. 316 */ 317 public void clear() { 318 319 if (stopped.get()) { 320 return; 321 } 322 323 getConnectionsPool().clear(); 324 } 325 326 /** 327 * Returns the currently configured maximum number of sessions a pooled Connection will 328 * create before it either blocks or throws an exception when a new session is requested, 329 * depending on configuration. 330 * 331 * @return the number of session instances that can be taken from a pooled connection. 332 */ 333 public int getMaximumActiveSessionPerConnection() { 334 return maximumActiveSessionPerConnection; 335 } 336 337 /** 338 * Sets the maximum number of active sessions per connection 339 * 340 * @param maximumActiveSessionPerConnection 341 * The maximum number of active session per connection in the pool. 342 */ 343 public void setMaximumActiveSessionPerConnection(int maximumActiveSessionPerConnection) { 344 this.maximumActiveSessionPerConnection = maximumActiveSessionPerConnection; 345 } 346 347 /** 348 * Controls the behavior of the internal session pool. By default the call to 349 * Connection.getSession() will block if the session pool is full. If the 350 * argument false is given, it will change the default behavior and instead the 351 * call to getSession() will throw a JMSException. 352 * 353 * The size of the session pool is controlled by the @see #maximumActive 354 * property. 355 * 356 * @param block - if true, the call to getSession() blocks if the pool is full 357 * until a session object is available. defaults to true. 358 */ 359 public void setBlockIfSessionPoolIsFull(boolean block) { 360 this.blockIfSessionPoolIsFull = block; 361 } 362 363 /** 364 * Returns whether a pooled Connection will enter a blocked state or will throw an Exception 365 * once the maximum number of sessions has been borrowed from the the Session Pool. 366 * 367 * @return true if the pooled Connection createSession method will block when the limit is hit. 368 * @see #setBlockIfSessionPoolIsFull(boolean) 369 */ 370 public boolean isBlockIfSessionPoolIsFull() { 371 return this.blockIfSessionPoolIsFull; 372 } 373 374 /** 375 * Returns the maximum number to pooled Connections that this factory will allow before it 376 * begins to return connections from the pool on calls to ({@link #createConnection}. 377 * 378 * @return the maxConnections that will be created for this pool. 379 */ 380 public int getMaxConnections() { 381 return getConnectionsPool().getMaxIdle(); 382 } 383 384 /** 385 * Sets the maximum number of pooled Connections (defaults to one). Each call to 386 * {@link #createConnection} will result in a new Connection being create up to the max 387 * connections value. 388 * 389 * @param maxConnections the maxConnections to set 390 */ 391 public void setMaxConnections(int maxConnections) { 392 getConnectionsPool().setMaxIdle(maxConnections); 393 } 394 395 /** 396 * Gets the Idle timeout value applied to new Connection's that are created by this pool. 397 * <p/> 398 * The idle timeout is used determine if a Connection instance has sat to long in the pool unused 399 * and if so is closed and removed from the pool. The default value is 30 seconds. 400 * 401 * @return idle timeout value (milliseconds) 402 */ 403 public int getIdleTimeout() { 404 return idleTimeout; 405 } 406 407 /** 408 * Sets the idle timeout value for Connection's that are created by this pool in Milliseconds, 409 * defaults to 30 seconds. 410 * <p/> 411 * For a Connection that is in the pool but has no current users the idle timeout determines how 412 * long the Connection can live before it is eligible for removal from the pool. Normally the 413 * connections are tested when an attempt to check one out occurs so a Connection instance can sit 414 * in the pool much longer than its idle timeout if connections are used infrequently. 415 * 416 * @param idleTimeout 417 * The maximum time a pooled Connection can sit unused before it is eligible for removal. 418 */ 419 public void setIdleTimeout(int idleTimeout) { 420 this.idleTimeout = idleTimeout; 421 } 422 423 /** 424 * allow connections to expire, irrespective of load or idle time. This is useful with failover 425 * to force a reconnect from the pool, to reestablish load balancing or use of the master post recovery 426 * 427 * @param expiryTimeout non zero in milliseconds 428 */ 429 public void setExpiryTimeout(long expiryTimeout) { 430 this.expiryTimeout = expiryTimeout; 431 } 432 433 /** 434 * @return the configured expiration timeout for connections in the pool. 435 */ 436 public long getExpiryTimeout() { 437 return expiryTimeout; 438 } 439 440 /** 441 * @return true if a Connection is created immediately on a call to {@link start}. 442 */ 443 public boolean isCreateConnectionOnStartup() { 444 return createConnectionOnStartup; 445 } 446 447 /** 448 * Whether to create a connection on starting this {@link PooledConnectionFactory}. 449 * <p/> 450 * This can be used to warm-up the pool on startup. Notice that any kind of exception 451 * happens during startup is logged at WARN level and ignored. 452 * 453 * @param createConnectionOnStartup <tt>true</tt> to create a connection on startup 454 */ 455 public void setCreateConnectionOnStartup(boolean createConnectionOnStartup) { 456 this.createConnectionOnStartup = createConnectionOnStartup; 457 } 458 459 /** 460 * Should Sessions use one anonymous producer for all producer requests or should a new 461 * MessageProducer be created for each request to create a producer object, default is true. 462 * 463 * When enabled the session only needs to allocate one MessageProducer for all requests and 464 * the MessageProducer#send(destination, message) method can be used. Normally this is the 465 * right thing to do however it does result in the Broker not showing the producers per 466 * destination. 467 * 468 * @return true if a PooledSession will use only a single anonymous message producer instance. 469 */ 470 public boolean isUseAnonymousProducers() { 471 return this.useAnonymousProducers; 472 } 473 474 /** 475 * Sets whether a PooledSession uses only one anonymous MessageProducer instance or creates 476 * a new MessageProducer for each call the create a MessageProducer. 477 * 478 * @param value 479 * Boolean value that configures whether anonymous producers are used. 480 */ 481 public void setUseAnonymousProducers(boolean value) { 482 this.useAnonymousProducers = value; 483 } 484 485 /** 486 * Gets the Pool of ConnectionPool instances which are keyed by different ConnectionKeys. 487 * 488 * @return this factories pool of ConnectionPool instances. 489 */ 490 protected GenericKeyedObjectPool<ConnectionKey, ConnectionPool> getConnectionsPool() { 491 initConnectionsPool(); 492 return this.connectionsPool; 493 } 494 495 /** 496 * Sets the number of milliseconds to sleep between runs of the idle Connection eviction thread. 497 * When non-positive, no idle object eviction thread will be run, and Connections will only be 498 * checked on borrow to determine if they have sat idle for too long or have failed for some 499 * other reason. 500 * <p/> 501 * By default this value is set to -1 and no expiration thread ever runs. 502 * 503 * @param timeBetweenExpirationCheckMillis 504 * The time to wait between runs of the idle Connection eviction thread. 505 */ 506 public void setTimeBetweenExpirationCheckMillis(long timeBetweenExpirationCheckMillis) { 507 getConnectionsPool().setTimeBetweenEvictionRunsMillis(timeBetweenExpirationCheckMillis); 508 } 509 510 /** 511 * @return the number of milliseconds to sleep between runs of the idle connection eviction thread. 512 */ 513 public long getTimeBetweenExpirationCheckMillis() { 514 return getConnectionsPool().getTimeBetweenEvictionRunsMillis(); 515 } 516 517 /** 518 * @return the number of Connections currently in the Pool 519 */ 520 public int getNumConnections() { 521 return getConnectionsPool().getNumIdle(); 522 } 523 524 /** 525 * Delegate that creates each instance of an ConnectionPool object. Subclasses can override 526 * this method to customize the type of connection pool returned. 527 * 528 * @param connection 529 * 530 * @return instance of a new ConnectionPool. 531 */ 532 protected ConnectionPool createConnectionPool(Connection connection) { 533 return new ConnectionPool(connection); 534 } 535 536 /** 537 * Returns the timeout to use for blocking creating new sessions 538 * 539 * @return true if the pooled Connection createSession method will block when the limit is hit. 540 * @see #setBlockIfSessionPoolIsFull(boolean) 541 */ 542 public long getBlockIfSessionPoolIsFullTimeout() { 543 return blockIfSessionPoolIsFullTimeout; 544 } 545 546 /** 547 * Controls the behavior of the internal session pool. By default the call to 548 * Connection.getSession() will block if the session pool is full. This setting 549 * will affect how long it blocks and throws an exception after the timeout. 550 * 551 * The size of the session pool is controlled by the @see #maximumActive 552 * property. 553 * 554 * Whether or not the call to create session blocks is controlled by the @see #blockIfSessionPoolIsFull 555 * property 556 * 557 * @param blockIfSessionPoolIsFullTimeout - if blockIfSessionPoolIsFullTimeout is true, 558 * then use this setting to configure how long to block before retry 559 */ 560 public void setBlockIfSessionPoolIsFullTimeout(long blockIfSessionPoolIsFullTimeout) { 561 this.blockIfSessionPoolIsFullTimeout = blockIfSessionPoolIsFullTimeout; 562 } 563 564 /** 565 * @return true if the underlying connection will be renewed on JMSException, false otherwise 566 */ 567 public boolean isReconnectOnException() { 568 return reconnectOnException; 569 } 570 571 /** 572 * Controls weather the underlying connection should be reset (and renewed) on JMSException 573 * 574 * @param reconnectOnException 575 * Boolean value that configures whether reconnect on exception should happen 576 */ 577 public void setReconnectOnException(boolean reconnectOnException) { 578 this.reconnectOnException = reconnectOnException; 579 } 580 581 /** 582 * Called by any superclass that implements a JNDIReferencable or similar that needs to collect 583 * the properties of this class for storage etc. 584 * 585 * This method should be updated any time there is a new property added. 586 * 587 * @param props 588 * a properties object that should be filled in with this objects property values. 589 */ 590 protected void populateProperties(Properties props) { 591 props.setProperty("maximumActiveSessionPerConnection", Integer.toString(getMaximumActiveSessionPerConnection())); 592 props.setProperty("maxConnections", Integer.toString(getMaxConnections())); 593 props.setProperty("idleTimeout", Integer.toString(getIdleTimeout())); 594 props.setProperty("expiryTimeout", Long.toString(getExpiryTimeout())); 595 props.setProperty("timeBetweenExpirationCheckMillis", Long.toString(getTimeBetweenExpirationCheckMillis())); 596 props.setProperty("createConnectionOnStartup", Boolean.toString(isCreateConnectionOnStartup())); 597 props.setProperty("useAnonymousProducers", Boolean.toString(isUseAnonymousProducers())); 598 props.setProperty("blockIfSessionPoolIsFullTimeout", Long.toString(getBlockIfSessionPoolIsFullTimeout())); 599 props.setProperty("reconnectOnException", Boolean.toString(isReconnectOnException())); 600 } 601}