日韩性视频-久久久蜜桃-www中文字幕-在线中文字幕av-亚洲欧美一区二区三区四区-撸久久-香蕉视频一区-久久无码精品丰满人妻-国产高潮av-激情福利社-日韩av网址大全-国产精品久久999-日本五十路在线-性欧美在线-久久99精品波多结衣一区-男女午夜免费视频-黑人极品ⅴideos精品欧美棵-人人妻人人澡人人爽精品欧美一区-日韩一区在线看-欧美a级在线免费观看

歡迎訪問 生活随笔!

生活随笔

當(dāng)前位置: 首頁 > 编程资源 > 编程问答 >内容正文

编程问答

代理网关设计与实现(基于NETTY)

發(fā)布時間:2024/8/23 编程问答 35 豆豆
生活随笔 收集整理的這篇文章主要介紹了 代理网关设计与实现(基于NETTY) 小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

簡介:本文重點在代理網(wǎng)關(guān)本身的設(shè)計與實現(xiàn),而非代理資源的管理與維護。

作者 | 新然
來源 | 阿里技術(shù)公眾號

一 問題背景

  • 平臺端購置一批裸代理,來做廣告異地展現(xiàn)審核。從外部購置的代理,使用方式為:
  • 通過給定的HTTP 的 API 提取代理 IP:PORT,返回的結(jié)果會給出代理的有效時長 3~5 分鐘,以及代理所屬地域;
  • 從提取的代理中,選取指定地域,添加認(rèn)證信息,請求獲取結(jié)果;

    本文設(shè)計實現(xiàn)一個通過的代理網(wǎng)關(guān):

  • 管理維護代理資源,并做代理的認(rèn)證鑒權(quán);
  • 對外暴露統(tǒng)一的代理入口,而非動態(tài)變化的代理IP:PORT;
  • 流量過濾及限流,比如:靜態(tài)資源不走代理;
  • 本文重點在代理網(wǎng)關(guān)本身的設(shè)計與實現(xiàn),而非代理資源的管理與維護。

    注:本文包含大量可執(zhí)行的JAVA代碼以解釋代理相關(guān)的原理

    二 技術(shù)路線

    本文的技術(shù)路線。在實現(xiàn)代理網(wǎng)關(guān)之前,首先介紹下代理相關(guān)的原理及如何實現(xiàn)

  • 透明代理;
  • 非透明代理;
  • 透明的上游代理;
  • 非透明的上游代理;
  • 最后,本文要構(gòu)建代理網(wǎng)關(guān),本質(zhì)上就是一個非透明的上游代理,并給出詳細(xì)的設(shè)計與實現(xiàn)。

    1 透明代理

    透明代理是代理網(wǎng)關(guān)的基礎(chǔ),本文采用JAVA原生的NIO進行詳細(xì)介紹。在實現(xiàn)代理網(wǎng)關(guān)時,實際使用的為NETTY框架。原生NIO的實現(xiàn)對理解NETTY的實現(xiàn)有幫助。

    透明代理設(shè)計三個交互方,客戶端、代理服務(wù)、服務(wù)端,其原理是:

  • 代理服務(wù)在收到連接請求時,判定:如果是CONNECT請求,需要回應(yīng)代理連接成功消息到客戶端;
  • CONNECT請求回應(yīng)結(jié)束后,代理服務(wù)需要連接到CONNECT指定的遠(yuǎn)程服務(wù)器,然后直接轉(zhuǎn)發(fā)客戶端和遠(yuǎn)程服務(wù)通信;
  • 代理服務(wù)在收到非CONNECT請求時,需要解析出請求的遠(yuǎn)程服務(wù)器,然后直接轉(zhuǎn)發(fā)客戶端和遠(yuǎn)程服務(wù)通信;
  • 需要注意的點是:

  • 通常HTTPS請求,在通過代理前,會發(fā)送CONNECT請求;連接成功后,會在信道上進行加密通信的握手協(xié)議;因此連接遠(yuǎn)程的時機是在CONNECT請求收到時,因為此后是加密數(shù)據(jù);
  • 透明代理在收到CONNECT請求時,不需要傳遞到遠(yuǎn)程服務(wù)(遠(yuǎn)程服務(wù)不識別此請求);
  • 透明代理在收到非CONNECT請求時,要無條件轉(zhuǎn)發(fā);
  • 完整的透明代理的實現(xiàn)不到約300行代碼,完整摘錄如下:

    @Slf4j public class SimpleTransProxy {public static void main(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單線程調(diào)試while (localServer.isOpen()) {// 此處阻塞等待連接SocketChannel remoteClient = localServer.accept();// 工作線程GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {@SneakyThrows@Overridepublic void run() {// 代理到遠(yuǎn)程SocketChannel remoteServer = new ProxyHandler().proxy(remoteClient);// 透明傳輸reactor.pipe(remoteClient, remoteServer).pipe(remoteServer, remoteClient);}});}} }@Data class ProxyHandler {private String method;private String host;private int port;private SocketChannel remoteServer;private SocketChannel remoteClient;/*** 原始信息*/private List<ByteBuffer> buffers = new ArrayList<>();private StringBuilder stringBuilder = new StringBuilder();/*** 連接到遠(yuǎn)程* @param remoteClient* @return* @throws IOException*/public SocketChannel proxy(SocketChannel remoteClient) throws IOException {this.remoteClient = remoteClient;connect();return this.remoteServer;}public void connect() throws IOException {// 解析METHOD, HOST和PORTbeforeConnected();// 鏈接REMOTE SERVERcreateRemoteServer();// CONNECT請求回應(yīng),其他請求WRITE THROUGHafterConnected();}protected void beforeConnected() throws IOException {// 讀取HEADERreadAllHeader();// 解析HOST和PORTparseRemoteHostAndPort();}/*** 創(chuàng)建遠(yuǎn)程連接* @throws IOException*/protected void createRemoteServer() throws IOException {remoteServer = SocketChannel.open(new InetSocketAddress(host, port));}/*** 連接建立后預(yù)處理* @throws IOException*/protected void afterConnected() throws IOException {// 當(dāng)CONNECT請求時,默認(rèn)寫入200到CLIENTif ("CONNECT".equalsIgnoreCase(method)) {// CONNECT默認(rèn)為443端口,根據(jù)HOST再解析remoteClient.write(ByteBuffer.wrap("HTTP/1.0 200 Connection Established\r\nProxy-agent: nginx\r\n\r\n".getBytes()));} else {writeThrouth();}}protected void writeThrouth() {buffers.forEach(byteBuffer -> {try {remoteServer.write(byteBuffer);} catch (IOException e) {e.printStackTrace();}});}/*** 讀取請求內(nèi)容* @throws IOException*/protected void readAllHeader() 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*/protected void parseRemoteHostAndPort() throws IOException {// 讀取第一批,獲取到METHODmethod = parseRequestMethod(stringBuilder.toString());// 默認(rèn)為80端口,根據(jù)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();}}protected void appendClientBuffer(ByteBuffer clientBuffer) {buffers.add(clientBuffer);stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit()));}protected static ByteBuffer newByteBuffer() {// buffer必須大于7,保證能讀到methodreturn ByteBuffer.allocate(128);}private static String parseRequestMethod(String rawContent) {// create urireturn rawContent.split("\r\n")[0].split(" ")[0];}private static 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 @Data class Reactor {private Selector selector;private volatile boolean finish = false;@SneakyThrowspublic Reactor() {selector = Selector.open();}@SneakyThrowspublic Reactor pipe(SocketChannel from, SocketChannel to) {from.configureBlocking(false);from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));return this;}@SneakyThrowspublic void run() {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();}}@SneakyThrowspublic synchronized void close() {if (finish) {return;}finish = true;if (!selector.isOpen()) {return;}for (SelectionKey key : selector.keys()) {closeChannel(key.channel());key.cancel();}if (selector != null) {selector.close();}}public void cancel(SelectableChannel channel) {SelectionKey key = channel.keyFor(selector);if (Objects.isNull(key)) {return;}key.cancel();}@SneakyThrowspublic void closeChannel(Channel channel) {SocketChannel socketChannel = (SocketChannel)channel;if (socketChannel.isConnected() && socketChannel.isOpen()) {socketChannel.shutdownOutput();socketChannel.shutdownInput();}socketChannel.close();} }@Data @AllArgsConstructor class SocketPipe {private Reactor reactor;private SocketChannel from;private SocketChannel to;@SneakyThrowspublic void pipe() {// 取消監(jiān)聽clearInterestOps();GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {@SneakyThrows@Overridepublic void run() {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 {// 重置監(jiān)聽resetInterestOps();}}});}protected void clearInterestOps() {from.keyFor(reactor.getSelector()).interestOps(0);to.keyFor(reactor.getSelector()).interestOps(0);}protected void resetInterestOps() {from.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);to.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);}private boolean valid(SocketChannel channel) {return channel.isConnected() && channel.isRegistered() && channel.isOpen();} }

    以上,借鑒NETTY:

  • 首先初始化REACTOR線程,然后開啟代理監(jiān)聽,當(dāng)收到代理請求時處理。
  • 代理服務(wù)在收到代理請求時,首先做代理的預(yù)處理,然后又SocketPipe做客戶端和遠(yuǎn)程服務(wù)端雙向轉(zhuǎn)發(fā)。
  • 代理預(yù)處理,首先讀取第一個HTTP請求,解析出METHOD, HOST, PORT。
  • 如果是CONNECT請求,發(fā)送回應(yīng)Connection Established,然后連接遠(yuǎn)程服務(wù)端,并返回SocketChannel
  • 如果是非CONNECT請求,連接遠(yuǎn)程服務(wù)端,寫入原始請求,并返回SocketChannel
  • SocketPipe在客戶端和遠(yuǎn)程服務(wù)端,做雙向的轉(zhuǎn)發(fā);其本身是將客戶端和服務(wù)端的SocketChannel注冊到REACTOR
  • REACTOR在監(jiān)測到READABLE的CHANNEL,派發(fā)給SocketPipe做雙向轉(zhuǎn)發(fā)。
  • 測試

    代理的測試比較簡單,指向代碼后,代理服務(wù)監(jiān)聽8006端口,此時:

    curl -x 'localhost:8006'?http://httpbin.org/get測試HTTP請求

    curl -x 'localhost:8006'?https://httpbin.org/get測試HTTPS請求

    注意,此時代理服務(wù)代理了HTTPS請求,但是并不需要-k選項,指示非安全的代理。因為代理服務(wù)本身并沒有作為一個中間人,并沒有解析出客戶端和遠(yuǎn)程服務(wù)端通信的內(nèi)容。在非透明代理時,需要解決這個問題。

    2 非透明代理

    非透明代理,需要解析出客戶端和遠(yuǎn)程服務(wù)端傳輸?shù)膬?nèi)容,并做相應(yīng)的處理。

    當(dāng)傳輸為HTTP協(xié)議時,SocketPipe傳輸?shù)臄?shù)據(jù)即為明文的數(shù)據(jù),可以攔截后直接做處理。

    當(dāng)傳輸為HTTPS協(xié)議時,SocketPipe傳輸?shù)挠行?shù)據(jù)為加密數(shù)據(jù),并不能透明處理。
    另外,無論是傳輸?shù)腍TTP協(xié)議還是HTTPS協(xié)議,SocketPipe讀到的都為非完整的數(shù)據(jù),需要做聚批的處理。

  • SocketPipe聚批問題,可以采用類似BufferedInputStream對InputStream做Decorate的模式來實現(xiàn),相對比較簡單;詳細(xì)可以參考NETTY的HttpObjectAggregator;
  • HTTPS原始請求和結(jié)果數(shù)據(jù)的加密和解密的處理,需要實現(xiàn)的NIO的SOCKET CHANNEL;
  • SslSocketChannel封裝原理

    考慮到目前JDK自帶的NIO的SocketChannel并不支持SSL;已有的SSLSocket是阻塞的OIO。如圖:

    可以看出

  • 每次入站數(shù)據(jù)和出站數(shù)據(jù)都需要 SSL SESSION 做握手;
  • 入站數(shù)據(jù)做解密,出站數(shù)據(jù)做加密;
  • 握手,數(shù)據(jù)加密和數(shù)據(jù)解密是統(tǒng)一的一套狀態(tài)機;
  • 以下,代碼實現(xiàn) SslSocketChannel

    public class SslSocketChannel {/*** 握手加解密需要的四個存儲*/protected ByteBuffer myAppData; // 明文protected ByteBuffer myNetData; // 密文protected ByteBuffer peerAppData; // 明文protected ByteBuffer peerNetData; // 密文/*** 握手加解密過程中用到的異步執(zhí)行器*/protected ExecutorService executor = Executors.newSingleThreadExecutor();/*** 原NIO 的 CHANNEL*/protected SocketChannel socketChannel;/*** SSL 引擎*/protected SSLEngine engine;public SslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode) throws Exception {// 原始的NIO SOCKETthis.socketChannel = socketChannel;// 初始化BUFFERSSLSession 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* 實現(xiàn)的 SSL 的握手協(xié)議* @return* @throws IOException*/protected boolean doHandshake() 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()) {return false;}try {engine.closeInbound();} catch (SSLException e) {log.debug("收到END OF STREAM,關(guān)閉連接.", 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()) {return false;} else {engine.closeOutbound();handshakeStatus = engine.getHandshakeStatus();break;}default:throw new IllegalStateException("無效的握手狀態(tài): " + 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:throw new SSLException("加密后消息內(nèi)容為空,報錯");case CLOSED:try {myNetData.flip();while (myNetData.hasRemaining()) {socketChannel.write(myNetData);}peerNetData.clear();} catch (Exception e) {handshakeStatus = engine.getHandshakeStatus();}break;default:throw new IllegalStateException("無效的握手狀態(tài): " + 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:throw new IllegalStateException("無效的握手狀態(tài): " + handshakeStatus);}}return true;}/*** 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html* 實現(xiàn)的 SSL 的傳輸讀取協(xié)議* @param consumer* @throws IOException*/public void read(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("收到遠(yuǎn)程的返回結(jié)果消息為:" + 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("收到遠(yuǎn)程連接關(guān)閉消息.");closeConnection();return;default:throw new IllegalStateException("無效的握手狀態(tài): " + result.getStatus());}}} else if (bytesRead < 0) {log.debug("收到END OF STREAM,關(guān)閉連接.");handleEndOfStream();}}public void write(String message) throws IOException {write(ByteBuffer.wrap(message.getBytes()));}/*** 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html* 實現(xiàn)的 SSL 的傳輸寫入?yún)f(xié)議* @param message* @throws IOException*/public void write(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("寫入遠(yuǎn)程的消息為: {}", message);break;case BUFFER_OVERFLOW:myNetData = enlargePacketBuffer(engine, myNetData);break;case BUFFER_UNDERFLOW:throw new SSLException("加密后消息內(nèi)容為空.");case CLOSED:closeConnection();return;default:throw new IllegalStateException("無效的握手狀態(tài): " + result.getStatus());}}}/*** 關(guān)閉連接* @throws IOException*/public void closeConnection() throws IOException {engine.closeOutbound();doHandshake();socketChannel.close();executor.shutdown();}/*** END OF STREAM(-1)默認(rèn)是關(guān)閉連接* @throws IOException*/protected void handleEndOfStream() throws IOException {try {engine.closeInbound();} catch (Exception e) {log.error("END OF STREAM 關(guān)閉失敗.", e);}closeConnection();}}

    以上:

  • 基于 SSL 協(xié)議,實現(xiàn)統(tǒng)一的握手動作;
  • 分別實現(xiàn)讀取的解密,和寫入的加密方法;
  • 將 SslSocketChannel 實現(xiàn)為 SocketChannel的Decorator;
  • SslSocketChannel測試服務(wù)端

    基于以上封裝,簡單測試服務(wù)端如下

    @Slf4j public class NioSslServer {public static void main(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;public NioSslServer(String hostAddress, int port) throws Exception {// 初始化SSL Contextcontext = serverSSLContext();// 注冊監(jiān)聽器selector = SelectorProvider.provider().openSelector();ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);}public void start() 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);} else if (key.isReadable()) {((SslSocketChannel)key.attachment()).read(buf->{});// 直接回應(yīng)一個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();}}}}private void accept(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("握手失敗,關(guān)閉連接.");}} }

    以上:

  • 由于是NIO,簡單的測試需要用到NIO的基礎(chǔ)組件Selector進行測試;
  • 首先初始化ServerSocketChannel,監(jiān)聽8006端口;
  • 接收到請求后,將SocketChannel封裝為SslSocketChannel,注冊到Selector
  • 接收到數(shù)據(jù)后,通過SslSocketChannel做read和write;
  • SslSocketChannel測試客戶端

    基于以上服務(wù)端封裝,簡單測試客戶端如下

    @Slf4j public class NioSslClient {public static void main(String[] args) throws Exception {NioSslClient sslClient = new NioSslClient("httpbin.org", 443);sslClient.connect();// 請求 'https://httpbin.org/get'}private String remoteAddress;private int port;private SSLEngine engine;private SocketChannel socketChannel;private SSLContext context;/*** 需要遠(yuǎn)程的HOST和PORT* @param remoteAddress* @param port* @throws Exception*/public NioSslClient(String remoteAddress, int port) throws Exception {this.remoteAddress = remoteAddress;this.port = port;context = clientSSLContext();engine = context.createSSLEngine(remoteAddress, port);engine.setUseClientMode(true);}public boolean connect() throws Exception {socketChannel = SocketChannel.open();socketChannel.configureBlocking(false);socketChannel.connect(new InetSocketAddress(remoteAddress, port));while (!socketChannel.finishConnect()) {// 通過REACTOR,不會出現(xiàn)等待情況//log.debug("連接中..");}SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true);sslSocketChannel.doHandshake();// 握手完成后,開啟SELECTORSelector 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");// 讀取結(jié)果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();return true;}}}} }

    以上:

  • 客戶端的封裝測試,是為了驗證封裝 SSL 協(xié)議雙向都是OK的,
  • 在后文的非透明上游代理中,會同時使用 SslSocketChannel做服務(wù)端和客戶端
  • 以上封裝與服務(wù)端封裝類似,不同的是初始化 SocketChannel,做connect而非bind
  • 總結(jié)

    以上:

  • 非透明代理需要拿到完整的請求數(shù)據(jù),可以通過 Decorator模式,聚批實現(xiàn);
  • 非透明代理需要拿到解密后的HTTPS請求數(shù)據(jù),可以通過SslSocketChannel對原始的SocketChannel做封裝實現(xiàn);
  • 最后,拿到請求后,做相應(yīng)的處理,最終實現(xiàn)非透明的代理。
  • 3 透明上游代理

    透明上游代理相比透明代理要簡單,區(qū)別是

  • 透明代理需要響應(yīng) CONNECT請求,透明上游代理不需要,直接轉(zhuǎn)發(fā)即可;
  • 透明代理需要解析CONNECT請求中的HOST和PORT,并連接服務(wù)端;透明上游代理只需要連接下游代理的IP:PORT,直接轉(zhuǎn)發(fā)請求即可;
  • 透明的上游代理,只是一個簡單的SocketChannel管道;確定下游的代理服務(wù)端,連接轉(zhuǎn)發(fā)請求;
  • 只需要對透明代理做以上簡單的修改,即可實現(xiàn)透明的上游代理。

    4 非透明上游代理

    非透明的上游代理,相比非透明的代理要復(fù)雜一些

    以上,分為四個組件:客戶端,代理服務(wù)(ServerHandler),代理服務(wù)(ClientHandler),服務(wù)端

  • 如果是HTTP的請求,數(shù)據(jù)直接通過 客戶端<->ServerHandler<->ClientHandler<->服務(wù)端,代理網(wǎng)關(guān)只需要做簡單的請求聚批,就可以應(yīng)用相應(yīng)的管理策略;
  • 如果是HTTPS請求,代理作為客戶端和服務(wù)端的中間人,只能拿到加密的數(shù)據(jù);因此,代理網(wǎng)關(guān)需要作為HTTPS的服務(wù)方與客戶端通信;然后作為HTTPS的客戶端與服務(wù)端通信;
  • 代理作為HTTPS服務(wù)方時,需要考慮到其本身是個非透明的代理,需要實現(xiàn)非透明代理相關(guān)的協(xié)議;
  • 代理作為HTTPS客戶端時,需要考慮到其下游是個透明的代理,真正的服務(wù)方是客戶端請求的服務(wù)方;
  • 三 設(shè)計與實現(xiàn)

    本文需要構(gòu)建的是非透明上游代理,以下采用NETTY框架給出詳細(xì)的設(shè)計實現(xiàn)。上文將統(tǒng)一代理網(wǎng)關(guān)分為兩大部分,ServerHandler和ClientHandler,以下

  • 介紹代理網(wǎng)關(guān)服務(wù)端相關(guān)實現(xiàn);
  • 介紹代理網(wǎng)關(guān)客戶端相關(guān)實現(xiàn);
  • 1 代理網(wǎng)關(guān)服務(wù)端

    主要包括

  • 初始化代理網(wǎng)關(guān)服務(wù)端
  • 初始化服務(wù)端處理器
  • 服務(wù)端協(xié)議升級與處理
  • 初始化代理網(wǎng)關(guān)服務(wù)

    public void start() {HookedExecutors.newSingleThreadExecutor().submit(() ->{log.info("開始啟動代理服務(wù)器,監(jiān)聽端口:{}", 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("代理服務(wù)器被中斷.", e);Thread.currentThread().interrupt();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}});}

    代理網(wǎng)關(guān)初始化相對簡單,

  • bossGroup線程組,負(fù)責(zé)接收請求
  • workerGroup線程組,負(fù)責(zé)處理接收的請求數(shù)據(jù),具體處理邏輯封裝在ServerChannelInitializer中。
  • 代理網(wǎng)關(guān)服務(wù)的請求處理器在 ServerChannelInitializer中定義為

    @Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new HttpRequestDecoder()).addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize())).addLast(new ServerChannelHandler(auditProxyConfig));}

    首先解析HTTP請求,然后做聚批的處理,最后ServerChannelHandler實現(xiàn)代理網(wǎng)關(guān)協(xié)議;

    代理網(wǎng)關(guān)協(xié)議:

  • 判定是否是CONNECT請求,如果是,會存儲CONNECT請求;暫停讀取,發(fā)送代理成功的響應(yīng),并在回應(yīng)成功后,升級協(xié)議;
  • 升級引擎,本質(zhì)上是采用SslSocketChannel對原SocketChannel做透明的封裝;
  • 最后根據(jù)CONNECT請求連接遠(yuǎn)程服務(wù)端;
  • 詳細(xì)實現(xiàn)為:

    @Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {FullHttpRequest request = (FullHttpRequest)msg;try {if (isConnectRequest(request)) {// CONNECT 請求,存儲待處理saveConnectRequest(ctx, request);// 禁止讀取ctx.channel().config().setAutoRead(false);// 發(fā)送回應(yīng)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);}// 連接遠(yuǎn)程connectRemote(ctx, request);}} finally {ctx.fireChannelRead(msg);}}

    2 代理網(wǎng)關(guān)客戶端

    代理網(wǎng)關(guān)服務(wù)端需要連接遠(yuǎn)程服務(wù),進入代理網(wǎng)關(guān)客戶端部分。

    代理網(wǎng)關(guān)客戶端初始化:

    /*** 初始化遠(yuǎn)程連接* @param ctx* @param httpRequest*/protected void connectRemote(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)));// 動態(tài)連接代理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); }

    以上:

  • 復(fù)用代理網(wǎng)關(guān)服務(wù)端的workerGroup線程組;
  • 請求和結(jié)果的處理封裝在ClientChannelInitializer;
  • 連接的遠(yuǎn)程服務(wù)端的HOST和PORT在服務(wù)端收到的請求中可以解析到。
  • 代理網(wǎng)關(guān)客戶端的處理器的初始化邏輯:

    @Overrideprotected void initChannel(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));}

    以上:

  • 如果下游是代理,那么會采用HttpProxyHandler,經(jīng)由下游代理與遠(yuǎn)程服務(wù)端通信;
  • 如果當(dāng)前需要升級為SSL協(xié)議,會對SocketChannel做透明的封裝,實現(xiàn)SSL通信。
  • 最后,ClientChannelHandler只是簡單消息的轉(zhuǎn)發(fā);唯一的不同是,由于代理網(wǎng)關(guān)攔截了第一個請求,此時需要將攔截的請求,轉(zhuǎn)發(fā)到服務(wù)端。
  • 四 其他問題

    代理網(wǎng)關(guān)實現(xiàn)可能面臨的問題:

    1 內(nèi)存問題

    代理通常面臨的問題是OOM。本文在實現(xiàn)代理網(wǎng)關(guān)時保證內(nèi)存中緩存時當(dāng)前正在處理的HTTP/HTTPS請求體。內(nèi)存使用的上限理論上為實時處理的請求數(shù)量*請求體的平均大小,HTTP/HTTPS的請求結(jié)果,直接使用堆外內(nèi)存,零拷貝轉(zhuǎn)發(fā)。

    2 性能問題

    性能問題不應(yīng)提早考慮。本文使用NETTY框架實現(xiàn)的代理網(wǎng)關(guān),內(nèi)部大量使用堆外內(nèi)存,零拷貝轉(zhuǎn)發(fā),避免了性能問題。

    代理網(wǎng)關(guān)一期上線后曾面臨一個長連接導(dǎo)致的性能問題,

  • CLIENT和SERVER建立TCP長連接后(比如,TCP心跳檢測),通常要么是CLIENT關(guān)閉TCP連接,或者是SERVER關(guān)閉;
  • 如果雙方長時間占用TCP連接資源而不關(guān)閉,就會導(dǎo)致SOCKET資源泄漏;現(xiàn)象是:CPU資源爆滿,處理空閑連接;新連接無法建立;
  • 使用IdleStateHandler定時監(jiān)控空閑的TCP連接,強制關(guān)閉;解決了該問題。

    五 總結(jié)

    本文聚焦于統(tǒng)一代理網(wǎng)關(guān)的核心,詳細(xì)介紹了代理相關(guān)的技術(shù)原理。

    代理網(wǎng)關(guān)的管理部分,可以在ServerHandler部分維護,也可以在ClientHandler部分維護;

  • ServerHandler可以攔截轉(zhuǎn)換請求
  • ClientHanlder可控制請求的出口
  • 注:本文使用Netty的零拷貝;存儲請求以解析處理;但并未實現(xiàn)對RESPONSE的處理;也就是RESPONSE是直接通過網(wǎng)關(guān),此方面避免了常見的代理實現(xiàn),內(nèi)存泄漏OOM相關(guān)問題;

    最后,本文實現(xiàn)代理網(wǎng)關(guān)后,針對代理的資源和流經(jīng)代理網(wǎng)關(guān)的請求做了相應(yīng)的控制,主要包括:

  • 當(dāng)遇到靜態(tài)資源的請求時,代理網(wǎng)關(guān)會直接請求遠(yuǎn)程服務(wù)端,不會通過下游代理
  • 當(dāng)請求HEADER中包含地域標(biāo)識時,代理網(wǎng)關(guān)會盡力保證請求打入指定的地域代理,經(jīng)由地域代理訪問遠(yuǎn)程服務(wù)端
  • 原文鏈接
    本文為阿里云原創(chuàng)內(nèi)容,未經(jīng)允許不得轉(zhuǎn)載。?

    總結(jié)

    以上是生活随笔為你收集整理的代理网关设计与实现(基于NETTY)的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。

    如果覺得生活随笔網(wǎng)站內(nèi)容還不錯,歡迎將生活随笔推薦給好友。