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

2021 年 11 月 23 日 阿里技术


一  问题背景


平台端购置一批裸代理,来做广告异地展现审核。从外部购置的代理,使用方式为:

  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行代码,完整摘录如下:

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

  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

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

以上:

  1. 基于  SSL  协议,实现统一的握手动作;
  2. 分别实现读取的解密,和写入的加密方法;
  3. 将  SslSocketChannel  实现为  SocketChannel Decorator;


SslSocketChannel测试服务端


基于以上封装,简单测试服务端如下

  
  
    
@Slf4jpublic 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("握手失败,关闭连接."); } }}

以上:

  1. 由于是 NIO ,简单的测试需要用到 NIO 的基础组件 Selector 进行测试;
  2. 首先初始化 ServerSocketChannel ,监听 8006 端口;
  3. 接收到请求后,将 SocketChannel 封装为 SslSocketChannel ,注册到 Selector
  4. 接收到数据后,通过 SslSocketChannel read write


SslSocketChannel测试客户端


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

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

以上:

  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. 服务端协议升级与处理

初始化代理网关服务

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

代理网关初始化相对简单,

  1. bossGroup 线程组,负责接收请求


  2. 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实现代理网关协议;

代理网关协议:

  1. 判定是否是CONNECT请求,如果是,会存储CONNECT请求;暂停读取,发送代理成功的响应,并在回应成功后,升级协议;


  2. 升级引擎,本质上是采用SslSocketChannel对原SocketChannel做透明的封装;


  3. 最后根据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); }

以上:

  1. 复用代理网关服务端的workerGroup线程组;
  2. 请求和结果的处理封装在ClientChannelInitializer;
  3. 连接的远程服务端的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));    }

以上:

  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://doc s.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html 实现   SslSocketChannel ,以透明处理 HTTP HTTPS 协议。



基于ELK+Flink日志全观测最佳实践

点击阅读原文查看详情!

登录查看更多
0

相关内容

深度神经网络FPGA设计进展、实现与展望
专知会员服务
34+阅读 · 2022年3月21日
图对抗防御研究进展
专知会员服务
37+阅读 · 2021年12月13日
专知会员服务
12+阅读 · 2021年9月1日
专知会员服务
45+阅读 · 2021年5月24日
专知会员服务
29+阅读 · 2021年2月21日
【AAAI2021】组合对抗攻击
专知会员服务
47+阅读 · 2021年2月17日
轻量级神经网络架构综述
专知会员服务
95+阅读 · 2020年4月29日
在.NET 6 中如何创建和使用 HTTP 客户端 SDK
如何使用流处理器 Pipy 来创建网络代理
InfoQ
0+阅读 · 2022年2月25日
大规模 Node.js 网关的架构设计与工程实践
Dubbo-go 优雅上下线设计与实践
阿里技术
0+阅读 · 2022年1月27日
HTTP客户端演进之路
InfoQ
0+阅读 · 2021年12月6日
开源微服务编排框架:Netflix Conductor
阿里技术
1+阅读 · 2021年12月2日
WhatsApp 是如何实现端到端加密备份的?
InfoQ
0+阅读 · 2021年11月26日
如何在微服务中设计用户权限策略?
InfoQ
0+阅读 · 2021年11月19日
Spring Cloud Gateway一次请求调用源码解析
阿里技术
0+阅读 · 2021年11月15日
设计和实现一款轻量级的爬虫框架
架构文摘
13+阅读 · 2018年1月17日
国家自然科学基金
0+阅读 · 2014年12月31日
国家自然科学基金
0+阅读 · 2013年12月31日
国家自然科学基金
1+阅读 · 2013年12月31日
国家自然科学基金
0+阅读 · 2013年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
1+阅读 · 2012年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
3+阅读 · 2012年12月31日
国家自然科学基金
1+阅读 · 2009年12月31日
Arxiv
0+阅读 · 2022年4月20日
Arxiv
10+阅读 · 2018年4月19日
Arxiv
26+阅读 · 2018年2月27日
VIP会员
相关VIP内容
深度神经网络FPGA设计进展、实现与展望
专知会员服务
34+阅读 · 2022年3月21日
图对抗防御研究进展
专知会员服务
37+阅读 · 2021年12月13日
专知会员服务
12+阅读 · 2021年9月1日
专知会员服务
45+阅读 · 2021年5月24日
专知会员服务
29+阅读 · 2021年2月21日
【AAAI2021】组合对抗攻击
专知会员服务
47+阅读 · 2021年2月17日
轻量级神经网络架构综述
专知会员服务
95+阅读 · 2020年4月29日
相关资讯
在.NET 6 中如何创建和使用 HTTP 客户端 SDK
如何使用流处理器 Pipy 来创建网络代理
InfoQ
0+阅读 · 2022年2月25日
大规模 Node.js 网关的架构设计与工程实践
Dubbo-go 优雅上下线设计与实践
阿里技术
0+阅读 · 2022年1月27日
HTTP客户端演进之路
InfoQ
0+阅读 · 2021年12月6日
开源微服务编排框架:Netflix Conductor
阿里技术
1+阅读 · 2021年12月2日
WhatsApp 是如何实现端到端加密备份的?
InfoQ
0+阅读 · 2021年11月26日
如何在微服务中设计用户权限策略?
InfoQ
0+阅读 · 2021年11月19日
Spring Cloud Gateway一次请求调用源码解析
阿里技术
0+阅读 · 2021年11月15日
设计和实现一款轻量级的爬虫框架
架构文摘
13+阅读 · 2018年1月17日
相关基金
国家自然科学基金
0+阅读 · 2014年12月31日
国家自然科学基金
0+阅读 · 2013年12月31日
国家自然科学基金
1+阅读 · 2013年12月31日
国家自然科学基金
0+阅读 · 2013年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
1+阅读 · 2012年12月31日
国家自然科学基金
0+阅读 · 2012年12月31日
国家自然科学基金
3+阅读 · 2012年12月31日
国家自然科学基金
1+阅读 · 2009年12月31日
Top
微信扫码咨询专知VIP会员