Connector概述
Connector是Jetty中可以直接接受客戶端連接的抽象，一個Connector監聽Jetty服務器的一個端口，所有客戶端的連接請求首先通過該端口，而后由操作系統分配一個新的端口(Socket)與客戶端進行數據通信(先握手，然后建立連接，但是使用不同的端口)。不同的Connector實現可以使用不同的底層結構，如Socket Connector、NIO Connector等，也可以使用不同的協議，如Ssl Connector、AJP Connector，從而在不同的Connector中配置不同的EndPoint和Connection。EndPoint用于連接(Socket)的讀寫數據(參見
深入Jetty源碼之EndPoint)，Connection用于關聯Request、Response、EndPoint、Server，并將解析出來的Request、Response傳遞給在Server中注冊的Handler來處理(參見
深入Jetty源碼之Connection)。在Jetty中Connector的有：SocketConnector、SslSocketConnector、Ajp13SocketConnector、BlockingChannelConnector、SelectChannelConnector、SslSelectChannelConnector、LocalConnector、NestedConnector等，其類圖如下。
Connector類圖
Connector接口
首先Connector實現了LifeCycle接口，在啟動Jetty服務器時，會調用其的start方法，用于初始化Connector內部狀態，并打開Connector以接受客戶端的請求(調用open方法)；而在停止Jetty服務器時會調用其stop方法，以關閉Connector以及內部元件(如Connection等)以及做一些清理工作。因而open、close方法是Connector中用于處理生命周期的方法；對每個Connector都有name字段用于標記該Connector，默認值為：hostname:port；Connector中還有Server的引用，可以從中獲取ThreadPool，并作為Handler的容器被使用(在創建HttpConnection時，Server實例作為構造函數參數傳入，并在handleRequest()方法中將解析出來的Request、Response傳遞給Server注冊的Handler)；Connector還定義了一些用于配置當前Connector的方法，如Buffer Size、Max Idle Time、Low Resource Max Idle Time，以及一些統計信息，如當前Connector總共處理過的請求數、總共處理過的連接數、當前打開的連接數等信息。
Connector的接口定義如下：
public interface Connector
extends LifeCycle {
// Connector名字，默認值hostname:port
String getName();
// 打開當前Connector
void open()
throws IOException;
// 關閉當前Connector
void close()
throws IOException;
// 對Server的引用
void setServer(Server server);
Server getServer();
// 在處理請求消息頭時使用的Buffer大小
int getRequestHeaderSize();
void setRequestHeaderSize(
int size);
// 在處理響應消息頭時使用的Buffer大小
int getResponseHeaderSize();
void setResponseHeaderSize(
int size);
// 在處理請求消息內容時使用的Buffer大小
int getRequestBufferSize();
void setRequestBufferSize(
int requestBufferSize);
// 在處理響應消息內容使用的Buffer大小
int getResponseBufferSize();
void setResponseBufferSize(
int responseBufferSize);
// 在處理請求消息時使用的Buffer工廠
Buffers getRequestBuffers();
// 在處理響應消息時使用的Buffer工廠
Buffers getResponseBuffers();
// User Data Constraint的配置可以是None、Integral、Confidential，對這三種值的解釋：
// None：A value of NONE means that the application does not require any transport guarantees.
// Integral：A value of INTEGRAL means that the application requires the data sent between the client and server to be sent in such a way that it can't be changed in transit.
// Confidential：A value of CONFIDENTIAL means that the application requires the data to be transmitted in a fashion that prevents other entities from observing the contents of the transmission on.
// In most cases, the presence of the INTEGRAL or CONFIDENTIAL flag indicates that the use of SSL is required.參考：這里 // 如果配置了user-data-constraint為Integral或confidential表示所有相應請求都會重定向到使用Integral/Confidential Schema/Port構建的新的URL中。
int getIntegralPort();
String getIntegralScheme();
boolean isIntegral(Request request);
int getConfidentialPort();
String getConfidentialScheme();
boolean isConfidential(Request request);
// 在將HttpConnection交給Server中的Handlers處理前，根據當前Connector，自定義一些EndPoint和Request的配置，如設置EndPoint的MaxIdleTime，Request的timestamp，
// 清除SelectChannelEndPoint中的idleTimestamp，檢查forward頭等。 void customize(EndPoint endpoint, Request request)
throws IOException;
// 主要用于SelectChannelConnector中，重置SelectChannelEndPoint中的idleTimestamp，即重新計時idle的時間。
// 它在每個Request處理結束，EndPoint還未關閉，并且當前連接屬于keep-alive類型的時候被調用。 void persist(EndPoint endpoint)
throws IOException;
// 底層的鏈接實例，如ServerSocket、ServerSocketChannel等。 Object getConnection();
//是否對"X-Forwarded-For"頭進行DNS名字解析 boolean getResolveNames();
// 當前Connector綁定的主機名、端口號等。貌似在Connector的實現中沒有一個默認的主機名。
// 由于端口號可以設置為0，表示由操作系統隨機的分配一個還沒有被使用的端口，因而這里由LocalPort用于存儲Connector實際上綁定的端口號；
// 其中-1表示這個Connector還未開啟，-2表示Connector已經關閉。 String getHost();
void setHost(String hostname);
void setPort(
int port);
int getPort();
int getLocalPort();
// Socket的最大空閑時間，以及在資源比較少(如線程池中的任務數比最大可用線程數要多)的情況下的最大空閑時間，當空閑時間超過這個時間后關閉當前連接(Socket)。 int getMaxIdleTime();
void setMaxIdleTime(
int ms);
int getLowResourceMaxIdleTime();
void setLowResourceMaxIdleTime(
int ms);
// 是否當前Connector處于LowResources狀態，即線程池中的任務數比最大可用線程數要多 public boolean isLowResources();
/* ------------------------以下是一些獲取和當前Connector相關的統計信息------------------------------------ */
// 打開或關閉統計功能 public void setStatsOn(
boolean on);
// 統計功能的開閉狀態 public boolean getStatsOn();
// 重置統計數據，以及統計開始時間 public void statsReset();
// 統計信息的開啟時間戳 public long getStatsOnMs();
// 當前Connector處理的請求數 public int getRequests();
// 當前Connector接收到過的連接數 public int getConnections() ;
// 當前Connector所有當前還處于打開狀態的連接數 public int getConnectionsOpen() ;
// 當前Connector歷史上同時處于打開狀態的最大連接數 public int getConnectionsOpenMax() ;
// 當前Connector所有連接的持續時間總和 public long getConnectionsDurationTotal();
// 當前Connector的最長連接持續時間 public long getConnectionsDurationMax();
// 當前Connector平均連接持續時間 public double getConnectionsDurationMean() ;
// 當前Connector所有連接持續時間的標準偏差 public double getConnectionsDurationStdDev() ;
// 當前Connector的所有連接的平均請求數 public double getConnectionsRequestsMean() ;
// 當前Connector的所有連接的請求數標準偏差 public double getConnectionsRequestsStdDev() ;
// 當前Connector的所有連接的最大請求數 public int getConnectionsRequestsMax();
}
AbstractConnector實現
Jetty中所有的Connector都繼承自AbstractConnector，而它自身繼承自HttpBuffers，HttpBuffers包含了Request、Response的Buffer工廠的創建以及相應Size的配置。AbstractConnector還包含了Name、Server、maxIdleTime以及一些統計信息的引用，用于實現Connector接口中的方法，只是一些Get、Set方法，不詳述。除了Connector接口相關的配置，AbstractConnector還為定義了兩個字段：_acceptQueueSize用于表示ServerSocket
或ServerSocketChannel中最大可等待的請求數，_acceptors用于表示用于調用ServerSocket或ServerSocketChannel中accept方法的線程數，建議這個數字小于或等于可用處理器的2倍。
在AbstractConnector中還定義了Acceptor內部類，它實現了Runnable接口，在其run方法實現中，它將自己的Thread實例賦值給AbstractConnector中的_acceptorThread數組中acceptor號對應的bucket，并更新線程名為：<name> Acceptor<index> <Connector.toString>。然后根據配置的_acceptorPriorityOffset設置當前線程的priority。只要當前Connector處于Running狀態，并且底層鏈接實例不為null，不斷的調用accept方法()。在退出的finally語句快中清理_acceptorThread數組中相應的Bucket值為null，并將線程原來的Name、Priority設置回來。
AbstractConnector實現了doStart()方法，它首先保證Server實例的存在；然后打開當前Connector(調用其open()方法)，并調用父類的doStart方法(這里是HttpBuffers，用于初始化對應的Buffers)；如果沒有自定義的ThreadPool，則從Server中獲取ThreadPool；最后根據acceptors的值創建_acceptorThread數組，將acceptors個Acceptor實例dispatch給ThreadPool。在doStop方法實現中，它首先調用close方法，然后對非Server中的ThreadPool調用其stop方法，再調用父類的doStop方法清理Buffers的引用，最后遍歷_acceptorThread數據，調用每個Thread的interrupt方法。
在各個子類的accept方法實現中，他們在獲取客戶端過來的Socket連接后，都會對該Socket做一些配置，即調用AbstractConnector的configure方法，它首先設置Socket的TCP_NODELAY為true，即禁用Nagle算法(關于禁用的理由可以參考：http://jerrypeng.me/2013/08/mythical-40ms-delay-and-tcp-nodelay/#sec-4-2，簡單的，如果該值為false，則TCP的數據包要么達到TCP Segment Size，要么收到一個Ack，才會發送出去，即有Delay)；然后如果設置了_soLingerTime，則開啟Socket中SO_LINGER選項，否則，關閉該選項(SO_LINGER選項用于控制關閉一個Socket的行為，如果開啟了該選項，則在關閉Socket時會等待_soLingerTime時間，此時如果有數據還未發送完，則會發送這些數據；如果關閉了該選項，則Socket的關閉會立即返回，此時也有可能繼續發送未發送完成的數據，具體參考：http://blog.csdn.net/factor2000/article/details/3929816)。
在ServerSocket和ServerSocketChannel中還有一個SO_REUSEADDR的配置，一般來說當一個端口被釋放后會等待兩分鐘再被使用，此時如果重啟服務器，可能會導致啟動時的綁定錯誤，設置該值可以讓端口釋放后可以立即被使用(具體參考：http://www.cnblogs.com/mydomain/archive/2011/08/23/2150567.html)。在AbstractConnector中可以使用setReuseAddress方法來配置，默認該值設置為true。
AbstractConnector中還實現了customize方法，它在forwarded設置為true的情況下設置相應的attribute：javax.servlet.request.cipher_suite, javax.servlet.request.ssl_session_id，以及Request中對應Host、Server等頭信息。這個邏輯具體含義目前還不是很了解。。。。
最后關于統計數據的更新方法，AbstractConnector定義了如下方法：
protected void connectionOpened(Connection connection)
protected void connectionUpgraded(Connection oldConnection, Connection newConnection)
protected void connectionClosed(Connection connection)
SocketConnector實現
有了AbstractConnector的實現，SocketConnector的實現就變的非常簡單了，它保存了一個EndPoint的Set，表示所有在這個Connector下正在使用的EndPoint，然后是ServerSocket，在open方法中創建，并在getConnection()方法中返回，還有一個localPort字段，當ServerSocket被創建時從ServerSocket實例中獲取，并在getLocalPort()方法中返回。在close方法中關閉ServerSocket，并設置localPort為-2；在accept方法中，調用ServerSocket的accept方法，返回一個Socket，調用configure方法對新創建的Socket做一些基本的配置，然后使用該Socket創建ConnectorEndPoint，并調用其dispatch方法；在customize方法中，在調用AbstractConnector的customize方法的同時還設置ConnectorEndPoint的MaxIdleTime，即設置Socket的SO_TIMEOUT選項，用于配置該Socket的空閑可等待時間；在doStart中會先清理ConnectorEndPoint的集合，而在doStop中會關閉所有還處于打開狀態的ConnectorEndPoint。
SelectChannelConnector實現
SelectChannelConnector內部使用ServerSocketChannel，在open方法中創建ServerSocketChannel，配置其為非blocking模式，并設置localPort值；在accept方法中調用ServerSocket的accept方法獲得一個SocketChannel，配置該Channel為非blocking模式，調用AbstractChannel的configure方法做相應Socket配置，最后將該SocketChannel注冊給ConnectorSelectManager；在doStart方法中，它會初始化ConnectorSelectManager的SelectSets值為acceptors值、MaxIdleTime、LowResourceConnections、LowResourcesMaxIdleTime等值，并啟動該Manager，并dispatch acceptors個線程，不斷的調用Manager的doSelect方法；在close方法中會先stop ConnectorSelectManager，然后關閉ServerSocketChannel，設置localPort為-2；在customize方法中會清除SelectChannelEndPoint的idleTimestamp，重置其MaxIdleTime以及重置Request中的timestamp的值；在persist方法中會重置SelectChannelEndPoint中idleTimestamp的值。
BlockingChannelConnector實現
BlockingChannelConnector實現類似SocketConnector，不同的是它使用ServerSocketChannel，并且其EndPoint為BlockingChannelEndPoint。所不同的是它需要在doStart方法中啟動一個線程不斷的檢查所有還在connections集合中的BlockingChannelEndPoint是否已經超時，每400ms檢查一次，如果超時則關閉該EndPoint。
其他的Connector的實現都比較類似，而SSL相關的Connector需要也只是加入了SSL相關的邏輯，這里不再贅述。
posted on 2014-05-01 18:40
DLevin 閱讀(4686)
評論(0) 編輯 收藏 所屬分類:
Jetty