IT极限技术分享汇Connector的初始化
catalina解析server.xml是通过degister来实现的，degister解析到<Connector标签后做的事情如下代码所见
ConnectorCreateRule
@Override
public void begin(String namespace, String name, Attributes attributes)
throws Exception {
Service svc = (Service)digester.peek();
Executor ex = null;
if ( attributes.getValue("executor")!=null ) {
ex = svc.getExecutor(attributes.getValue("executor"));
}
Connector con = new Connector(attributes.getValue("protocol"));
if (ex != null) {
setExecutor(con, ex);
}
String sslImplementationName = attributes.getValue("sslImplementationName");
if (sslImplementationName != null) {
setSSLImplementationName(con, sslImplementationName);
}
digester.push(con);
}
connector根据标签属性，拿到对应的protocol协议，拿到配在service标签内部的线程池，protocol的名称转化成Connector中ProtocolHandler类型的成员变量, 后续将以Http11NioProtocol来做讲解
public Connector(String protocol) {
setProtocol(protocol);
// Instantiate protocol handler
ProtocolHandler p = null;
try {
Class<?> clazz = Class.forName(protocolHandlerClassName);
// 反射调用ProtocolHandler的构造方法的时候会做后续的初始化
p = (ProtocolHandler) clazz.getConstructor().newInstance();
} catch (Exception e) {
log.error(sm.getString("coyoteConnector.protocolHandlerInstantiationFailed"), e);
} finally {
this.protocolHandler = p;
}ProtocolHandler的构造
ProtocolHandler有其抽象方法，Http11NioProtocol构造方法中的tcp实现由NioEndpoint来做，因此Connector构造起来的时候，对应的ProtocolHanlder、endpoint的关联关系已经关联好
public AbstractProtocol(AbstractEndpoint<S, ?> endpoint) {
this.endpoint = endpoint;
ConnectionHandler<S> cHandler = new ConnectionHandler<>(this);
setHandler(cHandler);
getEndpoint().setHandler(cHandler);
setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
}构造后的init方法
在解析server.xml的基本层次结构，成员变量填充完整后，需要调用生命周期的init方法
org/apache/catalina/startup/Catalina.load getServer().init();
-----> connector.init(); --------------> protocolHandler.init();
----------> endpoint.init(); ------------->
看一下NioEndpoint的init方法
protected void initServerSocket() throws Exception {
if (getUseInheritedChannel()) {
// 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"));
}
} else {
// 绑定服务端的传输层端口
serverSock = ServerSocketChannel.open();
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
serverSock.socket().bind(addr,getAcceptCount());
}
serverSock.configureBlocking(true); //mimic APR behavior
}建立Nio的Acceptor线程和Selector事件线程
org.apache.tomcat.util.net.NioEndpoint#startInternal
public void startInternal() throws Exception {
if (!running) {
running = true;
paused = false;
if (socketProperties.getProcessorCache() != 0) {
processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getProcessorCache());
}
if (socketProperties.getEventCache() != 0) {
eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getEventCache());
}
if (socketProperties.getBufferPool() != 0) {
nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getBufferPool());
}
// Create worker collection
if (getExecutor() == null) {
// 创建Tomcat默认线程池，未手动配置 线程数10 - 200
createExecutor();
}
// 这个是tomcat的连接数限制器
initializeConnectionLatch();
// Start poller thread
poller = new Poller();
Thread pollerThread = new Thread(poller, getName() + "-Poller");
pollerThread.setPriority(threadPriority);
pollerThread.setDaemon(true);
pollerThread.start();
startAcceptorThread();
}
}
我们来看一下Poller的实现和Acctptor的实现
Poller的实现
public class Poller implements Runnable {
private Selector selector;
// PollerEvent private NioSocketWrapper socketWrapper; private int interestOps;
// 向多路复用器注册socket和需要处理的socket时间
private final SynchronizedQueue<PollerEvent> events =
new SynchronizedQueue<>();
.........
public Poller() throws IOException {
this.selector = Selector.open();
}
@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 {
// 默认阻塞1秒钟，监听需要交给线程池的处理任务
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
...
} catch (Throwable x) {
ExceptionUtils.handleThrowable(x);
log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
continue;
}
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();
iterator.remove();
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (socketWrapper != null) {
// 主角， 任务处理都在里面
processKey(sk, socketWrapper);
}
}
// Process timeouts
timeout(keyCount,hasEvents);
}
.............
}
Acctptor的实现
public class Acceptor<U> implements Runnable {
@Override
public void run() {
int errorDelay = 0;
long pauseStart = 0;
try {
// Loop until we receive a shutdown command
while (!stopCalled) {
............
if (stopCalled) {
break;
}
state = AcceptorState.RUNNING;
try {
//if we have reached max connections, wait
endpoint.countUpOrAwaitConnection();
// Endpoint might have been paused while waiting for latch
// If that is the case, don't accept new connections
if (endpoint.isPaused()) {
continue;
}
U socket = null;
try {
// 等待新连接
socket = endpoint.serverSocketAccept();
} catch (Exception ioe) {
// We didn't get a socket
endpoint.countDownConnection();
if (endpoint.isRunning()) {
// 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 (!stopCalled && !endpoint.isPaused()) {
// 由endpoint来处理新连接、把新连接存在map里、向poller注册PollerEndpoint
// 以后读写事件就由Poller交给线程池来管理
if (!endpoint.setSocketOptions(socket)) {
endpoint.closeSocket(socket);
}
} else {
endpoint.destroySocket(socket);
}
}
........
} finally {
stopLatch.countDown();
}
state = AcceptorState.ENDED;
}
}当新连接接入时，会把新连接注册到至底层为Selector的多路复用器上，Tomcat的Connector拥有了接受新连接和处理socket事件的能力
项目实战
依旧使用mytomcat.war， 里面有一个servlet FirstServlet,跟随tomcat一起启动，容器启动，项目的部署暂且不讨论
接下来我们看一下一个简单的Servlet如何在tomcat中来流转
tomcat启动时， Acceptor将阻塞到 socket = endpoint.serverSocketAccept();这一行
在浏览器输入http://localhost:8080/mytomcat/servlet1， Acceptor解除阻塞并且获取到客户端socket
来看一下endpoint.setSocketOptions(socket)做了啥
@Override
protected boolean setSocketOptions(SocketChannel socket) {
.........
NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
channel.reset(socket, newWrapper);
connections.put(socket, newWrapper);
socketWrapper = newWrapper;
// Set socket properties
// Disable blocking, polling will be used
socket.configureBlocking(false);
socketProperties.setProperties(socket.socket());
// 注册新连接
poller.register(socketWrapper);
return true;
...
再看一下Poller的代码 监听了新连接的OP_READ事件
public void register(final NioSocketWrapper socketWrapper) {
socketWrapper.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
PollerEvent pollerEvent = createPollerEvent(socketWrapper, OP_REGISTER);
addEvent(pollerEvent);
}这样，http://localhost:8080/mytomcat/servlet1的请求就会交给Poller处理，详细代码见Poller的run函数
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 {
// 默认阻塞1秒钟，监听需要交给线程池的处理任务
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
...
} catch (Throwable x) {
ExceptionUtils.handleThrowable(x);
log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
continue;
}
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();
iterator.remove();
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (socketWrapper != null) {
// 主角， 任务处理都在里面
processKey(sk, socketWrapper);
}
}
// Process timeouts
timeout(keyCount,hasEvents);
}
.............
}processKey的时候向线程池提交一个SocketProcessor任务
多线程任务的执行
经过一系列变换，connecotr对应的Handler为Http11Nio Protocol，
将协议部分的处理交由Http11Processor
CoyoteAdapter
tomcat并非直接把请求封装为HttpServletRequest对象和HttpServletResponse对象，本身connector支持多种协议，不只使用http协议。所以tomcat存在比较底层的org.apache.coyote.Request， 也存在继承HttpServletRequest的org.apache.catalina.connector.Request，CoyoteAdapter负责做二者的转化并且把转化后的HttpServletRequest对象和HttpServletResponse对象交给tomcat容器的Pipeline。
来看一下CoyoteAdapter的service方法
public void service(org.apache.coyote.Request req, org.apache.coyote.Response res) throws Exception {
Request request = (Request) req.getNote(ADAPTER_NOTES);
Response response = (Response) res.getNote(ADAPTER_NOTES);
if (request == null) {
// Create objects
request = connector.createRequest();
request.setCoyoteRequest(req);
response = connector.createResponse();
response.setCoyoteResponse(res);
// 做两种类型request的转换
request.setResponse(response);
response.setRequest(request);
}
try {
// Parse and set Catalina and configuration specific
// 下面有讲解这个方法
postParseSuccess = postParseRequest(req, request, res, response);
if (postParseSuccess) {
// check valves if we support async
request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported());
// Calling the container
//把HttpServletRequest 和HttpServletRequest对象交给tomcat的职责链Pipeline。
connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
}
........
}
我们来看一下postParseRequest(req, resp)方法
protected boolean postParseRequest(org.apache.coyote.Request req, Request request, org.apache.coyote.Response res,
Response response) throws IOException, ServletException {
// 解析出是否要Https
if (req.scheme().isNull()) {
// Use connector scheme and secure configuration, (defaults to
// "http" and false respectively)
req.scheme().setString(connector.getScheme());
request.setSecure(connector.getSecure());
} else {
// Use processor specified scheme to determine secure state
request.setSecure(req.scheme().equals("https"));
}
// 代理相关
String proxyName = connector.getProxyName();
int proxyPort = connector.getProxyPort();
if (proxyPort != 0) {
req.setServerPort(proxyPort);
} else if (req.getServerPort() == -1) {
// Not explicitly set. Use default ports based on the scheme
if (req.scheme().equals("https")) {
req.setServerPort(443);
} else {
req.setServerPort(80);
}
}
if (proxyName != null) {
req.serverName().setString(proxyName);
}
// 预检方法请求
// Check for ping OPTIONS * request
if (undecodedURI.equals("*")) {
if (req.method().equalsIgnoreCase("OPTIONS")) {
StringBuilder allow = new StringBuilder();
allow.append("GET, HEAD, POST, PUT, DELETE, OPTIONS");
// Trace if allowed
if (connector.getAllowTrace()) {
allow.append(", TRACE");
}
res.setHeader("Allow", allow.toString());
// Access log entry as processing won't reach AccessLogValve
connector.getService().getContainer().logAccess(request, response, 0, true);
return false;
} else {
response.sendError(400, sm.getString("coyoteAdapter.invalidURI"));
}
}
........
while (mapRequired) {
// This will map the the latest version by default
// tomcat可以部署多个应用，找到具体在某个Context、Host、填充到request对象中。
connector.getService().getMapper().map(serverName, decodedURI, version, request.getMappingData());
}
.......
探索一下tomcat的职责链
正式进入Tomcat的职责链中，下面我们来看一下pipeline中有什么
因为已有环境，本文直接给出默认的链装结构
StandardEngineValve—-> AccessLogValve—-> ErrorReportValve—- > StandardHostValve —-> StandardContextValve —–> StandardWrapperValve—> ApplicationFilterChain —后续执行指定的Servlet
排除容器本身的节点StandardEngineValve， StandardHostValve，StandardContextValve三个节点。 AccessLogValve，ErrorReportValve， StandardWrapperValve值得我们关注
AccessLogValve:使用来记录访问请求的基本信息，记录到access.log中等
ErrorReportValve:来处理后面的错误，跳到失败页面等
StandardWrapperValve: 如下详细解说
@Override
public void invoke(Request request, Response response) throws IOException, ServletException {
...........
// 获取对应servlet对应的Wrapper， servlet可能还未加载
StandardWrapper wrapper = (StandardWrapper) getContainer();
Servlet servlet = null;
Context context = (Context) wrapper.getParent();
........
try {
if (!unavailable) {
// 获取真正的servlet， 如果未加载初始化，在加载初始化后返回
servlet = wrapper.allocate();
}
} catch (UnavailableException e) {
container.getLogger().error(sm.getString("standardWrapper.allocateException", wrapper.getName()), e);
checkWrapperAvailable(response, wrapper);
}
.................
// ApplicationFilterChain本身是servlet规范中的过滤器， 持有servelt对象，使用职责链调用完成后，再调用servlet， 至此，tomcat的请求流程到此，后续交给上层应用程序来处理请求
ApplicationFilterChain filterChain = ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);
.........
filterChain.doFilter(request.getRequest(), response.getResponse());
至此，tomcat讲请求处理完后递交给上层应用程序来处理，并返回结果，依次写回
本文到此结束，tomcat的源码还有很多，如容器之间的关系，多类的加载安全性，动态部署等，有兴趣的朋友可以继续研究