前言
上面一篇主要介绍了Container，下面我们来看下Connector容器。
Connector结构图
Connector结构图
Connector具体是用ProtocolHandler来处理请求的，不同的ProtocolHandler代表着不同的连接类型。ProtocolHandler 包含了三个非常重要的组件：
- Endpoint：用于处理底层Socket的网络连接；
- Processor：用于将Endpoint接受到的Socket封装成Request；
- Adapter：用于将封装好的Request交给Container进行具体处理。
Endpoint的抽象实例AbstractEndpoint里面定义的Acceptor和AsyncTimeout两个内部类和一个Handler接口。Acceptor用于监听请求，AsyncTimeout用于检查异步request的超时，Handler用于处理接收到的Socket，在内部调用Processor进行处理。
Connector
- #initInternal()方法
protected void initInternal() throws LifecycleException {
super.initInternal();
if (protocolHandler == null) {
throw new LifecycleException(
sm.getString("coyoteConnector.protocolHandlerInstantiationFailed"));
}
//1. 初始化适配器，并设置给protocolHandler
adapter = new CoyoteAdapter(this);
protocolHandler.setAdapter(adapter);
if (service != null) {
protocolHandler.setUtilityExecutor(service.getServer().getUtilityExecutor());
}
// 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());
}
}
//2. 初始化protocolHandler
try {
protocolHandler.init();
} catch (Exception e) {
throw new LifecycleException(
sm.getString("coyoteConnector.protocolHandlerInitializationFailed"), e);
}
}
- #startInternal()
protected void startInternal() throws LifecycleException {
// Validate settings before starting
if (getPortWithOffset() < 0) {
throw new LifecycleException(sm.getString(
"coyoteConnector.invalidPort", Integer.valueOf(getPortWithOffset())));
}
//1. 设置生命周期
setState(LifecycleState.STARTING);
//2. 启动protocolHandler
try {
protocolHandler.start();
} catch (Exception e) {
throw new LifecycleException(
sm.getString("coyoteConnector.protocolHandlerStartFailed"), e);
}
}
Connector的初始化和启动主要为了初始化和启动ProtocolHandler，下面去看下ProtocolHandler。
ProtocolHandler
ProtocolHandler是一个接口，其有很多实现类：
image.png
这里我们主要看Http11NioProtocol：
- #init()
public void init() throws Exception {
// 必须先配置升级协议，因为端点init（通过下面的super.init（）触发）使用此列表配置要通告的ALPN协议列表
for (UpgradeProtocol upgradeProtocol : upgradeProtocols) {
configureUpgradeProtocol(upgradeProtocol);
}
super.init();
}
//AbstractProtocol
public void init() throws Exception {
if (getLog().isInfoEnabled()) {
getLog().info(sm.getString("abstractProtocolHandler.init", getName()));
logPortOffset();
}
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);
}
//设置endpoint的name和domain
String endpointName = getName();
endpoint.setName(endpointName.substring(1, endpointName.length()-1));
endpoint.setDomain(domain);
//endpoint初始化
endpoint.init();
}
#init()主要调用了初始化了endpoint，下面会单独看。
- #start()
//AbstractProtocol
public void start() throws Exception {
if (getLog().isInfoEnabled()) {
getLog().info(sm.getString("abstractProtocolHandler.start", getName()));
logPortOffset();
}
//启动endpoint
endpoint.start();
//创建一个60秒一次的定时任务，用来启动异步超时检查
monitorFuture = getUtilityExecutor().scheduleWithFixedDelay(
new Runnable() {
@Override
public void run() {
if (!isPaused()) {
startAsyncTimeout();
}
}
}, 0, 60, TimeUnit.SECONDS);
}
protected void startAsyncTimeout() {
if (asyncTimeoutFuture == null || (asyncTimeoutFuture != null && asyncTimeoutFuture.isDone())) {
if (asyncTimeoutFuture != null && asyncTimeoutFuture.isDone()) {
// There was an error executing the scheduled task, get it and log it
try {
asyncTimeoutFuture.get();
} catch (InterruptedException | ExecutionException e) {
getLog().error(sm.getString("abstractProtocolHandler.asyncTimeoutError"), e);
}
}
asyncTimeoutFuture = getUtilityExecutor().scheduleAtFixedRate(
new Runnable() {
@Override
public void run() {
long now = System.currentTimeMillis();
for (Processor processor : waitingProcessors) {
processor.timeoutAsync(now);
}
}
}, 1, 1, TimeUnit.SECONDS);
}
}
AbstractEndpoint
AbstractEndpoint有一个子类AbstractJsseEndpoint，而AbstractJsseEndpoint有三个子类，这里我们看的是NioEndpoint：
- #init()
//AbstractEndpoint
public final void init() throws Exception {
//bindOnInit控制端点绑定端口。 true ，默认#init()上的端口，并在#destroy()上解除绑定。
//如果设置为false，则端口绑定在#start()上，unbound绑定在#stop()上。
if (bindOnInit) {
//开启服务端socket，并打开初始化共享Selector，启动辅助poller线程
bindWithCleanup();
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);
}
}
}
#init()方法中，主要完成了Socket服务的初始化，#bindWithCleanup()方法在后面详细介绍。
- #start()
//AbstractEndpoint
public final void start() throws Exception {
//1.如果socket服务还未绑定，重新初始化一次
if (bindState == BindState.UNBOUND) {
bindWithCleanup();
bindState = BindState.BOUND_ON_START;
}
//2.模板方法，子类实现
startInternal();
}
- #startInternal()
public void startInternal() throws Exception {
//1. endpoint还未运行，初始化processorCache，eventCache，nioChannels
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());
//2. 创建工作线程池，用于后续执行SocketProcessor线程
if ( getExecutor() == null ) {
createExecutor();
}
//3. 初始化最大连接数计数器
initializeConnectionLatch();
//4. 启动主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();
}
//5. 启动Acceptor线程
startAcceptorThreads();
}
}
主要流程如下：
- 初始化processorCache，eventCache，nioChannels
- 创建工作线程池，用于后续执行SocketProcessor线程
- 初始化最大连接数计数器
- 启动主Poller线程
- 启动Acceptor线程
总结
整个流程下来，Connector也启动完成了，但是存在着很多疑问，比如Connector是如何工作的？为什么要使用辅助poller？后面会有一篇专门的Connector介绍。
