代理閘道器設計與實現（基於NETTY）

2025-04-20 04:21 阿里技術

一  問題背景

平臺端購置一批裸代理，來做廣告異地展現稽核。從外部購置的代理，使用方式為：
  1. 透過給定的HTTP 的 API 提取代理 IP:PORT，返回的結果會給出代理的有效時長 3~5 分鐘，以及代理所屬地域；
  2. 從提取的代理中，選取指定地域，新增認證資訊，請求獲取結果；
本文設計實現一個透過的代理閘道器：
  1. 管理維護代理資源，並做代理的認證鑑權；
  2. 對外暴露統一的代理入口，而非動態變化的代理IP:PORT；
  3. 流量過濾及限流，比如：靜態資源不走代理；
本文重點在代理閘道器本身的設計與實現，而非代理資源的管理與維護。
注：本文包含大量可執行的JAVA程式碼以解釋代理相關的原理

二  技術路線

本文的技術路線。在實現代理閘道器之前，首先介紹下代理相關的原理及如何實現
  1. 透明代理；
  2. 非透明代理；
  3. 透明的上游代理；
  4. 非透明的上游代理；
最後，本文要構建代理閘道器，本質上就是一個非透明的上游代理，並給出詳細的設計與實現。

1  透明代理

透明代理是代理閘道器的基礎，本文采用JAVA原生的NIO進行詳細介紹。在實現代理閘道器時，實際使用的為NETTY框架。原生NIO的實現對理解NETTY的實現有幫助。
透明代理設計三個互動方，客戶端、代理服務、服務端，其原理是：
  1. 代理服務在收到連線請求時，判定：如果是CONNECT請求，需要回應代理連線成功訊息到客戶端；
  2. CONNECT請求回應結束後，代理服務需要連線到CONNECT指定的遠端伺服器，然後直接轉發客戶端和遠端服務通訊；
  3. 代理服務在收到非CONNECT請求時，需要解析出請求的遠端伺服器，然後直接轉發客戶端和遠端服務通訊；
需要注意的點是：
  1. 通常HTTPS請求，在透過代理前，會發送CONNECT請求；連線成功後，會在通道上進行加密通訊的握手協議；因此連線遠端的時機是在CONNECT請求收到時，因為此後是加密資料；
  2. 透明代理在收到CONNECT請求時，不需要傳遞到遠端服務（遠端服務不識別此請求）；
  3. 透明代理在收到非CONNECT請求時，要無條件轉發；
完整的透明代理的實現不到約300行程式碼，完整摘錄如下：
@Slf4jpublicclassSimpleTransProxy{publicstaticvoidmain(String[] args)throws IOException {int port = 8006; ServerSocketChannel localServer = ServerSocketChannel.open(); localServer.bind(new InetSocketAddress(port)); Reactor reactor = new Reactor();// REACTOR執行緒 GlobalThreadPool.REACTOR_EXECUTOR.submit(reactor::run);// WORKER單執行緒除錯while (localServer.isOpen()) {// 此處阻塞等待連線 SocketChannel remoteClient = localServer.accept();// 工作執行緒 GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {@SneakyThrows@Overridepublicvoidrun(){// 代理到遠端 SocketChannel remoteServer = new ProxyHandler().proxy(remoteClient);// 透明傳輸 reactor.pipe(remoteClient, remoteServer) .pipe(remoteServer, remoteClient); } }); } }}@DataclassProxyHandler{private String method;private String host;privateint port;private SocketChannel remoteServer;private SocketChannel remoteClient;/** * 原始資訊 */private List<ByteBuffer> buffers = new ArrayList<>();private StringBuilder stringBuilder = new StringBuilder();/** * 連線到遠端 * @param remoteClient * @return * @throws IOException */public SocketChannel proxy(SocketChannel remoteClient)throws IOException {this.remoteClient = remoteClient; connect();returnthis.remoteServer; }publicvoidconnect()throws IOException {// 解析METHOD, HOST和PORT beforeConnected();// 連結REMOTE SERVER createRemoteServer();// CONNECT請求回應，其他請求WRITE THROUGH afterConnected(); }protectedvoidbeforeConnected()throws IOException {// 讀取HEADER readAllHeader();// 解析HOST和PORT parseRemoteHostAndPort(); }/** * 建立遠端連線 * @throws IOException */protectedvoidcreateRemoteServer()throws IOException { remoteServer = SocketChannel.open(new InetSocketAddress(host, port)); }/** * 連線建立後預處理 * @throws IOException */protectedvoidafterConnected()throws IOException {// 當CONNECT請求時，預設寫入200到CLIENTif ("CONNECT".equalsIgnoreCase(method)) {// CONNECT預設為443埠，根據HOST再解析 remoteClient.write(ByteBuffer.wrap("HTTP/1.0 200 Connection Established\r\nProxy-agent: nginx\r\n\r\n".getBytes())); } else { writeThrouth(); } }protectedvoidwriteThrouth(){ buffers.forEach(byteBuffer -> {try { remoteServer.write(byteBuffer); } catch (IOException e) { e.printStackTrace(); } }); }/** * 讀取請求內容 * @throws IOException */protectedvoidreadAllHeader()throws IOException {while (true) { ByteBuffer clientBuffer = newByteBuffer();int read = remoteClient.read(clientBuffer); clientBuffer.flip(); appendClientBuffer(clientBuffer);if (read < clientBuffer.capacity()) {break; } } }/** * 解析出HOST和PROT * @throws IOException */protectedvoidparseRemoteHostAndPort()throws IOException {// 讀取第一批，獲取到METHOD method = parseRequestMethod(stringBuilder.toString());// 預設為80埠，根據HOST再解析 port = 80;if ("CONNECT".equalsIgnoreCase(method)) { port = 443; }this.host = parseHost(stringBuilder.toString()); URI remoteServerURI = URI.create(host); host = remoteServerURI.getHost();if (remoteServerURI.getPort() > 0) { port = remoteServerURI.getPort(); } }protectedvoidappendClientBuffer(ByteBuffer clientBuffer){ buffers.add(clientBuffer); stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit())); }protectedstatic ByteBuffer newByteBuffer(){// buffer必須大於7，保證能讀到methodreturn ByteBuffer.allocate(128); }privatestatic String parseRequestMethod(String rawContent){// create urireturn rawContent.split("\r\n")[0].split(" ")[0]; }privatestatic String parseHost(String rawContent){ String[] headers = rawContent.split("\r\n"); String host = "host:";for (String header : headers) {if (header.length() > host.length()) { String key = header.substring(0, host.length()); String value = header.substring(host.length()).trim();if (host.equalsIgnoreCase(key)) {if (!value.startsWith("http://") && !value.startsWith("https://")) { value = "http://" + value; }return value; } } }return""; }}@Slf4j@DataclassReactor{private Selector selector;privatevolatileboolean finish = false;@SneakyThrowspublicReactor(){ selector = Selector.open(); }@SneakyThrowspublic Reactor pipe(SocketChannel from, SocketChannel to){ from.configureBlocking(false); from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));returnthis; }@SneakyThrowspublicvoidrun(){try {while (!finish) {if (selector.selectNow() > 0) { Iterator<SelectionKey> it = selector.selectedKeys().iterator();while (it.hasNext()) { SelectionKey selectionKey = it.next();if (selectionKey.isValid() && selectionKey.isReadable()) { ((SocketPipe) selectionKey.attachment()).pipe(); } it.remove(); } } } } finally { close(); } }@SneakyThrowspublicsynchronizedvoidclose(){if (finish) {return; } finish = true;if (!selector.isOpen()) {return; }for (SelectionKey key : selector.keys()) { closeChannel(key.channel()); key.cancel(); }if (selector != null) { selector.close(); } }publicvoidcancel(SelectableChannel channel){ SelectionKey key = channel.keyFor(selector);if (Objects.isNull(key)) {return; } key.cancel(); }@SneakyThrowspublicvoidcloseChannel(Channel channel){ SocketChannel socketChannel = (SocketChannel)channel;if (socketChannel.isConnected() && socketChannel.isOpen()) { socketChannel.shutdownOutput(); socketChannel.shutdownInput(); } socketChannel.close(); }}@Data@AllArgsConstructorclassSocketPipe{private Reactor reactor;private SocketChannel from;private SocketChannel to;@SneakyThrowspublicvoidpipe(){// 取消監聽 clearInterestOps(); GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {@SneakyThrows@Overridepublicvoidrun(){int totalBytesRead = 0; ByteBuffer byteBuffer = ByteBuffer.allocate(1024);while (valid(from) && valid(to)) { byteBuffer.clear();int bytesRead = from.read(byteBuffer); totalBytesRead = totalBytesRead + bytesRead; byteBuffer.flip(); to.write(byteBuffer);if (bytesRead < byteBuffer.capacity()) {break; } }if (totalBytesRead < 0) { reactor.closeChannel(from); reactor.cancel(from); } else {// 重置監聽 resetInterestOps(); } } }); }protectedvoidclearInterestOps(){ from.keyFor(reactor.getSelector()).interestOps(0); to.keyFor(reactor.getSelector()).interestOps(0); }protectedvoidresetInterestOps(){ from.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ); to.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ); }privatebooleanvalid(SocketChannel channel){return channel.isConnected() && channel.isRegistered() && channel.isOpen(); }}
以上，借鑑NETTY：
  1. 首先初始化REACTOR執行緒，然後開啟代理監聽，當收到代理請求時處理。
  2. 代理服務在收到代理請求時，首先做代理的預處理，然後又SocketPipe做客戶端和遠端服務端雙向轉發。
  3. 代理預處理，首先讀取第一個HTTP請求，解析出METHOD, HOST, PORT。
  4. 如果是CONNECT請求，傳送回應Connection Established，然後連線遠端服務端，並返回SocketChannel
  5. 如果是非CONNECT請求，連線遠端服務端，寫入原始請求，並返回SocketChannel
  6. SocketPipe在客戶端和遠端服務端，做雙向的轉發；其本身是將客戶端和服務端的SocketChannel註冊到REACTOR
  7. REACTOR在監測到READABLE的CHANNEL，派發給SocketPipe做雙向轉發。
測試
代理的測試比較簡單，指向程式碼後，代理服務監聽8006埠，此時：
curl -x 'localhost:8006' http://httpbin.org/get測試HTTP請求
curl -x 'localhost:8006' https://httpbin.org/get測試HTTPS請求
注意，此時代理服務代理了HTTPS請求，但是並不需要-k選項，指示非安全的代理。因為代理服務本身並沒有作為一箇中間人，並沒有解析出客戶端和遠端服務端通訊的內容。在非透明代理時，需要解決這個問題。

2  非透明代理

非透明代理，需要解析出客戶端和遠端服務端傳輸的內容，並做相應的處理。

當傳輸為HTTP協議時，SocketPipe傳輸的資料即為明文的資料，可以攔截後直接做處理。

當傳輸為HTTPS協議時，SocketPipe

傳輸的有效資料為加密資料，並不能透明處理。

另外，無論是傳輸的

HTTP協議還是HTTPS協議，SocketPipe讀到的都為非完整的資料，需要做聚批的處理。

  1. SocketPipe聚批問題，可以採用類似BufferedInputStream對InputStream做Decorate的模式來實現，相對比較簡單；詳細可以參考NETTY的HttpObjectAggregator；
  2. HTTPS原始請求和結果資料的加密和解密的處理，需要實現的NIO的SOCKET CHANNEL；

SslSocketChannel封裝原理

考慮到目前JDK自帶的NIO的SocketChannel並不支援SSL；已有的SSLSocket是阻塞的OIO。如圖：
可以看出
  1. 每次入站資料和出站資料都需要 SSL SESSION 做握手；
  2. 入站資料做解密，出站資料做加密；
  3. 握手，資料加密和資料解密是統一的一套狀態機；
以下，程式碼實現 SslSocketChannel
publicclassSslSocketChannel{/** * 握手加解密需要的四個儲存 */protected ByteBuffer myAppData; // 明文protected ByteBuffer myNetData; // 密文protected ByteBuffer peerAppData; // 明文protected ByteBuffer peerNetData; // 密文/** * 握手加解密過程中用到的非同步執行器 */protected ExecutorService executor = Executors.newSingleThreadExecutor();/** * 原NIO 的 CHANNEL */protected SocketChannel socketChannel;/** * SSL 引擎 */protected SSLEngine engine;publicSslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode)throws Exception {// 原始的NIO SOCKETthis.socketChannel = socketChannel;// 初始化BUFFER SSLSession dummySession = context.createSSLEngine().getSession(); myAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize()); myNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize()); peerAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize()); peerNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize()); dummySession.invalidate(); engine = context.createSSLEngine(); engine.setUseClientMode(clientMode); engine.beginHandshake(); }/** * 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html * 實現的 SSL 的握手協議 * @return * @throws IOException */protectedbooleandoHandshake()throws IOException { SSLEngineResult result; HandshakeStatus handshakeStatus;int appBufferSize = engine.getSession().getApplicationBufferSize(); ByteBuffer myAppData = ByteBuffer.allocate(appBufferSize); ByteBuffer peerAppData = ByteBuffer.allocate(appBufferSize); myNetData.clear(); peerNetData.clear(); handshakeStatus = engine.getHandshakeStatus();while (handshakeStatus != HandshakeStatus.FINISHED && handshakeStatus != HandshakeStatus.NOT_HANDSHAKING) {switch (handshakeStatus) {case NEED_UNWRAP:if (socketChannel.read(peerNetData) < 0) {if (engine.isInboundDone() && engine.isOutboundDone()) {returnfalse; }try { engine.closeInbound(); } catch (SSLException e) { log.debug("收到END OF STREAM，關閉連線.", e); } engine.closeOutbound(); handshakeStatus = engine.getHandshakeStatus();break; } peerNetData.flip();try { result = engine.unwrap(peerNetData, peerAppData); peerNetData.compact(); handshakeStatus = result.getHandshakeStatus(); } catch (SSLException sslException) { engine.closeOutbound(); handshakeStatus = engine.getHandshakeStatus();break; }switch (result.getStatus()) {case OK:break;case BUFFER_OVERFLOW: peerAppData = enlargeApplicationBuffer(engine, peerAppData);break;case BUFFER_UNDERFLOW: peerNetData = handleBufferUnderflow(engine, peerNetData);break;case CLOSED:if (engine.isOutboundDone()) {returnfalse; } else { engine.closeOutbound(); handshakeStatus = engine.getHandshakeStatus();break; }default:thrownew IllegalStateException("無效的握手狀態: " + result.getStatus()); }break;case NEED_WRAP: myNetData.clear();try { result = engine.wrap(myAppData, myNetData); handshakeStatus = result.getHandshakeStatus(); } catch (SSLException sslException) { engine.closeOutbound(); handshakeStatus = engine.getHandshakeStatus();break; }switch (result.getStatus()) {case OK : myNetData.flip();while (myNetData.hasRemaining()) { socketChannel.write(myNetData); }break;case BUFFER_OVERFLOW: myNetData = enlargePacketBuffer(engine, myNetData);break;case BUFFER_UNDERFLOW:thrownew SSLException("加密後訊息內容為空，報錯");case CLOSED:try { myNetData.flip();while (myNetData.hasRemaining()) { socketChannel.write(myNetData); } peerNetData.clear(); } catch (Exception e) { handshakeStatus = engine.getHandshakeStatus(); }break;default:thrownew IllegalStateException("無效的握手狀態: " + result.getStatus()); }break;case NEED_TASK: Runnable task;while ((task = engine.getDelegatedTask()) != null) { executor.execute(task); } handshakeStatus = engine.getHandshakeStatus();break;case FINISHED:break;case NOT_HANDSHAKING:break;default:thrownew IllegalStateException("無效的握手狀態: " + handshakeStatus); } }returntrue; }/** * 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html * 實現的 SSL 的傳輸讀取協議 * @param consumer * @throws IOException */publicvoidread(Consumer<ByteBuffer> consumer)throws IOException {// BUFFER初始化 peerNetData.clear();int bytesRead = socketChannel.read(peerNetData);if (bytesRead > 0) { peerNetData.flip();while (peerNetData.hasRemaining()) { peerAppData.clear(); SSLEngineResult result = engine.unwrap(peerNetData, peerAppData);switch (result.getStatus()) {case OK: log.debug("收到遠端的返回結果訊息為：" + new String(peerAppData.array(), 0, peerAppData.position())); consumer.accept(peerAppData); peerAppData.flip();break;case BUFFER_OVERFLOW: peerAppData = enlargeApplicationBuffer(engine, peerAppData);break;case BUFFER_UNDERFLOW: peerNetData = handleBufferUnderflow(engine, peerNetData);break;case CLOSED: log.debug("收到遠端連線關閉訊息."); closeConnection();return;default:thrownew IllegalStateException("無效的握手狀態: " + result.getStatus()); } } } elseif (bytesRead < 0) { log.debug("收到END OF STREAM，關閉連線."); handleEndOfStream(); } }publicvoidwrite(String message)throws IOException { write(ByteBuffer.wrap(message.getBytes())); }/** * 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html * 實現的 SSL 的傳輸寫入協議 * @param message * @throws IOException */publicvoidwrite(ByteBuffer message)throws IOException { myAppData.clear(); myAppData.put(message); myAppData.flip();while (myAppData.hasRemaining()) { myNetData.clear(); SSLEngineResult result = engine.wrap(myAppData, myNetData);switch (result.getStatus()) {case OK: myNetData.flip();while (myNetData.hasRemaining()) { socketChannel.write(myNetData); } log.debug("寫入遠端的訊息為: {}", message);break;case BUFFER_OVERFLOW: myNetData = enlargePacketBuffer(engine, myNetData);break;case BUFFER_UNDERFLOW:thrownew SSLException("加密後訊息內容為空.");case CLOSED: closeConnection();return;default:thrownew IllegalStateException("無效的握手狀態: " + result.getStatus()); } } }/** * 關閉連線 * @throws IOException */publicvoidcloseConnection()throws IOException { engine.closeOutbound(); doHandshake(); socketChannel.close(); executor.shutdown(); }/** * END OF STREAM(-1)預設是關閉連線 * @throws IOException */protectedvoidhandleEndOfStream()throws IOException {try { engine.closeInbound(); } catch (Exception e) { log.error("END OF STREAM 關閉失敗.", e); } closeConnection(); }}
以上：
  1. 基於 SSL 協議，實現統一的握手動作；
  2. 分別實現讀取的解密，和寫入的加密方法；
  3. 將 SslSocketChannel 實現為 SocketChannel的Decorator；

SslSocketChannel測試服務端

基於以上封裝，簡單測試服務端如下
@Slf4jpublicclassNioSslServer {publicstaticvoidmain(String[] args) throws Exception { NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006); sslServer.start();// 使用 curl -vv -k 'https://localhost:8006' 連線 }private SSLContext context;private Selector selector;publicNioSslServer(String hostAddress, int port) throws Exception {// 初始化SSL Context context = serverSSLContext();// 註冊監聽器 selector = SelectorProvider.provider().openSelector(); ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port)); serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); }publicvoidstart() throws Exception {log.debug("等待連線中.");while (true) { selector.select(); Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();while (selectedKeys.hasNext()) { SelectionKey key = selectedKeys.next(); selectedKeys.remove();if (!key.isValid()) {continue; }if (key.isAcceptable()) { accept(key); } elseif (key.isReadable()) { ((SslSocketChannel)key.attachment()).read(buf->{});// 直接回應一個OK ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n"); ((SslSocketChannel)key.attachment()).closeConnection(); } } } }privatevoidaccept(SelectionKey key) throws Exception {log.debug("接收新的請求."); SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept(); socketChannel.configureBlocking(false); SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);if (sslSocketChannel.doHandshake()) { socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel); } else { socketChannel.close();log.debug("握手失敗，關閉連線."); } }}
以上：
  1. 由於是NIO，簡單的測試需要用到NIO的基礎元件Selector進行測試；
  2. 首先初始化ServerSocketChannel，監聽8006埠；
  3. 接收到請求後，將SocketChannel封裝為SslSocketChannel，註冊到Selector
  4. 接收到資料後，透過SslSocketChannel做read和write；

SslSocketChannel測試客戶端

基於以上服務端封裝，簡單測試客戶端如下
@Slf4jpublicclassNioSslClient{publicstaticvoidmain(String[] args)throws Exception { NioSslClient sslClient = new NioSslClient("httpbin.org", 443); sslClient.connect();// 請求 'https://httpbin.org/get' }private String remoteAddress;privateint port;private SSLEngine engine;private SocketChannel socketChannel;private SSLContext context;/** * 需要遠端的HOST和PORT * @param remoteAddress * @param port * @throws Exception */publicNioSslClient(String remoteAddress, int port)throws Exception {this.remoteAddress = remoteAddress;this.port = port; context = clientSSLContext(); engine = context.createSSLEngine(remoteAddress, port); engine.setUseClientMode(true); }publicbooleanconnect()throws Exception { socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(remoteAddress, port));while (!socketChannel.finishConnect()) {// 透過REACTOR，不會出現等待情況//log.debug("連線中.."); } SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true); sslSocketChannel.doHandshake();// 握手完成後，開啟SELECTOR Selector selector = SelectorProvider.provider().openSelector(); socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);// 寫入請求 sslSocketChannel.write("GET /get HTTP/1.1\r\n" + "Host: httpbin.org:443\r\n" + "User-Agent: curl/7.62.0\r\n" + "Accept: */*\r\n" + "\r\n");// 讀取結果while (true) { selector.select(); Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();while (selectedKeys.hasNext()) { SelectionKey key = selectedKeys.next(); selectedKeys.remove();if (key.isValid() && key.isReadable()) { ((SslSocketChannel)key.attachment()).read(buf->{ log.info("{}", new String(buf.array(), 0, buf.position())); }); ((SslSocketChannel)key.attachment()).closeConnection();returntrue; } } } }}
以上：
  1. 客戶端的封裝測試，是為了驗證封裝 SSL 協議雙向都是OK的，
  2. 在後文的非透明上游代理中，會同時使用 SslSocketChannel做服務端和客戶端
  3. 以上封裝與服務端封裝類似，不同的是初始化 SocketChannel，做connect而非bind

總結

以上：
  1. 非透明代理需要拿到完整的請求資料，可以透過 Decorator模式，聚批實現；
  2. 非透明代理需要拿到解密後的HTTPS請求資料，可以透過SslSocketChannel對原始的SocketChannel做封裝實現；
  3. 最後，拿到請求後，做相應的處理，最終實現非透明的代理。

3  透明上游代理

透明上游代理相比透明代理要簡單，區別是
  1. 透明代理需要響應 CONNECT請求，透明上游代理不需要，直接轉發即可；
  2. 透明代理需要解析CONNECT請求中的HOST和PORT，並連線服務端；透明上游代理只需要連線下游代理的IP:PORT，直接轉發請求即可；
  3. 透明的上游代理，只是一個簡單的SocketChannel管道；確定下游的代理服務端，連線轉發請求；
只需要對透明代理做以上簡單的修改，即可實現透明的上游代理。

4  非透明上游代理

非透明的上游代理，相比非透明的代理要複雜一些
以上，分為四個元件：客戶端，代理服務（ServerHandler），代理服務（ClientHandler），服務端
  1. 如果是HTTP的請求，資料直接透過 客戶端<->ServerHandler<->ClientHandler<->服務端，代理閘道器只需要做簡單的請求聚批，就可以應用相應的管理策略；
  2. 如果是HTTPS請求，代理作為客戶端和服務端的中間人，只能拿到加密的資料；因此，代理閘道器需要作為HTTPS的服務方與客戶端通訊；然後作為HTTPS的客戶端與服務端通訊；
  3. 代理作為HTTPS服務方時，需要考慮到其本身是個非透明的代理，需要實現非透明代理相關的協議；
  4. 代理作為HTTPS客戶端時，需要考慮到其下游是個透明的代理，真正的服務方是客戶端請求的服務方；

三  設計與實現

本文需要構建的是非透明上游代理，以下采用NETTY框架給出詳細的設計實現。上文將統一代理閘道器分為兩大部分，ServerHandler和ClientHandler，以下
  1. 介紹代理閘道器服務端相關實現；
  2. 介紹代理閘道器客戶端相關實現；

1  代理閘道器服務端

主要包括
  1. 初始化代理閘道器服務端
  2. 初始化服務端處理器
  3. 服務端協議升級與處理
初始化代理閘道器服務
publicvoidstart() { HookedExecutors.newSingleThreadExecutor().submit(() ->{ log.info("開始啟動代理伺服器，監聽埠：{}", auditProxyConfig.getProxyServerPort()); EventLoopGroup bossGroup = new NioEventLoopGroup(auditProxyConfig.getBossThreadCount()); EventLoopGroup workerGroup = new NioEventLoopGroup(auditProxyConfig.getWorkThreadCount());try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.DEBUG)) .childHandler(new ServerChannelInitializer(auditProxyConfig)) .bind(auditProxyConfig.getProxyServerPort()).sync().channel().closeFuture().sync(); } catch (InterruptedException e) { log.error("代理伺服器被中斷.", e); Thread.currentThread().interrupt(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }); }
代理閘道器初始化相對簡單，
  1. bossGroup執行緒組，負責接收請求
  2. workerGroup執行緒組，負責處理接收的請求資料，具體處理邏輯封裝在ServerChannelInitializer中。
代理閘道器服務的請求處理器在 ServerChannelInitializer中定義為
@OverrideprotectedvoidinitChannel(SocketChannel ch)throws Exception { ch.pipeline() .addLast(new HttpRequestDecoder()) .addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize())) .addLast(new ServerChannelHandler(auditProxyConfig)); }
首先解析HTTP請求，然後做聚批的處理，最後ServerChannelHandler實現代理閘道器協議；
代理閘道器協議：
  1. 判定是否是CONNECT請求，如果是，會儲存CONNECT請求；暫停讀取，傳送代理成功的響應，並在回應成功後，升級協議；
  2. 升級引擎，本質上是採用SslSocketChannel對原SocketChannel做透明的封裝；
  3. 最後根據CONNECT請求連線遠端服務端；
詳細實現為：
@OverridepublicvoidchannelRead(ChannelHandlerContext ctx, Object msg)throws Exception { FullHttpRequest request = (FullHttpRequest)msg;try {if (isConnectRequest(request)) {// CONNECT 請求，儲存待處理 saveConnectRequest(ctx, request);// 禁止讀取 ctx.channel().config().setAutoRead(false);// 傳送回應 connectionEstablished(ctx, ctx.newPromise().addListener(future -> {if (future.isSuccess()) {// 升級if (isSslRequest(request) && !isUpgraded(ctx)) { upgrade(ctx); }// 開放訊息讀取 ctx.channel().config().setAutoRead(true); ctx.read(); } })); } else {// 其他請求，判定是否已升級if (!isUpgraded(ctx)) {// 升級引擎 upgrade(ctx); }// 連線遠端 connectRemote(ctx, request); } } finally { ctx.fireChannelRead(msg); } }

2  代理閘道器客戶端

代理閘道器服務端需要連線遠端服務，進入代理閘道器客戶端部分。
代理閘道器客戶端初始化：
/** * 初始化遠端連線 * @param ctx * @param httpRequest */protectedvoidconnectRemote(ChannelHandlerContext ctx, FullHttpRequest httpRequest) { Bootstrap b = new Bootstrap(); b.group(ctx.channel().eventLoop()) // use the same EventLoop .channel(ctx.channel().getClass()) .handler(new ClientChannelInitializer(auditProxyConfig, ctx, safeCopy(httpRequest)));// 動態連線代理 FullHttpRequest originRequest = ctx.channel().attr(CONNECT_REQUEST).get();if (originRequest == null) { originRequest = httpRequest; } ChannelFuture cf = b.connect(new InetSocketAddress(calculateHost(originRequest), calculatePort(originRequest))); Channel cch = cf.channel(); ctx.channel().attr(CLIENT_CHANNEL).set(cch); }
以上：
  1. 複用代理閘道器服務端的workerGroup執行緒組；
  2. 請求和結果的處理封裝在ClientChannelInitializer；
  3. 連線的遠端服務端的HOST和PORT在服務端收到的請求中可以解析到。
代理閘道器客戶端的處理器的初始化邏輯：
@OverrideprotectedvoidinitChannel(SocketChannel ch)throws Exception { SocketAddress socketAddress = calculateProxy();if (!Objects.isNull(socketAddress)) { ch.pipeline().addLast(new HttpProxyHandler(calculateProxy(), auditProxyConfig.getUserName(), auditProxyConfig .getPassword())); }if (isSslRequest()) { String host = host();int port = port();if (StringUtils.isNoneBlank(host) && port > 0) { ch.pipeline().addLast(new SslHandler(sslEngine(host, port))); } } ch.pipeline().addLast(new ClientChannelHandler(clientContext, httpRequest)); }
以上：
  1. 如果下游是代理，那麼會採用HttpProxyHandler，經由下游代理與遠端服務端通訊；
  2. 如果當前需要升級為SSL協議，會對SocketChannel做透明的封裝，實現SSL通訊。
  3. 最後，ClientChannelHandler只是簡單訊息的轉發；唯一的不同是，由於代理閘道器攔截了第一個請求，此時需要將攔截的請求，轉發到服務端。

四  其他問題

代理閘道器實現可能面臨的問題：

1  記憶體問題

代理通常面臨的問題是OOM。本文在實現代理閘道器時保證記憶體中快取時當前正在處理的HTTP/HTTPS請求體。記憶體使用的上限理論上為即時處理的請求數量*請求體的平均大小，HTTP/HTTPS的請求結果，直接使用堆外記憶體，零複製轉發。

2  效能問題

效能問題不應提早考慮。本文使用NETTY框架實現的代理閘道器，內部大量使用堆外記憶體，零複製轉發，避免了效能問題。
代理閘道器一期上線後曾面臨一個長連線導致的效能問題，
  1. CLIENT和SERVER建立TCP長連線後（比如，TCP心跳檢測），通常要麼是CLIENT關閉TCP連線，或者是SERVER關閉；
  2. 如果雙方長時間佔用TCP連線資源而不關閉，就會導致SOCKET資源洩漏；現象是：CPU資源爆滿，處理空閒連線；新連線無法建立；
使用IdleStateHandler定時監控空閒的TCP連線，強制關閉；解決了該問題。

五  總結

本文聚焦於統一代理閘道器的核心，詳細介紹了代理相關的技術原理。
代理閘道器的管理部分，可以在ServerHandler部分維護，也可以在ClientHandler部分維護；
  1. ServerHandler可以攔截轉換請求
  2. ClientHanlder可控制請求的出口
注：本文使用Netty的零複製；儲存請求以解析處理；但並未實現對RESPONSE的處理；也就是RESPONSE是直接透過閘道器，此方面避免了常見的代理實現，記憶體洩漏OOM相關問題；
最後，本文實現代理閘道器後，針對代理的資源和流經代理閘道器的請求做了相應的控制，主要包括：
  1. 當遇到靜態資源的請求時，代理閘道器會直接請求遠端服務端，不會透過下游代理
  2. 當請求HEADER中包含地域標識時，代理閘道器會盡力保證請求打入指定的地域代理，經由地域代理訪問遠端服務端
本文參考https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html實現SslSocketChannel，以透明處理HTTP和HTTPS協議。

基於ELK+Flink日誌全觀測最佳實踐

點選閱讀原文檢視詳情！

關鍵詞
資料
服務端
記憶體
問題
ByteBuffer.allocate


相關文章