web/tomcat/tomcat基础

tomcat源码编译： https://blog.csdn.net/qq_35262405/article/details/101780644

Tomcat

init

LifecycleBase
    所有的组件都继承这个抽象类
    
    public final synchronized void init() throws LifecycleException {
        if (!state.equals(LifecycleState.NEW)) {
            invalidTransition(Lifecycle.BEFORE_INIT_EVENT);
        }

        try {
            setStateInternal(LifecycleState.INITIALIZING, null, false);
            //初始化内部
            initInternal();
            setStateInternal(LifecycleState.INITIALIZED, null, false);
        } catch (Throwable t) {
            handleSubClassException(t, "lifecycleBase.initFail", toString());
        }
    }

Server的初始化

initInternal

 @Override
    protected void initInternal() throws LifecycleException {

        super.initInternal();

        // Register global String cache
        // Note although the cache is global, if there are multiple Servers
        // present in the JVM (may happen when embedding) then the same cache
        // will be registered under multiple names
        onameStringCache = register(new StringCache(), "type=StringCache");

        // Register the MBeanFactory
        MBeanFactory factory = new MBeanFactory();
        factory.setContainer(this);
        onameMBeanFactory = register(factory, "type=MBeanFactory");

        // Register the naming resources
        globalNamingResources.init();

        // Populate the extension validator with JARs from common and shared
        // class loaders
        if (getCatalina() != null) {
            ClassLoader cl = getCatalina().getParentClassLoader();
            // Walk the class loader hierarchy. Stop at the system class loader.
            // This will add the shared (if present) and common class loaders
            while (cl != null && cl != ClassLoader.getSystemClassLoader()) {
                if (cl instanceof URLClassLoader) {
                    URL[] urls = ((URLClassLoader) cl).getURLs();
                    for (URL url : urls) {
                        if (url.getProtocol().equals("file")) {
                            try {
                                File f = new File (url.toURI());
                                if (f.isFile() &&
                                        f.getName().endsWith(".jar")) {
                                    ExtensionValidator.addSystemResource(f);
                                }
                            } catch (URISyntaxException e) {
                                // Ignore
                            } catch (IOException e) {
                                // Ignore
                            }
                        }
                    }
                }
                cl = cl.getParent();
            }
        }
        // Initialize our defined Services
        // 初始内部的service
        for (Service service : services) {
            service.init();
        }
    }
所以 一个Server里面有多个Service

Service初始化

initInternal

@Override
  protected void initInternal() throws LifecycleException {

      super.initInternal();
// 一个Service中有一个引擎engine
      if (engine != null) {
          engine.init();
      }
// 一个Service中有多个执行器
      // Initialize any Executors
      for (Executor executor : findExecutors()) {
          if (executor instanceof JmxEnabled) {
              ((JmxEnabled) executor).setDomain(getDomain());
          }
          executor.init();
      }
//  一个Service中有一个映射监听器
      // Initialize mapper listener
      mapperListener.init();

      // 一个Service中有多个连接器
      // Initialize our defined Connectors
      synchronized (connectorsLock) {
          for (Connector connector : connectors) {
              try {
                  connector.init();
              } catch (Exception e) {
                  String message = sm.getString(
                          "standardService.connector.initFailed", connector);
                  log.error(message, e);

                  if (Boolean.getBoolean("org.apache.catalina.startup.EXIT_ON_INIT_FAILURE"))
                      throw new LifecycleException(message);
              }
          }
      }
  }

Engine初始化

initInternal

    @Override
    protected void initInternal() throws LifecycleException {
        // Ensure that a Realm is present before any attempt is made to start
        // one. This will create the default NullRealm if necessary.
        getRealm();
        super.initInternal();
    }
ContainerBase
    @Override
    protected void initInternal() throws LifecycleException {
        BlockingQueue<Runnable> startStopQueue = new LinkedBlockingQueue<>();
    //初始化一个启动或者停止的线程池
        startStopExecutor = new ThreadPoolExecutor(
                getStartStopThreadsInternal(),
                getStartStopThreadsInternal(), 10, TimeUnit.SECONDS,
                startStopQueue,
                new StartStopThreadFactory(getName() + "-startStop-"));
        startStopExecutor.allowCoreThreadTimeOut(true);
        super.initInternal();
    }

mapperListener初始化

initInternal

@Override
protected void initInternal() throws LifecycleException {
    // If oname is not null then registration has already happened via
    // preRegister().
    if (oname == null) {
        mserver = Registry.getRegistry(null, null).getMBeanServer();

        oname = register(this, getObjectNameKeyProperties());
    }
}

Connector初始化

initInternal

protected void initInternal() throws LifecycleException {

       super.initInternal();

       // Initialize adapter
       adapter = new CoyoteAdapter(this);
       protocolHandler.setAdapter(adapter);

       // Make sure parseBodyMethodsSet has a default
       if (null == parseBodyMethodsSet) {
           setParseBodyMethods(getParseBodyMethods());
       }

       if (protocolHandler.isAprRequired() && !AprLifecycleListener.isInstanceCreated()) {
           throw new LifecycleException(sm.getString("coyoteConnector.protocolHandlerNoAprListener",
                   getProtocolHandlerClassName()));
       }
       if (protocolHandler.isAprRequired() && !AprLifecycleListener.isAprAvailable()) {
           throw new LifecycleException(sm.getString("coyoteConnector.protocolHandlerNoAprLibrary",
                   getProtocolHandlerClassName()));
       }
       if (AprLifecycleListener.isAprAvailable() && AprLifecycleListener.getUseOpenSSL() &&
               protocolHandler instanceof AbstractHttp11JsseProtocol) {
           AbstractHttp11JsseProtocol<?> jsseProtocolHandler =
                   (AbstractHttp11JsseProtocol<?>) protocolHandler;
           if (jsseProtocolHandler.isSSLEnabled() &&
                   jsseProtocolHandler.getSslImplementationName() == null) {
               // OpenSSL is compatible with the JSSE configuration, so use it if APR is available
               jsseProtocolHandler.setSslImplementationName(OpenSSLImplementation.class.getName());
           }
       }

       try {
           //一个连接器只有一个处理器
           protocolHandler.init();
       } catch (Exception e) {
           throw new LifecycleException(
                   sm.getString("coyoteConnector.protocolHandlerInitializationFailed"), e);
       }
   }

protocolHandler初始化

  @Override
   public void init() throws Exception {
       // Upgrade protocols have to be configured first since the endpoint
       // init (triggered via super.init() below) uses this list to configure
       // the list of ALPN protocols to advertise
       for (UpgradeProtocol upgradeProtocol : upgradeProtocols) {
           configureUpgradeProtocol(upgradeProtocol);
       }

       super.init();
   }


public void init() throws Exception {
       if (getLog().isInfoEnabled()) {
           getLog().info(sm.getString("abstractProtocolHandler.init", getName()));
       }

       if (oname == null) {
           // Component not pre-registered so register it
           oname = createObjectName();
           if (oname != null) {
               Registry.getRegistry(null, null).registerComponent(this, oname, null);
           }
       }

       if (this.domain != null) {
           rgOname = new ObjectName(domain + ":type=GlobalRequestProcessor,name=" + getName());
           Registry.getRegistry(null, null).registerComponent(
                   getHandler().getGlobal(), rgOname, null);
       }

       String endpointName = getName();
       endpoint.setName(endpointName.substring(1, endpointName.length()-1));
       endpoint.setDomain(domain);
	// 一个协议处理器只有一个 远程连接对象
       endpoint.init();
   }

endpoint初始化

   @Override
   public void init() throws Exception {
       testServerCipherSuitesOrderSupport();
       super.init();
   }

public void init() throws Exception {
       if (bindOnInit) {
           bind();
           bindState = BindState.BOUND_ON_INIT;
       }
       if (this.domain != null) {
           // Register endpoint (as ThreadPool - historical name)
           oname = new ObjectName(domain + ":type=ThreadPool,name=\"" + getName() + "\"");
           Registry.getRegistry(null, null).registerComponent(this, oname, null);

           ObjectName socketPropertiesOname = new ObjectName(domain +
                   ":type=SocketProperties,name=\"" + getName() + "\"");
           socketProperties.setObjectName(socketPropertiesOname);
           Registry.getRegistry(null, null).registerComponent(socketProperties, socketPropertiesOname, null);

           for (SSLHostConfig sslHostConfig : findSslHostConfigs()) {
               registerJmx(sslHostConfig);
           }
       }
   }

bind

  @Override
    public void bind() throws Exception {

        if (!getUseInheritedChannel()) {
            //NIO编程
            serverSock = ServerSocketChannel.open();
            socketProperties.setProperties(serverSock.socket());
            //获取网络地址
            InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
           
 //指定连接数，默认100
  serverSock.socket().bind(addr,getAcceptCount());
        } else {
            // Retrieve the channel provided by the OS
            Channel ic = System.inheritedChannel();
            if (ic instanceof ServerSocketChannel) {
                serverSock = (ServerSocketChannel) ic;
            }
            if (serverSock == null) {
                throw new IllegalArgumentException(sm.getString("endpoint.init.bind.inherited"));
            }
        }
        // 注意看，接待Channel是阻塞的，
        serverSock.configureBlocking(true); //mimic APR behavior，模板ARP行为

        // Initialize thread count defaults for acceptor, poller
        //接待的线程数，默认为1个，也只需要一个
        if (acceptorThreadCount == 0) {
            // FIXME: Doesn't seem to work that well with multiple accept threads
            acceptorThreadCount = 1;
        }
        // poller的线程数，默认为两个
        if (pollerThreadCount <= 0) {
            //minimum one poller thread
            pollerThreadCount = 1;
        }
        //设置一把锁，CountDownLatch，数量为poll的线程数
        //猜一下就是两个线程都启动了，这把锁才打开
        setStopLatch(new CountDownLatch(pollerThreadCount));

        //初始化SSL
        // Initialize SSL if needed
        initialiseSsl();

        selectorPool.open();
    }

//----
selectorPool
    public void open() throws IOException {
        enabled = true;
    //就是去获取一个共享Seletor。NIO编程
        getSharedSelector();
        if (SHARED) {
            //把这个 获取到的Selector给到blockingSelector
            blockingSelector = new NioBlockingSelector();
            blockingSelector.open(getSharedSelector());
        }

    }

NioBlockingSelector

public void open(Selector selector) {
     sharedSelector = selector;
     poller = new BlockPoller();
     poller.selector = sharedSelector;
     poller.setDaemon(true);
     poller.setName("NioBlockingSelector.BlockPoller-"+(threadCounter.getAndIncrement()));
     poller.start();
 }

BlockPoller

protected static class BlockPoller extends Thread {
    //是一个线程
    
     protected volatile boolean run = true;
        protected Selector selector = null;
        protected final SynchronizedQueue<Runnable> events = new SynchronizedQueue<>();
        public void disable() { run = false; selector.wakeup();}
        protected final AtomicInteger wakeupCounter = new AtomicInteger(0);
    
    
    
}

run

public void run() {
          while (run) {
              try {
                  //从队列中取事件进行调用
                  events();
                  int keyCount = 0;
                  try {
                      // 使用selector进行对感兴趣io事件选择
                      int i = wakeupCounter.get();
                      if (i>0)
                          keyCount = selector.selectNow();
                      else {
                          wakeupCounter.set(-1);
                          keyCount = selector.select(1000);
                      }
                      wakeupCounter.set(0);
                      if (!run) break;
                  }catch ( NullPointerException x ) {
                      //sun bug 5076772 on windows JDK 1.5
                      if (selector==null) throw x;
                      if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                      continue;
                  } catch ( CancelledKeyException x ) {
                      //sun bug 5076772 on windows JDK 1.5
                      if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                      continue;
                  } catch (Throwable x) {
                      ExceptionUtils.handleThrowable(x);
                      log.error("",x);
                      continue;
                  }
			
                  Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;

                  // Walk through the collection of ready keys and dispatch
                  // any active event.
                  while (run && iterator != null && iterator.hasNext()) {
                      SelectionKey sk = iterator.next();
                      NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
                      try {
                          iterator.remove();
                          //拿出sk
                          sk.interestOps(sk.interestOps() & (~sk.readyOps()));
                          if ( sk.isReadable() ) {
                              //把读锁减-
                              countDown(attachment.getReadLatch());
                          }
                          if (sk.isWritable()) {
                              //把写锁减-
                              countDown(attachment.getWriteLatch());
                          }
                      }catch (CancelledKeyException ckx) {
                          sk.cancel();
                          countDown(attachment.getReadLatch());
                          countDown(attachment.getWriteLatch());
                      }
                  }//while
              }catch ( Throwable t ) {
                  log.error("",t);
              }
          }
    
    // 正常执行结束
          events.clear();
          // If using a shared selector, the NioSelectorPool will also try and
          // close the selector. Try and avoid the ClosedSelectorException
          // although because multiple threads are involved there is always
          // the possibility of an Exception here.
          if (selector.isOpen()) {
              try {
                  // Cancels all remaining keys
                  selector.selectNow();
              }catch( Exception ignore ) {
                  if (log.isDebugEnabled())log.debug("",ignore);
              }
          }
          try {
              selector.close();
          }catch( Exception ignore ) {
              if (log.isDebugEnabled())log.debug("",ignore);
          }
      }

start

@Override
   public final synchronized void start() throws LifecycleException {

       if (LifecycleState.STARTING_PREP.equals(state) || LifecycleState.STARTING.equals(state) ||
               LifecycleState.STARTED.equals(state)) {

           if (log.isDebugEnabled()) {
               Exception e = new LifecycleException();
               log.debug(sm.getString("lifecycleBase.alreadyStarted", toString()), e);
           } else if (log.isInfoEnabled()) {
               log.info(sm.getString("lifecycleBase.alreadyStarted", toString()));
           }

           return;
       }

       if (state.equals(LifecycleState.NEW)) {
           init();
       } else if (state.equals(LifecycleState.FAILED)) {
           stop();
       } else if (!state.equals(LifecycleState.INITIALIZED) &&
               !state.equals(LifecycleState.STOPPED)) {
           invalidTransition(Lifecycle.BEFORE_START_EVENT);
       }

       try {
           setStateInternal(LifecycleState.STARTING_PREP, null, false);
           startInternal();
           if (state.equals(LifecycleState.FAILED)) {
               // This is a 'controlled' failure. The component put itself into the
               // FAILED state so call stop() to complete the clean-up.
               stop();
           } else if (!state.equals(LifecycleState.STARTING)) {
               // Shouldn't be necessary but acts as a check that sub-classes are
               // doing what they are supposed to.
               invalidTransition(Lifecycle.AFTER_START_EVENT);
           } else {
               setStateInternal(LifecycleState.STARTED, null, false);
           }
       } catch (Throwable t) {
           // This is an 'uncontrolled' failure so put the component into the
           // FAILED state and throw an exception.
           handleSubClassException(t, "lifecycleBase.startFail", toString());
       }
   }

Server的启动

startInternal

@Override
   protected void startInternal() throws LifecycleException {
	//发射声明周期的事件
       fireLifecycleEvent(CONFIGURE_START_EVENT, null);
       setState(LifecycleState.STARTING);

       globalNamingResources.start();

       // Start our defined Services
       synchronized (servicesLock) {
           for (Service service : services) {
               service.start();
           }
       }
   }

globalNamingResources启动

@Override
protected void startInternal() throws LifecycleException {
    fireLifecycleEvent(CONFIGURE_START_EVENT, null);
    setState(LifecycleState.STARTING);
}

Service启动

@Override
   protected void startInternal() throws LifecycleException {

       if(log.isInfoEnabled())
           log.info(sm.getString("standardService.start.name", this.name));
       setState(LifecycleState.STARTING);
	//引擎启动
       // Start our defined Container first
       if (engine != null) {
           synchronized (engine) {
               engine.start();
           }
       }
	// 执行器启动
       synchronized (executors) {
           for (Executor executor: executors) {
               executor.start();
           }
       }
	//mapperListener启动
       
       mapperListener.start();

       // Start our defined Connectors second
       synchronized (connectorsLock) {
           for (Connector connector: connectors) {
               try {
                   // If it has already failed, don't try and start it
                   if (connector.getState() != LifecycleState.FAILED) {
                       //连接器启动
                       connector.start();
                   }
               } catch (Exception e) {
                   log.error(sm.getString(
                           "standardService.connector.startFailed",
                           connector), e);
               }
           }
       }
   }

Engine启动

 protected synchronized void startInternal() throws LifecycleException {

        // Log our server identification information
        if(log.isInfoEnabled())
            log.info( "Starting Servlet Engine: " + ServerInfo.getServerInfo());

        // Standard container startup
        super.startInternal();
    }


@Override
    protected synchronized void startInternal() throws LifecycleException {

        // Start our subordinate components, if any
        logger = null;
        getLogger();
        Cluster cluster = getClusterInternal();
        if (cluster instanceof Lifecycle) {
            ((Lifecycle) cluster).start();
        }
        Realm realm = getRealmInternal();
        if (realm instanceof Lifecycle) {
            ((Lifecycle) realm).start();
        }
		
        //Host交给线程池启动
        // Start our child containers, if any
        Container children[] = findChildren();
        List<Future<Void>> results = new ArrayList<>();
        for (Container child : children) {
            results.add(startStopExecutor.submit(new StartChild(child)));
        }

        MultiThrowable multiThrowable = null;

        for (Future<Void> result : results) {
            try {
                result.get();
            } catch (Throwable e) {
                log.error(sm.getString("containerBase.threadedStartFailed"), e);
                if (multiThrowable == null) {
                    multiThrowable = new MultiThrowable();
                }
                multiThrowable.add(e);
            }

        }
        if (multiThrowable != null) {
            throw new LifecycleException(sm.getString("containerBase.threadedStartFailed"),
                    multiThrowable.getThrowable());
        }

        // Start the Valves in our pipeline (including the basic), if any
        if (pipeline instanceof Lifecycle) {
            ((Lifecycle) pipeline).start();
        }


        setState(LifecycleState.STARTING);

        // Start our thread
        threadStart();
    }

threadStart

protected class ContainerBackgroundProcessor implements Runnable {

        @Override
        public void run() {
            Throwable t = null;
            String unexpectedDeathMessage = sm.getString(
                    "containerBase.backgroundProcess.unexpectedThreadDeath",
                    Thread.currentThread().getName());
            try {
                while (!threadDone) {
                    try {
                        Thread.sleep(backgroundProcessorDelay * 1000L);
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                    if (!threadDone) {
                        processChildren(ContainerBase.this);
                    }
                }
            } catch (RuntimeException|Error e) {
                t = e;
                throw e;
            } finally {
                if (!threadDone) {
                    log.error(unexpectedDeathMessage, t);
                }
            }
        }

        protected void processChildren(Container container) {
            ClassLoader originalClassLoader = null;

            try {
                if (container instanceof Context) {
                    Loader loader = ((Context) container).getLoader();
                    // Loader will be null for FailedContext instances
                    if (loader == null) {
                        return;
                    }

                    // Ensure background processing for Contexts and Wrappers
                    // is performed under the web app's class loader
                    originalClassLoader = ((Context) container).bind(false, null);
                }
                container.backgroundProcess();
                Container[] children = container.findChildren();
                for (Container child : children) {
                    if (child.getBackgroundProcessorDelay() <= 0) {
                        processChildren(child);
                    }
                }
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error("Exception invoking periodic operation: ", t);
            } finally {
                if (container instanceof Context) {
                    ((Context) container).unbind(false, originalClassLoader);
               }
            }
        }
    }

pipeline启动

@Override
    protected synchronized void startInternal() throws LifecycleException {

        // Start the Valves in our pipeline (including the basic), if any
        Valve current = first;
        if (current == null) {
            current = basic;
        }
        while (current != null) {
            if (current instanceof Lifecycle)
                ((Lifecycle) current).start();
            current = current.getNext();
        }

        setState(LifecycleState.STARTING);
    }

MapperListener启动

public void startInternal() throws LifecycleException {

        setState(LifecycleState.STARTING);

        Engine engine = service.getContainer();
        if (engine == null) {
            return;
        }

        findDefaultHost();

        addListeners(engine);

        Container[] conHosts = engine.findChildren();
        for (Container conHost : conHosts) {
            Host host = (Host) conHost;
            if (!LifecycleState.NEW.equals(host.getState())) {
                // Registering the host will register the context and wrappers
                registerHost(host);
            }
        }
    }

findDefaultHost

private void findDefaultHost() {

     Engine engine = service.getContainer();
     String defaultHost = engine.getDefaultHost();

     boolean found = false;

     if (defaultHost != null && defaultHost.length() >0) {
         Container[] containers = engine.findChildren();

         for (Container container : containers) {
             Host host = (Host) container;
             if (defaultHost.equalsIgnoreCase(host.getName())) {
                 found = true;
                 break;
             }

             String[] aliases = host.findAliases();
             for (String alias : aliases) {
                 if (defaultHost.equalsIgnoreCase(alias)) {
                     found = true;
                     break;
                 }
             }
         }
     }

     if(found) {
         mapper.setDefaultHostName(defaultHost);
     } else {
         log.warn(sm.getString("mapperListener.unknownDefaultHost",
                 defaultHost, service));
     }
 }

addListeners

private void addListeners(Container container) {
    container.addContainerListener(this);
    container.addLifecycleListener(this);
    for (Container child : container.findChildren()) {
        addListeners(child);
    }
}

registerHost

private void registerHost(Host host) {

       String[] aliases = host.findAliases();
       mapper.addHost(host.getName(), aliases, host);

       for (Container container : host.findChildren()) {
           if (container.getState().isAvailable()) {
               registerContext((Context) container);
           }
       }
       if(log.isDebugEnabled()) {
           log.debug(sm.getString("mapperListener.registerHost",
                   host.getName(), domain, service));
  
       }
   }

registerContext

private void registerContext(Context context) {

      String contextPath = context.getPath();
      if ("/".equals(contextPath)) {
          contextPath = "";
      }
      Host host = (Host)context.getParent();

      WebResourceRoot resources = context.getResources();
      String[] welcomeFiles = context.findWelcomeFiles();
      List<WrapperMappingInfo> wrappers = new ArrayList<>();

      for (Container container : context.findChildren()) {
          prepareWrapperMappingInfo(context, (Wrapper) container, wrappers);

          if(log.isDebugEnabled()) {
              log.debug(sm.getString("mapperListener.registerWrapper",
                      container.getName(), contextPath, service));
          }
      }

      mapper.addContextVersion(host.getName(), host, contextPath,
              context.getWebappVersion(), context, welcomeFiles, resources,
              wrappers);

      if(log.isDebugEnabled()) {
          log.debug(sm.getString("mapperListener.registerContext",
                  contextPath, service));
      }
  }

prepareWrapperMappingInfo

private void prepareWrapperMappingInfo(Context context, Wrapper wrapper,
           List<WrapperMappingInfo> wrappers) {
       String wrapperName = wrapper.getName();
       boolean resourceOnly = context.isResourceOnlyServlet(wrapperName);
       String[] mappings = wrapper.findMappings();
       for (String mapping : mappings) {
           boolean jspWildCard = (wrapperName.equals("jsp")
                                  && mapping.endsWith("/*"));
           wrappers.add(new WrapperMappingInfo(mapping, wrapper, jspWildCard,
                   resourceOnly));
       }
   }

Connector启动

@Override
protected void startInternal() throws LifecycleException {

    // Validate settings before starting
    if (getPort() < 0) {
        throw new LifecycleException(sm.getString(
                "coyoteConnector.invalidPort", Integer.valueOf(getPort())));
    }

    setState(LifecycleState.STARTING);

    try {
        protocolHandler.start();
    } catch (Exception e) {
        throw new LifecycleException(
                sm.getString("coyoteConnector.protocolHandlerStartFailed"), e);
    }
}

protocolHandler启动

@Override
   public void start() throws Exception {
       if (getLog().isInfoEnabled()) {
           getLog().info(sm.getString("abstractProtocolHandler.start", getName()));
       }

       endpoint.start();

       // Start timeout thread
       asyncTimeout = new AsyncTimeout();
       Thread timeoutThread = new Thread(asyncTimeout, getNameInternal() + "-AsyncTimeout");
       int priority = endpoint.getThreadPriority();
       if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
           priority = Thread.NORM_PRIORITY;
       }
       timeoutThread.setPriority(priority);
       timeoutThread.setDaemon(true);
       timeoutThread.start();
   }

AsyncTimeout

protected class AsyncTimeout implements Runnable {

      private volatile boolean asyncTimeoutRunning = true;

      /**
       * The background thread that checks async requests and fires the
       * timeout if there has been no activity.
       */
      @Override
      public void run() {

          // Loop until we receive a shutdown command
          while (asyncTimeoutRunning) {
              try {
                  Thread.sleep(1000);
              } catch (InterruptedException e) {
                  // Ignore
              }
              long now = System.currentTimeMillis();
              for (Processor processor : waitingProcessors) {
                 processor.timeoutAsync(now);
              }

              // Loop if endpoint is paused
              while (endpoint.isPaused() && asyncTimeoutRunning) {
                  try {
                      Thread.sleep(1000);
                  } catch (InterruptedException e) {
                      // Ignore
                  }
              }
          }
      }


      protected void stop() {
          asyncTimeoutRunning = false;

          // Timeout any pending async request
          for (Processor processor : waitingProcessors) {
              processor.timeoutAsync(-1);
          }
      }
  }

endpoint启动

public final void start() throws Exception {
      if (bindState == BindState.UNBOUND) {
          bind();
          bindState = BindState.BOUND_ON_START;
      }
      startInternal();
  }

startInternal

public void startInternal() throws Exception {

       if (!running) {
           running = true;
           paused = false;

           processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                   socketProperties.getProcessorCache());
           eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                           socketProperties.getEventCache());
           nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                   socketProperties.getBufferPool());

           // Create worker collection
           if ( getExecutor() == null ) {
               createExecutor();
           }

           initializeConnectionLatch();

           // Start poller threads
           // 两个poller的启动
           pollers = new Poller[getPollerThreadCount()];
           for (int i=0; i<pollers.length; i++) {
               pollers[i] = new Poller();
               Thread pollerThread = new Thread(pollers[i], getName() + "-ClientPoller-"+i);
               pollerThread.setPriority(threadPriority);
               pollerThread.setDaemon(true);
               pollerThread.start();
           }

           startAcceptorThreads();
       }
   }

createExecutor

public void createExecutor() {
    internalExecutor = true;
    TaskQueue taskqueue = new TaskQueue();
    TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority());
    //这个线程就是业务线程
    executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf);
    taskqueue.setParent( (ThreadPoolExecutor) executor);
}

initializeConnectionLatch

protected LimitLatch initializeConnectionLatch() {
    if (maxConnections==-1) return null;
    if (connectionLimitLatch==null) {
        connectionLimitLatch = new LimitLatch(getMaxConnections());
    }
    return connectionLimitLatch;
}

Poller

public class Poller implements Runnable {
 private Selector selector;
      private final SynchronizedQueue<PollerEvent> events =
              new SynchronizedQueue<>();

      private volatile boolean close = false;
      private long nextExpiration = 0;//optimize expiration handling

      private AtomicLong wakeupCounter = new AtomicLong(0);

      private volatile int keyCount = 0;   

    //poller的selector是新的
         public Poller() throws IOException {
          this.selector = Selector.open();
      }

}

run

@Override
      public void run() {
          // Loop until destroy() is called
          while (true) {

              boolean hasEvents = false;

              try {
                  if (!close) {
                      //取出所有事件并执行
                      hasEvents = events();
                      if (wakeupCounter.getAndSet(-1) > 0) {
                          //if we are here, means we have other stuff to do
                          //do a non blocking select
                          keyCount = selector.selectNow();
                      } else {
                          keyCount = selector.select(selectorTimeout);
                      }
                      wakeupCounter.set(0);
                  }
                  if (close) {
                      events();
                      timeout(0, false);
                      try {
                          selector.close();
                      } catch (IOException ioe) {
                          log.error(sm.getString("endpoint.nio.selectorCloseFail"), ioe);
                      }
                      break;
                  }
              } catch (Throwable x) {
                  ExceptionUtils.handleThrowable(x);
                  log.error("",x);
                  continue;
              }
              //either we timed out or we woke up, process events first
              if ( keyCount == 0 ) hasEvents = (hasEvents | events());

              Iterator<SelectionKey> iterator =
                  keyCount > 0 ? selector.selectedKeys().iterator() : null;
              // Walk through the collection of ready keys and dispatch
              // any active event.
              while (iterator != null && iterator.hasNext()) {
                  SelectionKey sk = iterator.next();
                  NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
                  // Attachment may be null if another thread has called
                  // cancelledKey()
                  if (attachment == null) {
                      iterator.remove();
                  } else {
                      iterator.remove();
                      processKey(sk, attachment);
                  }
              }//while

              //process timeouts
              timeout(keyCount,hasEvents);
          }//while

          getStopLatch().countDown();
      }

processKey

protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
            try {
                if ( close ) {
                    cancelledKey(sk);
                } else if ( sk.isValid() && attachment != null ) {
                    if (sk.isReadable() || sk.isWritable() ) {
                        if ( attachment.getSendfileData() != null ) {
                            processSendfile(sk,attachment, false);
                        } else {
                            unreg(sk, attachment, sk.readyOps());
                            boolean closeSocket = false;
                            // Read goes before write
                            if (sk.isReadable()) {
                                //交给业务线程处理
                                if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
                                    closeSocket = true;
                                }
                            }
                            if (!closeSocket && sk.isWritable()) {
                                if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
                                    closeSocket = true;
                                }
                            }
                            if (closeSocket) {
                                cancelledKey(sk);
                            }
                        }
                    }
                } else {
                    //invalid key
                    cancelledKey(sk);
                }
            } catch ( CancelledKeyException ckx ) {
                cancelledKey(sk);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error("",t);
            }
        }

startAcceptorThreads

    protected final void startAcceptorThreads() {
        int count = getAcceptorThreadCount();
        acceptors = new Acceptor[count];

        for (int i = 0; i < count; i++) {
            acceptors[i] = createAcceptor();
            String threadName = getName() + "-Acceptor-" + i;
            acceptors[i].setThreadName(threadName);
            Thread t = new Thread(acceptors[i], threadName);
            t.setPriority(getAcceptorThreadPriority());
            t.setDaemon(getDaemon());
            t.start();
        }
    }

NioEndpoint
   @Override
    protected AbstractEndpoint.Acceptor createAcceptor() {
        return new Acceptor();
    }

Acceptor

protected class Acceptor extends AbstractEndpoint.Acceptor {

       @Override
       public void run() {

           int errorDelay = 0;

           // Loop until we receive a shutdown command
           while (running) {

               // Loop if endpoint is paused
               while (paused && running) {
                   state = AcceptorState.PAUSED;
                   try {
                       Thread.sleep(50);
                   } catch (InterruptedException e) {
                       // Ignore
                   }
               }

               if (!running) {
                   break;
               }
               state = AcceptorState.RUNNING;

               try {
                   //if we have reached max connections, wait
                   countUpOrAwaitConnection();

                   SocketChannel socket = null;
                   try {
                       // Accept the next incoming connection from the server
                       // socket
                       socket = serverSock.accept();
                   } catch (IOException ioe) {
                       // We didn't get a socket
                       countDownConnection();
                       if (running) {
                           // Introduce delay if necessary
                           errorDelay = handleExceptionWithDelay(errorDelay);
                           // re-throw
                           throw ioe;
                       } else {
                           break;
                       }
                   }
                   // Successful accept, reset the error delay
                   errorDelay = 0;

                   // Configure the socket
                   if (running && !paused) {
                       // setSocketOptions() will hand the socket off to
                       // an appropriate processor if successful
                       if (!setSocketOptions(socket)) {
                           closeSocket(socket);
                       }
                   } else {
                       closeSocket(socket);
                   }
               } catch (Throwable t) {
                   ExceptionUtils.handleThrowable(t);
                   log.error(sm.getString("endpoint.accept.fail"), t);
               }
           }
           state = AcceptorState.ENDED;
       }

setSocketOptions

protected boolean setSocketOptions(SocketChannel socket) {
      // Process the connection
      try {
          //disable blocking, APR style, we are gonna be polling it
          socket.configureBlocking(false);
          Socket sock = socket.socket();
          //给socket设置参数
          socketProperties.setProperties(sock);

          NioChannel channel = nioChannels.pop();
          if (channel == null) {
              SocketBufferHandler bufhandler = new SocketBufferHandler(
                      socketProperties.getAppReadBufSize(),
                      socketProperties.getAppWriteBufSize(),
                      socketProperties.getDirectBuffer());
              if (isSSLEnabled()) {
                  channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
              } else {
                  channel = new NioChannel(socket, bufhandler);
              }
          } else {
              channel.setIOChannel(socket);
              channel.reset();
          }
          getPoller0().register(channel);
      } catch (Throwable t) {
          ExceptionUtils.handleThrowable(t);
          try {
              log.error("",t);
          } catch (Throwable tt) {
              ExceptionUtils.handleThrowable(tt);
          }
          // Tell to close the socket
          return false;
      }
      return true;
  }

Poller注册

public void register(final NioChannel socket) {
           socket.setPoller(this);
           NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
           socket.setSocketWrapper(ka);
           ka.setPoller(this);
           ka.setReadTimeout(getSocketProperties().getSoTimeout());
           ka.setWriteTimeout(getSocketProperties().getSoTimeout());
           ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
           ka.setReadTimeout(getConnectionTimeout());
           ka.setWriteTimeout(getConnectionTimeout());
           PollerEvent r = eventCache.pop();
           ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
           if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
           else r.reset(socket,ka,OP_REGISTER);
           addEvent(r);
       }

PollerEvent

public static class PollerEvent implements Runnable {

      private NioChannel socket;
      private int interestOps;
      private NioSocketWrapper socketWrapper;

      public PollerEvent(NioChannel ch, NioSocketWrapper w, int intOps) {
          reset(ch, w, intOps);
      }

      public void reset(NioChannel ch, NioSocketWrapper w, int intOps) {
          socket = ch;
          interestOps = intOps;
          socketWrapper = w;
      }

      public void reset() {
          reset(null, null, 0);
      }

      @Override
      public void run() {
          if (interestOps == OP_REGISTER) {
              try {
                  socket.getIOChannel().register(
                          socket.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
              } catch (Exception x) {
                  log.error(sm.getString("endpoint.nio.registerFail"), x);
              }
          } else {
              final SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
              try {
                  if (key == null) {
                      // The key was cancelled (e.g. due to socket closure)
                      // and removed from the selector while it was being
                      // processed. Count down the connections at this point
                      // since it won't have been counted down when the socket
                      // closed.
                      socket.socketWrapper.getEndpoint().countDownConnection();
                      ((NioSocketWrapper) socket.socketWrapper).closed = true;
                  } else {
                      final NioSocketWrapper socketWrapper = (NioSocketWrapper) key.attachment();
                      if (socketWrapper != null) {
                          //we are registering the key to start with, reset the fairness counter.
                          int ops = key.interestOps() | interestOps;
                          socketWrapper.interestOps(ops);
                          key.interestOps(ops);
                      } else {
                          socket.getPoller().cancelledKey(key);
                      }
                  }
              } catch (CancelledKeyException ckx) {
                  try {
                      socket.getPoller().cancelledKey(key);
                  } catch (Exception ignore) {}
              }
          }
      }

      @Override
      public String toString() {
          return "Poller event: socket [" + socket + "], socketWrapper [" + socketWrapper +
                  "], interestOps [" + interestOps + "]";
      }
  }

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： 忘忧症
• 本文链接： https://NepenthesZGW.github.io/2020/09/02/web/tomcat/tomcat基础/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！