聊聊artemis的connectionTtlCheckInterval

本文主要研究一下artemis的connectionTtlCheckInterval

connectionTtlCheckInterval

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java

public class ConfigurationImpl implements Configuration, Serializable {
   private long connectionTtlCheckInterval = ActiveMQDefaultConfiguration.getDefaultConnectionTtlCheckInterval();
   //......
   @Override
   public long getConnectionTtlCheckInterval() {
      return connectionTtlCheckInterval;
   }
   @Override
   public ConfigurationImpl setConnectionTtlCheckInterval(long connectionTtlCheckInterval) {
      this.connectionTtlCheckInterval = connectionTtlCheckInterval;
      return this;
   }
   //......
}
  • ConfigurationImpl定义了connectionTtlCheckInterval属性,默认为2000

RemotingServiceImpl

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java

public class RemotingServiceImpl implements RemotingService, ServerConnectionLifeCycleListener {
   //......
   private FailureCheckAndFlushThread failureCheckAndFlushThread;
   private long connectionTtlCheckInterval;
   //......
   public synchronized void start() throws Exception {
      if (started) {
         return;
      }
      logger.tracef("Starting remoting service %s", this);
      paused = false;
      // The remoting service maintains it's own thread pool for handling remoting traffic
      // If OIO each connection will have it's own thread
      // If NIO these are capped at nio-remoting-threads which defaults to num cores * 3
      // This needs to be a different thread pool to the main thread pool especially for OIO where we may need
      // to support many hundreds of connections, but the main thread pool must be kept small for better performance
      ThreadFactory tFactory = AccessController.doPrivileged(new PrivilegedAction<ThreadFactory>() {
         @Override
         public ThreadFactory run() {
            return new ActiveMQThreadFactory("ActiveMQ-remoting-threads-" + server.toString() +
                                                "-" +
                                                System.identityHashCode(this), false, Thread.currentThread().getContextClassLoader());
         }
      });
      threadPool = Executors.newCachedThreadPool(tFactory);
      for (TransportConfiguration info : acceptorsConfig) {
         createAcceptor(info);
      }
      /**
       * Don't start the acceptors here.  Only start the acceptors at the every end of the start-up process to avoid
       * race conditions. See {@link #startAcceptors()}.
       */
      // This thread checks connections that need to be closed, and also flushes confirmations
      failureCheckAndFlushThread = new FailureCheckAndFlushThread(connectionTtlCheckInterval);
      failureCheckAndFlushThread.start();
      started = true;
   }
   //......
}
  • RemotingServiceImpl的start方法使用connectionTtlCheckInterval创建了FailureCheckAndFlushThread,并执行failureCheckAndFlushThread.start()方法

FailureCheckAndFlushThread

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java

   private final class FailureCheckAndFlushThread extends Thread {
      private final long pauseInterval;
      private volatile boolean closed;
      private final CountDownLatch latch = new CountDownLatch(1);
      FailureCheckAndFlushThread(final long pauseInterval) {
         super("activemq-failure-check-thread");
         this.pauseInterval = pauseInterval;
      }
      public void close(final boolean criticalError) {
         closed = true;
         latch.countDown();
         if (!criticalError) {
            try {
               join();
            } catch (InterruptedException e) {
               throw new ActiveMQInterruptedException(e);
            }
         }
      }
      @Override
      public void run() {
         while (!closed) {
            try {
               long now = System.currentTimeMillis();
               Set<Pair<Object, Long>> toRemove = new HashSet<>();
               for (ConnectionEntry entry : connections.values()) {
                  final RemotingConnection conn = entry.connection;
                  boolean flush = true;
                  if (entry.ttl != -1) {
                     if (!conn.checkDataReceived()) {
                        if (now >= entry.lastCheck + entry.ttl) {
                           toRemove.add(new Pair<>(conn.getID(), entry.ttl));
                           flush = false;
                        }
                     } else {
                        entry.lastCheck = now;
                     }
                  }
                  if (flush) {
                     flushExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                           try {
                              // this is using a different thread
                              // as if anything wrong happens on flush
                              // failure detection could be affected
                              conn.scheduledFlush();
                           } catch (Throwable e) {
                              ActiveMQServerLogger.LOGGER.failedToFlushOutstandingDataFromTheConnection(e);
                           }
                        }
                     });
                  }
               }
               for (final Pair<Object, Long> pair : toRemove) {
                  final RemotingConnection conn = getConnection(pair.getA());
                  if (conn != null) {
                     // In certain cases (replicationManager for instance) calling fail could take some time
                     // We can't pause the FailureCheckAndFlushThread as that would lead other clients to fail for
                     // missing pings
                     flushExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                           conn.fail(ActiveMQMessageBundle.BUNDLE.clientExited(conn.getRemoteAddress(), pair.getB()));
                        }
                     });
                     removeConnection(pair.getA());
                  }
               }
               if (latch.await(pauseInterval, TimeUnit.MILLISECONDS))
                  return;
            } catch (Throwable e) {
               ActiveMQServerLogger.LOGGER.errorOnFailureCheck(e);
            }
         }
      }
   }
  • FailureCheckAndFlushThread继承了Thread,其run方法会遍历connections的ConnectionEntry,对于entry.ttl不为-1的执行conn.checkDataReceived(),若为true则更新entry.lastCheck,为false则计算entry.lastCheck + entry.ttl,若小于等于当前时间,则添加到toRemove并设置flush为false;之后对于flush为true的调度执行conn.scheduledFlush();最后遍历toRemove的conn,调度执行conn.fail(ActiveMQMessageBundle.BUNDLE.clientExited(conn.getRemoteAddress(), pair.getB()))

小结

ConfigurationImpl定义了connectionTtlCheckInterval属性,默认为2000;RemotingServiceImpl的start方法使用connectionTtlCheckInterval创建了FailureCheckAndFlushThread,并执行failureCheckAndFlushThread.start()方法

doc

https://juejin.im/post/5e381398f265da5729787a2c

「点点赞赏,手留余香」

    还没有人赞赏,快来当第一个赞赏的人吧!
0 条回复 A 作者 M 管理员
    所有的伟大,都源于一个勇敢的开始!
欢迎您,新朋友,感谢参与互动!欢迎您 {{author}},您在本站有{{commentsCount}}条评论