关于云原生:代理网关设计与实现基于NETTY

简介：本文重点在代理网关自身的设计与实现，而非代理资源的治理与保护。

作者 | 新然
起源 | 阿里技术公众号

平台端购买一批裸代理，来做广告异地展示审核。从内部购买的代理，应用形式为：
通过给定的 HTTP 的 API 提取代理 IP:PORT，返回的后果会给出代理的无效时长 3~5 分钟，以及代理所属地区；

从提取的代理中，选取指定地区，增加认证信息，申请获取后果；

本文设计实现一个通过的代理网关：

治理保护代理资源，并做代理的认证鉴权；
对外裸露对立的代理入口，而非动态变化的代理 IP:PORT；
流量过滤及限流，比方：动态资源不走代理；

本文重点在代理网关自身的设计与实现，而非代理资源的治理与保护。

注：本文蕴含大量可执行的 JAVA 代码以解释代理相干的原理

本文的技术路线。在实现代理网关之前，首先介绍下代理相干的原理及如何实现

通明代理；
非通明代理；
通明的上游代理；
非通明的上游代理；

最初，本文要构建代理网关，实质上就是一个非通明的上游代理，并给出具体的设计与实现。

通明代理是代理网关的根底，本文采纳 JAVA 原生的 NIO 进行具体介绍。在实现代理网关时，理论应用的为 NETTY 框架。原生 NIO 的实现对了解 NETTY 的实现有帮忙。

通明代理设计三个交互方，客户端、代理服务、服务端，其原理是：

代理服务在收到连贯申请时，断定：如果是 CONNECT 申请，须要回应代理连贯胜利音讯到客户端；
CONNECT 申请回应完结后，代理服务须要连贯到 CONNECT 指定的近程服务器，而后间接转发客户端和近程服务通信；
代理服务在收到非 CONNECT 申请时，须要解析出申请的近程服务器，而后间接转发客户端和近程服务通信；

须要留神的点是：

通常 HTTPS 申请，在通过代理前，会发送 CONNECT 申请；连贯胜利后，会在信道上进行加密通信的握手协定；因而连贯近程的机会是在 CONNECT 申请收到时，因为尔后是加密数据；
通明代理在收到 CONNECT 申请时，不须要传递到近程服务（近程服务不辨认此申请）；
通明代理在收到非 CONNECT 申请时，要无条件转发；

残缺的通明代理的实现不到约 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 单线程调试
        while (localServer.isOpen()) {
            // 此处阻塞期待连贯
            SocketChannel remoteClient = localServer.accept();

            // 工作线程
            GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    // 代理到近程
                    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();

    /**
     * 连贯到近程
     * @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 和 PORT
        beforeConnected();

        // 链接 REMOTE SERVER
        createRemoteServer();

        // CONNECT 申请回应，其余申请 WRITE THROUGH
        afterConnected();}

    protected void beforeConnected() throws IOException {
        // 读取 HEADER
        readAllHeader();

        // 解析 HOST 和 PORT
        parseRemoteHostAndPort();}

    /**
     * 创立近程连贯
     * @throws IOException
     */
    protected void createRemoteServer() throws IOException {remoteServer = SocketChannel.open(new InetSocketAddress(host, port));
    }

    /**
     * 连贯建设后预处理
     * @throws IOException
     */
    protected void afterConnected() throws IOException {
        // 当 CONNECT 申请时，默认写入 200 到 CLIENT
        if ("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();
        }
    }

    protected void writeThrouth() {
        buffers.forEach(byteBuffer -> {
            try {remoteServer.write(byteBuffer);
            } catch (IOException e) {e.printStackTrace();
            }
        });
    }

    /**
     * 读取申请内容
     * @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 {
        // 读取第一批，获取到 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();
        }
    }

    protected void appendClientBuffer(ByteBuffer clientBuffer) {buffers.add(clientBuffer);
        stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit()));
    }

    protected static ByteBuffer newByteBuffer() {
        // buffer 必须大于 7，保障能读到 method
        return ByteBuffer.allocate(128);
    }

    private static String parseRequestMethod(String rawContent) {
        // create uri
        return 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;

    @SneakyThrows
    public Reactor() {selector = Selector.open();
    }

    @SneakyThrows
    public Reactor pipe(SocketChannel from, SocketChannel to) {from.configureBlocking(false);
        from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));
        return this;
    }

    @SneakyThrows
    public 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();
        }
    }

    @SneakyThrows
    public 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();}

    @SneakyThrows
    public 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;

    @SneakyThrows
    public void pipe() {
        // 勾销监听
        clearInterestOps();

        GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {
            @SneakyThrows
            @Override
            public 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 {
                    // 重置监听
                    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 线程，而后开启代理监听，当收到代理申请时解决。
代理服务在收到代理申请时，首先做代理的预处理，而后又 SocketPipe 做客户端和近程服务端双向转发。
代理预处理，首先读取第一个 HTTP 申请，解析出 METHOD, HOST, PORT。
如果是 CONNECT 申请，发送回应 Connection Established，而后连贯近程服务端，并返回 SocketChannel
如果是非 CONNECT 申请，连贯近程服务端，写入原始申请，并返回 SocketChannel
SocketPipe 在客户端和近程服务端，做双向的转发；其自身是将客户端和服务端的 SocketChannel 注册到 REACTOR
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 选项，批示非平安的代理。因为代理服务自身并没有作为一个中间人，并没有解析出客户端和近程服务端通信的内容。在非通明代理时，须要解决这个问题。

非通明代理，须要解析出客户端和近程服务端传输的内容，并做相应的解决。

当传输为 HTTP 协定时，SocketPipe 传输的数据即为明文的数据，能够拦挡后间接做解决。

当传输为 HTTPS 协定时，SocketPipe 传输的无效数据为加密数据，并不能通明解决。

另外，无论是传输的 HTTP 协定还是 HTTPS 协定，SocketPipe 读到的都为非残缺的数据，须要做聚批的解决。

SocketPipe 聚批问题，能够采纳相似 BufferedInputStream 对 InputStream 做 Decorate 的模式来实现，绝对比较简单；具体能够参考 NETTY 的 HttpObjectAggregator；
HTTPS 原始申请和后果数据的加密和解密的解决，须要实现的 NIO 的 SOCKET CHANNEL；

SslSocketChannel 封装原理

思考到目前 JDK 自带的 NIO 的 SocketChannel 并不反对 SSL；已有的 SSLSocket 是阻塞的 OIO。如图：

能够看出

每次入站数据和出站数据都须要 SSL SESSION 做握手；
入站数据做解密，出站数据做加密；
握手，数据加密和数据解密是对立的一套状态机；

以下，代码实现 SslSocketChannel

public class SslSocketChannel {

    /**
     * 握手加解密须要的四个存储
     */
    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;

    public SslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode) throws Exception {
        // 原始的 NIO SOCKET
        this.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
     */
    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，敞开连贯.", 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("有效的握手状态:" + 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("加密后音讯内容为空，报错");
                        case CLOSED:
                            try {myNetData.flip();
                                while (myNetData.hasRemaining()) {socketChannel.write(myNetData);
                                }
                                peerNetData.clear();} catch (Exception e) {handshakeStatus = engine.getHandshakeStatus();
                            }
                            break;
                        default:
                            throw new 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:
                    throw new IllegalStateException("有效的握手状态:" + handshakeStatus);
            }
        }

        return true;
    }

    /**
     * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
     * 实现的 SSL 的传输读取协定
     * @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("收到近程的返回后果音讯为：" + 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:
                        throw new IllegalStateException("有效的握手状态:" + result.getStatus());
                }
            }
        } else if (bytesRead < 0) {log.debug("收到 END OF STREAM，敞开连贯.");
            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
     * 实现的 SSL 的传输写入协定
     * @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("写入近程的音讯为: {}", message);
                    break;
                case BUFFER_OVERFLOW:
                    myNetData = enlargePacketBuffer(engine, myNetData);
                    break;
                case BUFFER_UNDERFLOW:
                    throw new SSLException("加密后音讯内容为空.");
                case CLOSED:
                    closeConnection();
                    return;
                default:
                    throw new IllegalStateException("有效的握手状态:" + result.getStatus());
            }
        }
    }

    /**
     * 敞开连贯
     * @throws IOException
     */
    public void closeConnection() throws IOException  {engine.closeOutbound();
        doHandshake();
        socketChannel.close();
        executor.shutdown();}

    /**
     * END OF STREAM(-1)默认是敞开连贯
     * @throws IOException
     */
    protected void handleEndOfStream() throws IOException  {
        try {engine.closeInbound();
        } catch (Exception e) {log.error("END OF STREAM 敞开失败.", e);
        }
        closeConnection();}

}

以上：

基于 SSL 协定，实现对立的握手动作；
别离实现读取的解密，和写入的加密办法；
将 SslSocketChannel 实现为 SocketChannel 的 Decorator；

SslSocketChannel 测试服务端

基于以上封装，简略测试服务端如下

@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 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);
    }

    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->{});
                    // 间接回应一个 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("握手失败，敞开连贯.");
        }
    }
}

以上：

因为是 NIO，简略的测试须要用到 NIO 的根底组件 Selector 进行测试；
首先初始化 ServerSocketChannel，监听 8006 端口；
接管到申请后，将 SocketChannel 封装为 SslSocketChannel，注册到 Selector
接管到数据后，通过 SslSocketChannel 做 read 和 write；

SslSocketChannel 测试客户端

基于以上服务端封装，简略测试客户端如下

@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;

    /**
     * 须要近程的 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，不会呈现期待状况
            //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();
                    return true;
                }
            }
        }
    }
}

以上：

客户端的封装测试，是为了验证封装 SSL 协定双向都是 OK 的，
在后文的非通明上游代理中，会同时应用 SslSocketChannel 做服务端和客户端
以上封装与服务端封装相似，不同的是初始化 SocketChannel，做 connect 而非 bind

以上：

非通明代理须要拿到残缺的申请数据，能够通过 Decorator 模式，聚批实现；
非通明代理须要拿到解密后的 HTTPS 申请数据，能够通过 SslSocketChannel 对原始的 SocketChannel 做封装实现；
最初，拿到申请后，做相应的解决，最终实现非通明的代理。

通明上游代理相比通明代理要简略，区别是

通明代理须要响应 CONNECT 申请，通明上游代理不须要，间接转发即可；
通明代理须要解析 CONNECT 申请中的 HOST 和 PORT，并连贯服务端；通明上游代理只须要连贯上游代理的 IP:PORT，间接转发申请即可；
通明的上游代理，只是一个简略的 SocketChannel 管道；确定上游的代理服务端，连贯转发申请；

只须要对通明代理做以上简略的批改，即可实现通明的上游代理。

非通明的上游代理，相比非通明的代理要简单一些

以上，分为四个组件：客户端，代理服务（ServerHandler），代理服务（ClientHandler），服务端

如果是 HTTP 的申请，数据间接通过客户端 <->ServerHandler<->ClientHandler<-> 服务端，代理网关只须要做简略的申请聚批，就能够利用相应的管理策略；
如果是 HTTPS 申请，代理作为客户端和服务端的中间人，只能拿到加密的数据；因而，代理网关须要作为 HTTPS 的服务方与客户端通信；而后作为 HTTPS 的客户端与服务端通信；
代理作为 HTTPS 服务方时，须要思考到其自身是个非通明的代理，须要实现非通明代理相干的协定；
代理作为 HTTPS 客户端时，须要思考到其上游是个通明的代理，真正的服务方是客户端申请的服务方；

本文须要构建的是非通明上游代理，以下采纳 NETTY 框架给出具体的设计实现。上文将对立代理网关分为两大部分，ServerHandler 和 ClientHandler，以下

介绍代理网关服务端相干实现；
介绍代理网关客户端相干实现；

次要包含

初始化代理网关服务端
初始化服务端处理器
服务端协定降级与解决

初始化代理网关服务

public void start() {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();}
    });
}

代理网关初始化绝对简略，

bossGroup 线程组，负责接管申请
workerGroup 线程组，负责解决接管的申请数据，具体解决逻辑封装在 ServerChannelInitializer 中。

代理网关服务的申请处理器在 ServerChannelInitializer 中定义为

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

首先解析 HTTP 申请，而后做聚批的解决，最初 ServerChannelHandler 实现代理网关协定；

代理网关协定：

断定是否是 CONNECT 申请，如果是，会存储 CONNECT 申请；暂停读取，发送代
理胜利的响应，并在回应胜利后，降级协定；
降级引擎，实质上是采纳 SslSocketChannel 对原 SocketChannel 做通明的封装；
最初依据 CONNECT 申请连贯近程服务端；

具体实现为：

 @Override
 public 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);

             // 发送回应
             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);
     }
 }

代理网关服务端须要连贯近程服务，进入代理网关客户端局部。

代理网关客户端初始化：

 /**
  * 初始化近程连贯
  * @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)));

     // 动静连贯代理
     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);    
 }

以上：

复用代理网关服务端的 workerGroup 线程组；
申请和后果的解决封装在 ClientChannelInitializer；
连贯的近程服务端的 HOST 和 PORT 在服务端收到的申请中能够解析到。

代理网关客户端的处理器的初始化逻辑：

@Override
  protected 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，经由上游代理与近程服务端通信；
如果以后须要降级为 SSL 协定，会对 SocketChannel 做通明的封装，实现 SSL 通信。
最初，ClientChannelHandler 只是简略音讯的转发；惟一的不同是，因为代理网关拦挡了第一个申请，此时须要将拦挡的申请，转发到服务端。

代理网关实现可能面临的问题：

代理通常面临的问题是 OOM。本文在实现代理网关时保障内存中缓存时以后正在解决的 HTTP/HTTPS 申请体。内存应用的下限实践上为实时处理的申请数量 * 申请体的均匀大小，HTTP/HTTPS 的申请后果，间接应用堆外内存，零拷贝转发。

性能问题不应提前思考。本文应用 NETTY 框架实现的代理网关，外部大量应用堆外内存，零拷贝转发，防止了性能问题。

代理网关一期上线后曾面临一个长连贯导致的性能问题，

CLIENT 和 SERVER 建设 TCP 长连贯后（比方，TCP 心跳检测），通常要么是 CLIENT 敞开 TCP 连贯，或者是 SERVER 敞开；
如果单方长时间占用 TCP 连贯资源而不敞开，就会导致 SOCKET 资源透露；景象是：CPU 资源爆满，解决闲暇连贯；新连贯无奈建设；

应用 IdleStateHandler 定时监控闲暇的 TCP 连贯，强制敞开；解决了该问题。

本文聚焦于对立代理网关的外围，具体介绍了代理相干的技术原理。

代理网关的治理局部，能够在 ServerHandler 局部保护，也能够在 ClientHandler 局部保护；

ServerHandler 能够拦挡转换申请
ClientHanlder 可管制申请的进口

注：本文应用 Netty 的零拷贝；存储申请以解析解决；但并未实现对 RESPONSE 的解决；也就是 RESPONSE 是间接通过网关，此方面防止了常见的代理实现，内存透露 OOM 相干问题；

最初，本文实现代理网关后，针对代理的资源和流经代理网关的申请做了相应的管制，次要包含：

当遇到动态资源的申请时，代理网关会间接申请近程服务端，不会通过上游代理
当申请 HEADER 中蕴含地区标识时，代理网关会尽力保障申请打入指定的地区代理，经由地区代理拜访近程服务端。

原文链接
本文为阿里云原创内容，未经容许不得转载。

一问题背景

二技术路线

1 通明代理

2 非通明代理

总结

3 通明上游代理

4 非通明上游代理

三设计与实现

1 代理网关服务端

2 代理网关客户端

四其余问题

1 内存问题

2 性能问题

五总结

站内搜索