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

47次阅读

共计 22187 个字符，预计需要花费 56 分钟才能阅读完成。

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

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

一 问题背景

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

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

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

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

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

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

二 技术路线

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

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

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

1 通明代理

通明代理是代理网关的根底，本文采纳 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 选项，批示非平安的代理。因为代理服务自身并没有作为一个中间人，并没有解析出客户端和近程服务端通信的内容。在非通明代理时，须要解决这个问题。

2 非通明代理

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

当传输为 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 做封装实现；
  • 最初，拿到申请后，做相应的解决，最终实现非通明的代理。

3 通明上游代理

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

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

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

4 非通明上游代理

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

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

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

三 设计与实现

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

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

1 代理网关服务端

次要包含

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

初始化代理网关服务

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

2 代理网关客户端

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

代理网关客户端初始化：

 /**
  * 初始化近程连贯
  * @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 只是简略音讯的转发；惟一的不同是，因为代理网关拦挡了第一个申请，此时须要将拦挡的申请，转发到服务端。

四 其余问题

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

1 内存问题

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

2 性能问题

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

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

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

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

五 总结

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

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

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

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

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

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

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

正文完
云原生
发表至： 云原生
2021-11-26
 0
关于云原生:云原生小课堂｜高性能高可用可扩展的MySQL集群如何组建
关于云原生:演讲实录｜云原生时代OAM模型加持下的应用交付与管理实践
关于云原生:邀请函｜SOFA-四周年开源正当时
关于云原生:云效制品仓库-Packages不限容量免费用