簡介：本文重點在代理網關本身的設計與實現，而非代理資源的管理與維護。
作者 | 新然
來源 | 阿里技術公眾號
一 問題背景
- 平台端購置一批裸代理，來做廣告異地展現審核。從外部購置的代理，使用方式為：
- 通過給定的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'
curl -x 'localhost:8006'
注意，此時代理服務代理了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 做握手；
- 入站數據做解密，出站數據做加密；
- 握手，數據加密和數據解密是統一的一套狀態機；
- 基於 SSL 協議，實現統一的握手動作；
- 分別實現讀取的解密，和寫入的加密方法；
- 將 SslSocketChannel 實現為 SocketChannel的Decorator；
SslSocketChannel測試服務端
基於以上封裝，簡單測試服務端如下
- 由於是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 非透明上游代理
非透明的上游代理，相比非透明的代理要復雜一些
- 如果是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請求連接遠程服務端；
詳細實現為：
代理網關服務端需要連接遠程服務，進入代理網關客戶端部分。
代理網關客戶端初始化：
/**
* 初始化遠程連接
* @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中包含地域標識時，代理網關會盡力保證請求打入指定的地域代理，經由地域代理訪問遠程服務端
原文鏈接
本文為阿里雲原創內容，未經允許不得轉載。