tomcat源码阅读之connector

痛定思痛。 2022-01-16 04:37 313阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

# 概述（是什么？） #

Tomcat的实现主要是依靠各种组件的配合使用，而连接器负责处理与客户端的通信。tomcat提供了多种连接器，可以在server.xml配置文件中指定想要使用的连接器，例如以下配置：

<Connector port="8080" protocol="org.apache.coyote.http11.Http11Nio2Protocol"  
            connectionTimeout="20000"
            redirectPort="8443"
            executor="tomcatThreadPool"
            enableLookups="false"
            acceptCount="100"
            maxPostSize="1024"
            compression="on"
            disableUploadTimeout="true"
            compressionMinSize="2048"
            acceptorThreadCount="1"
            compressableMimeType="text/html,text/xml,text/plain,text/css,text/javascript,application/javascript"   
            URIEncoding="utf-8"/>

在tomcat8版本中，主要使用的是coyote包中提供的一些连接器：Http11NioProtocol、Http11AprProtocol等等，今天主要分析tomcat8默认的连接器--Http11NioProtocol。

连接器的主要作用是接收请求，创建request与response，然后将请求交给processor处理，processor可以调用servlet中的处理逻辑，然后将返回的结果写入response，再响应客户端。以上是旧版本（4.0）的基本的逻辑，8.0版本的比这个逻辑要复杂许多，但是本质还是在干着几件事。接下来根据源码详细分析连接器的原理。

# Connector（org.apache.catalina.connector.Connector） #

tomcat源码中connector既是连接器的最外层表示，首先来分析一下她的构造方法：

/**
     * Coyote Protocol handler class name.
     * Defaults to the Coyote HTTP/1.1 protocolHandler.
     * 默认的protocolHandler
     */
    protected String protocolHandlerClassName =
    	"org.apache.coyote.http11.Http11NioProtocol";
    
    public Connector() {
    	this(null);
    }
    
    //Implementation of a Coyote connector
    //实现了coyote连接器
    public Connector(String protocol) {
    	setProtocol(protocol);
    	// Instantiate protocol handler
    	//协议处理器的实现
    	ProtocolHandler p = null;
    	try {
            //通过protocolHandlerClassName名称反射获取到class
    		Class<?> clazz = Class.forName(protocolHandlerClassName);
            //实例化ProtocolHandler
    		p = (ProtocolHandler) clazz.getConstructor().newInstance();
    	} catch (Exception e) {
    		log.error(sm.getString(
    				"coyoteConnector.protocolHandlerInstantiationFailed"), e);
    	} finally {
            //将实例化的handler保存在属性protocolHandler中
    		this.protocolHandler = p;
    	}
    
    	if (Globals.STRICT_SERVLET_COMPLIANCE) {
    		uriCharset = StandardCharsets.ISO_8859_1;
    	} else {
    		uriCharset = StandardCharsets.UTF_8;
    	}
    }

在实例化connector时，如果传送的protocol属性为空，这个时候connector会获取默认的protocolHandlerClassName来实例化protocolHandler。

接着在来分析connector是怎么初始化的，首先需要明确一个概念，tomcat中所有组件的生命周期都是由Lifecycle这个类来控制，connector也实现了这个类，该类的init、start、stop、destroy控制着connector的一生。LifecycleBase是抽象类，实现了Lifecycle的这些方法，她将init方法的核心逻辑封装进了initInternal方法（该方法可由子类实现），同样start方法的核心逻辑在startInternal中，所以，我们只要看一下connector的这两个方法是怎么实现的就知道，连接器是怎么初始化的。代码如下：

protected void initInternal() throws LifecycleException {
    
        //调用父类初始化方法将Connector名称注册到MBeanServer中托管。
    	super.initInternal();
    
    	// Initialize adapter
    	//初始化coyote适配器，在后续处理请求时需要用到
    	adapter = new CoyoteAdapter(this);
    	protocolHandler.setAdapter(adapter);
    
    	// Make sure parseBodyMethodsSet has a default
    	//保证有一个请求方式 默认是post
    	if (null == parseBodyMethodsSet) {
    		setParseBodyMethods(getParseBodyMethods());
    	}
    
    	if (protocolHandler.isAprRequired() && !AprLifecycleListener.isAprAvailable()) {
    		throw new LifecycleException(sm.getString("coyoteConnector.protocolHandlerNoApr",
    				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的初始化，重点
    		protocolHandler.init();
    	} catch (Exception e) {
    		throw new LifecycleException(
    				sm.getString("coyoteConnector.protocolHandlerInitializationFailed"), e);
    	}
    }

从代码中可以看出，初始化方法的核心是protocolHandler的初始化，而protocolHandler的初始化主要是在AbstractProtocol类的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的初始化
    	endpoint.init();
    }

通过代码上的注解可以知道这段代码的核心是endpoint的初始化，而endpoint的初始化在AbstractEndpoint的init方法中实现：

public void init() throws Exception {
    	if (bindOnInit) {
    		//重点 socket服务监听
    		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=ThreadPool,name=\"" + getName() + "\",subType=SocketProperties");
    		socketProperties.setObjectName(socketPropertiesOname);
    		Registry.getRegistry(null, null).registerComponent(socketProperties, socketPropertiesOname, null);
    
    		for (SSLHostConfig sslHostConfig : findSslHostConfigs()) {
    			registerJmx(sslHostConfig);
    		}
    	}
    }

终于看到熟悉的东西了，bind方法，清楚的告诉我们终于开始重要的socket的监听的初始化了。代码如下：

/**
     * Initialize the endpoint.
     * 实例化endpoint
     */
    @Override
    public void bind() throws Exception {
    
    	if (!getUseInheritedChannel()) {
    		//ServerSocketChannel（jdk nio包中的对象）socket服务管道打开
    		serverSock = ServerSocketChannel.open();
    		//socket配置
    		socketProperties.setProperties(serverSock.socket());
    		//socket监听地址+端口，默认监听本地地址+端口
    		InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
    		//ServerSocketChannel绑定端口，并接收连接的数量
    		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"));
    		}
    	}
    	//配置accept为阻塞模式
    	serverSock.configureBlocking(true); //mimic APR behavior
    
    	// Initialize thread count defaults for acceptor, poller
    	//初始化acceptor与poller线程数
    	if (acceptorThreadCount == 0) {
    		// FIXME: Doesn't seem to work that well with multiple accept threads
    		acceptorThreadCount = 1;
    	}
    	if (pollerThreadCount <= 0) {
    		//minimum one poller thread
    		pollerThreadCount = 1;
    	}
    	//处理线程计数器，初始化
    	setStopLatch(new CountDownLatch(pollerThreadCount));
    
    	// Initialize SSL if needed
    	initialiseSsl();
    
    	//开启NioBlockingSelector
    	selectorPool.open();
    }

可以看出bind中完成了，最核心的初始化任务，socket监听服务ip端口绑定，acceptor（请求接收者线程）与poller（轮询处理线程）线程数的确定，stopLatch线程计数器的实例化，selectorPool（NioSelectorPool）的开启。

注意到selectorPool这个池子，其具有两个属性：NioBlockingSelector和Selector（这个是个共享selector，selector是多路复用对象，她可以与内核进行沟通，告知用户线程socket准备好读了），进入NioBlockingSelector类发现，其仍然拥有一个selector与blockpoller（这个selector其实就是selectorPool的共享selector），进入blockpoller发现，还是拥有selector（这个selector其实就是NioBlockingSelector的共享selector）和一个保存事件的queue。如果将poller中的selector作为主，则这个blockpoller为辅助，当poller的主selector获取到准备好的请求，开始处理该请求，处理完成后，需要将响应内容返回给客户端，这个辅助blockpoller就是做这么一件事：轮询内核，是否准备好将response返回给客户端，如果准备好了就将请求响应返回。

poller是在请求（socket）接收后，具体处理之前，存放待处理请求的线程，其线程数一般为计算机的cpu数量，poller线程会轮询（selector会与内核沟通轮询socket状态）待处理的读写请求，如果内核准备好数据了，就开始将内核的数据copy到用户线程，接着开始请求的解析，处理，返回。这里有必要讲解一下I/O多路复用的原理，tomcat的I/O请求使用了该模型。

![962d412f337a600b7f2f9e324509a9a58d8.jpg][]

首先需要解释一下，线程分为好几种状态，这里涉及到用户态和内核态，网络I\\O读写其实本质上是由内核线程在做，那么将数据从内核线程复制到用户线程就是中间的桥梁，这就是selector，连接器接收到请求后，马上会响应（三次握手成功），然后客户端发送请求报文，连接器处理请求时，会阻塞在select方法上，直到轮询到内核线程发出一个标识：接收到了完整的请求可以开始数据的读取了，这个时候，socket准备好读取请求的数据了，数据可以read了，数据read后就可以进行请求的进一步解析。从这个过程可以了解到，单个的用户线程，可以接受多个的tcp请求，这便是多路复用。

init完毕后就开始start了，接下来看一下connector的startInternal方法：

/**
     * Begin processing requests via this Connector.
     *
     * @exception LifecycleException if a fatal startup error occurs
     */
    @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的开始。
    		protocolHandler.start();
    	} catch (Exception e) {
    		throw new LifecycleException(
    				sm.getString("coyoteConnector.protocolHandlerStartFailed"), e);
    	}
    }

同样的套路，connector.start的重点是protocolHandler的start，接下来直接贴代码有些不再详细讲解了。

//abstractProtocol.java
    
    @Override
    public void start() throws Exception {
    	if (getLog().isInfoEnabled()) {
    		getLog().info(sm.getString("abstractProtocolHandler.start", getName()));
    	}
    
        //重点 endpoint的开始
    	endpoint.start();
    
    	// Start async 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();
    }

//abstractendpoint.java
    
    public final void start() throws Exception {
        //检查绑定状态，防止未绑定就开启连接器服务
    	if (bindState == BindState.UNBOUND) {
    		bind();
    		bindState = BindState.BOUND_ON_START;
    	}
        //重点方法
    	startInternal();
    }

/**
     * Start the NIO endpoint, creating acceptor, poller threads.
     * 开始非阻塞endpoint，创建接收者线程，轮询处理线程
     */
    @Override
    public void startInternal() throws Exception {
    
    	if (!running) {
    		running = true;
    		paused = false;
    
    		//Cache for SocketProcessor objects
    		//缓存请求具体处理方法（当poller轮询完毕，数据copy完毕，接着处理请求时的处理方法）
    		processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
    				socketProperties.getProcessorCache());
    		//Cache for poller events
    		//缓存poller的事件（socket请求包装在里面）。
    		eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
    						socketProperties.getEventCache());
    		//Bytebuffer cache, each channel holds a set of buffers (two, except for SSL holds four)
    		//客户端请求的具体实现类。
    		nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
    				socketProperties.getBufferPool());
    
    		// Create worker collection
    		// 创建工作者
    		if ( getExecutor() == null ) {
    			createExecutor();
    		}
    
    		//初始化连接限制 accept有限制最大连接数
    		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);
                //thread.start 其实调用的是poller类的run方法
    			pollerThread.start();
    		}
    
    		//开启accept线程
    		startAcceptorThreads();
    	}
    }
    
    //abstractendpoint.java
    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());
            //同理start() -> run()
    		t.start();
    	}
    }

这里需要重点讲解一下connector的工作，主要由accept、poller、worker来实现如下图：

![d2a014392a6231f8423d3bf9dc83443f4b8.jpg][]

由此图可以清楚的看到，一个请求从产生，接收到流转给具体工作者的整个过程。

转载于:https://my.oschina.net/Spider001/blog/3037774

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/f81e5cc96c0d4ba1bf5d05103359b0ef.png
[962d412f337a600b7f2f9e324509a9a58d8.jpg]: /images/20220114/6b246969a4504070b35dc7628e0c0c52.png
[d2a014392a6231f8423d3bf9dc83443f4b8.jpg]: /images/20220114/d37b04005e734f69a690bd71e51e0650.png