基于Netty的代理网关设计与实现

一、问题背景

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

• 通过给定的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行代码，完整摘录如下：

1. @Slf4j
2. public class SimpleTransProxy {
4. public static void main(String[] args) throws IOException {
5. int port = 8006;
6. ServerSocketChannel localServer = ServerSocketChannel.open();
7. localServer.bind(new InetSocketAddress(port));
8. Reactor reactor = new Reactor();
9. // REACTOR线程
10. GlobalThreadPool.REACTOR_EXECUTOR.submit(reactor::run);
12. // WORKER单线程调试
13. while (localServer.isOpen()) {
14. // 此处阻塞等待连接
15. SocketChannel remoteClient = localServer.accept();
17. // 工作线程
18. GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {
19. @SneakyThrows
20. @Override
21. public void run() {
22. // 代理到远程
23. SocketChannel remoteServer = new ProxyHandler().proxy(remoteClient);
25. // 透明传输
26. reactor.pipe(remoteClient, remoteServer)
27. .pipe(remoteServer, remoteClient);
28. }
29. });
30. }
31. }
32. }
34. @Data
35. class ProxyHandler {
36. private String method;
37. private String host;
38. private int port;
39. private SocketChannel remoteServer;
40. private SocketChannel remoteClient;
42. /**
43. * 原始信息
44. */
45. private List buffers = new ArrayList<>();
46. private StringBuilder stringBuilder = new StringBuilder();
48. /**
49. * 连接到远程
50. * @param remoteClient
51. * @return
52. * @throws IOException
53. */
54. public SocketChannel proxy(SocketChannel remoteClient) throws IOException {
55. this.remoteClient = remoteClient;
56. connect();
57. return this.remoteServer;
58. }
60. public void connect() throws IOException {
61. // 解析METHOD, HOST和PORT
62. beforeConnected();
64. // 链接REMOTE SERVER
65. createRemoteServer();
67. // CONNECT请求回应，其他请求WRITE THROUGH
68. afterConnected();
69. }
71. protected void beforeConnected() throws IOException {
72. // 读取HEADER
73. readAllHeader();
75. // 解析HOST和PORT
76. parseRemoteHostAndPort();
77. }
79. /**
80. * 创建远程连接
81. * @throws IOException
82. */
83. protected void createRemoteServer() throws IOException {
84. remoteServer = SocketChannel.open(new InetSocketAddress(host, port));
85. }
87. /**
88. * 连接建立后预处理
89. * @throws IOException
90. */
91. protected void afterConnected() throws IOException {
92. // 当CONNECT请求时，默认写入200到CLIENT
93. if ("CONNECT".equalsIgnoreCase(method)) {
94. // CONNECT默认为443端口，根据HOST再解析
95. remoteClient.write(ByteBuffer.wrap("HTTP/1.0 200 Connection Established\r\nProxy-agent: nginx\r\n\r\n".getBytes()));
96. } else {
97. writeThrouth();
98. }
99. }
101. protected void writeThrouth() {
102. buffers.forEach(byteBuffer -> {
103. try {
104. remoteServer.write(byteBuffer);
105. } catch (IOException e) {
106. e.printStackTrace();
107. }
108. });
109. }
111. /**
112. * 读取请求内容
113. * @throws IOException
114. */
115. protected void readAllHeader() throws IOException {
116. while (true) {
117. ByteBuffer clientBuffer = newByteBuffer();
118. int read = remoteClient.read(clientBuffer);
119. clientBuffer.flip();
120. appendClientBuffer(clientBuffer);
121. if (read < clientBuffer.capacity()) {
122. break;
123. }
124. }
125. }
127. /**
128. * 解析出HOST和PROT
129. * @throws IOException
130. */
131. protected void parseRemoteHostAndPort() throws IOException {
132. // 读取第一批，获取到METHOD
133. method = parseRequestMethod(stringBuilder.toString());
135. // 默认为80端口，根据HOST再解析
136. port = 80;
137. if ("CONNECT".equalsIgnoreCase(method)) {
138. port = 443;
139. }
141. this.host = parseHost(stringBuilder.toString());
143. URI remoteServerURI = URI.create(host);
144. host = remoteServerURI.getHost();
146. if (remoteServerURI.getPort() > 0) {
147. port = remoteServerURI.getPort();
148. }
149. }
151. protected void appendClientBuffer(ByteBuffer clientBuffer) {
152. buffers.add(clientBuffer);
153. stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit()));
154. }
156. protected static ByteBuffer newByteBuffer() {
157. // buffer必须大于7，保证能读到method
158. return ByteBuffer.allocate(128);
159. }
161. private static String parseRequestMethod(String rawContent) {
162. // create uri
163. return rawContent.split("\r\n")[0].split(" ")[0];
164. }
166. private static String parseHost(String rawContent) {
167. String[] headers = rawContent.split("\r\n");
168. String host = "host:";
169. for (String header : headers) {
170. if (header.length() > host.length()) {
171. String key = header.substring(0, host.length());
172. String value = header.substring(host.length()).trim();
173. if (host.equalsIgnoreCase(key)) {
174. if (!value.startsWith("http://") && !value.startsWith("https://")) {
175. value = "http://" + value;
176. }
177. return value;
178. }
179. }
180. }
181. return "";
182. }
184. }
186. @Slf4j
187. @Data
188. class Reactor {
190. private Selector selector;
192. private volatile boolean finish = false;
194. @SneakyThrows
195. public Reactor() {
196. selector = Selector.open();
197. }
199. @SneakyThrows
200. public Reactor pipe(SocketChannel from, SocketChannel to) {
201. from.configureBlocking(false);
202. from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));
203. return this;
204. }
206. @SneakyThrows
207. public void run() {
208. try {
209. while (!finish) {
210. if (selector.selectNow() > 0) {
211. Iterator it = selector.selectedKeys().iterator();
212. while (it.hasNext()) {
213. SelectionKey selectionKey = it.next();
214. if (selectionKey.isValid() && selectionKey.isReadable()) {
215. ((SocketPipe) selectionKey.attachment()).pipe();
216. }
217. it.remove();
218. }
219. }
220. }
221. } finally {
222. close();
223. }
224. }
226. @SneakyThrows
227. public synchronized void close() {
228. if (finish) {
229. return;
230. }
231. finish = true;
232. if (!selector.isOpen()) {
233. return;
234. }
235. for (SelectionKey key : selector.keys()) {
236. closeChannel(key.channel());
237. key.cancel();
238. }
239. if (selector != null) {
240. selector.close();
241. }
242. }
244. public void cancel(SelectableChannel channel) {
245. SelectionKey key = channel.keyFor(selector);
246. if (Objects.isNull(key)) {
247. return;
248. }
249. key.cancel();
250. }
252. @SneakyThrows
253. public void closeChannel(Channel channel) {
254. SocketChannel socketChannel = (SocketChannel)channel;
255. if (socketChannel.isConnected() && socketChannel.isOpen()) {
256. socketChannel.shutdownOutput();
257. socketChannel.shutdownInput();
258. }
259. socketChannel.close();
260. }
261. }
263. @Data
264. @AllArgsConstructor
265. class SocketPipe {
267. private Reactor reactor;
269. private SocketChannel from;
271. private SocketChannel to;
273. @SneakyThrows
274. public void pipe() {
275. // 取消监听
276. clearInterestOps();
278. GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {
279. @SneakyThrows
280. @Override
281. public void run() {
282. int totalBytesRead = 0;
283. ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
284. while (valid(from) && valid(to)) {
285. byteBuffer.clear();
286. int bytesRead = from.read(byteBuffer);
287. totalBytesRead = totalBytesRead + bytesRead;
288. byteBuffer.flip();
289. to.write(byteBuffer);
290. if (bytesRead < byteBuffer.capacity()) {
291. break;
292. }
293. }
294. if (totalBytesRead < 0) {
295. reactor.closeChannel(from);
296. reactor.cancel(from);
297. } else {
298. // 重置监听
299. resetInterestOps();
300. }
301. }
302. });
303. }
305. protected void clearInterestOps() {
306. from.keyFor(reactor.getSelector()).interestOps(0);
307. to.keyFor(reactor.getSelector()).interestOps(0);
308. }
310. protected void resetInterestOps() {
311. from.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
312. to.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
313. }
315. private boolean valid(SocketChannel channel) {
316. return channel.isConnected() && channel.isRegistered() && channel.isOpen();
317. }
318. }

以上，借鉴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端口，此时：

1. curl -x 'localhost:8006' http://httpbin.org/get测试HTTP请求
3. 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

1. public class SslSocketChannel {
3. /**
4. * 握手加解密需要的四个存储
5. */
6. protected ByteBuffer myAppData; // 明文
7. protected ByteBuffer myNetData; // 密文
8. protected ByteBuffer peerAppData; // 明文
9. protected ByteBuffer peerNetData; // 密文
11. /**
12. * 握手加解密过程中用到的异步执行器
13. */
14. protected ExecutorService executor = Executors.newSingleThreadExecutor();
16. /**
17. * 原NIO 的 CHANNEL
18. */
19. protected SocketChannel socketChannel;
21. /**
22. * SSL 引擎
23. */
24. protected SSLEngine engine;
26. public SslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode) throws Exception {
27. // 原始的NIO SOCKET
28. this.socketChannel = socketChannel;
30. // 初始化BUFFER
31. SSLSession dummySession = context.createSSLEngine().getSession();
32. myAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
33. myNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
34. peerAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
35. peerNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
36. dummySession.invalidate();
38. engine = context.createSSLEngine();
39. engine.setUseClientMode(clientMode);
40. engine.beginHandshake();
41. }
43. /**
44. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
45. * 实现的 SSL 的握手协议
46. * @return
47. * @throws IOException
48. */
49. protected boolean doHandshake() throws IOException {
50. SSLEngineResult result;
51. HandshakeStatus handshakeStatus;
53. int appBufferSize = engine.getSession().getApplicationBufferSize();
54. ByteBuffer myAppData = ByteBuffer.allocate(appBufferSize);
55. ByteBuffer peerAppData = ByteBuffer.allocate(appBufferSize);
56. myNetData.clear();
57. peerNetData.clear();
59. handshakeStatus = engine.getHandshakeStatus();
60. while (handshakeStatus != HandshakeStatus.FINISHED && handshakeStatus != HandshakeStatus.NOT_HANDSHAKING) {
61. switch (handshakeStatus) {
62. case NEED_UNWRAP:
63. if (socketChannel.read(peerNetData) < 0) {
64. if (engine.isInboundDone() && engine.isOutboundDone()) {
65. return false;
66. }
67. try {
68. engine.closeInbound();
69. } catch (SSLException e) {
70. log.debug("收到END OF STREAM，关闭连接.", e);
71. }
72. engine.closeOutbound();
73. handshakeStatus = engine.getHandshakeStatus();
74. break;
75. }
76. peerNetData.flip();
77. try {
78. result = engine.unwrap(peerNetData, peerAppData);
79. peerNetData.compact();
80. handshakeStatus = result.getHandshakeStatus();
81. } catch (SSLException sslException) {
82. engine.closeOutbound();
83. handshakeStatus = engine.getHandshakeStatus();
84. break;
85. }
86. switch (result.getStatus()) {
87. case OK:
88. break;
89. case BUFFER_OVERFLOW:
90. peerAppData = enlargeApplicationBuffer(engine, peerAppData);
91. break;
92. case BUFFER_UNDERFLOW:
93. peerNetData = handleBufferUnderflow(engine, peerNetData);
94. break;
95. case CLOSED:
96. if (engine.isOutboundDone()) {
97. return false;
98. } else {
99. engine.closeOutbound();
100. handshakeStatus = engine.getHandshakeStatus();
101. break;
102. }
103. default:
104. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
105. }
106. break;
107. case NEED_WRAP:
108. myNetData.clear();
109. try {
110. result = engine.wrap(myAppData, myNetData);
111. handshakeStatus = result.getHandshakeStatus();
112. } catch (SSLException sslException) {
113. engine.closeOutbound();
114. handshakeStatus = engine.getHandshakeStatus();
115. break;
116. }
117. switch (result.getStatus()) {
118. case OK :
119. myNetData.flip();
120. while (myNetData.hasRemaining()) {
121. socketChannel.write(myNetData);
122. }
123. break;
124. case BUFFER_OVERFLOW:
125. myNetData = enlargePacketBuffer(engine, myNetData);
126. break;
127. case BUFFER_UNDERFLOW:
128. throw new SSLException("加密后消息内容为空，报错");
129. case CLOSED:
130. try {
131. myNetData.flip();
132. while (myNetData.hasRemaining()) {
133. socketChannel.write(myNetData);
134. }
135. peerNetData.clear();
136. } catch (Exception e) {
137. handshakeStatus = engine.getHandshakeStatus();
138. }
139. break;
140. default:
141. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
142. }
143. break;
144. case NEED_TASK:
145. Runnable task;
146. while ((task = engine.getDelegatedTask()) != null) {
147. executor.execute(task);
148. }
149. handshakeStatus = engine.getHandshakeStatus();
150. break;
151. case FINISHED:
152. break;
153. case NOT_HANDSHAKING:
154. break;
155. default:
156. throw new IllegalStateException("无效的握手状态: " + handshakeStatus);
157. }
158. }
160. return true;
161. }
163. /**
164. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
165. * 实现的 SSL 的传输读取协议
166. * @param consumer
167. * @throws IOException
168. */
169. public void read(Consumer consumer) throws IOException {
170. // BUFFER初始化
171. peerNetData.clear();
172. int bytesRead = socketChannel.read(peerNetData);
173. if (bytesRead > 0) {
174. peerNetData.flip();
175. while (peerNetData.hasRemaining()) {
176. peerAppData.clear();
177. SSLEngineResult result = engine.unwrap(peerNetData, peerAppData);
178. switch (result.getStatus()) {
179. case OK:
180. log.debug("收到远程的返回结果消息为：" + new String(peerAppData.array(), 0, peerAppData.position()));
181. consumer.accept(peerAppData);
182. peerAppData.flip();
183. break;
184. case BUFFER_OVERFLOW:
185. peerAppData = enlargeApplicationBuffer(engine, peerAppData);
186. break;
187. case BUFFER_UNDERFLOW:
188. peerNetData = handleBufferUnderflow(engine, peerNetData);
189. break;
190. case CLOSED:
191. log.debug("收到远程连接关闭消息.");
192. closeConnection();
193. return;
194. default:
195. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
196. }
197. }
198. } else if (bytesRead < 0) {
199. log.debug("收到END OF STREAM，关闭连接.");
200. handleEndOfStream();
201. }
202. }
204. public void write(String message) throws IOException {
205. write(ByteBuffer.wrap(message.getBytes()));
206. }
208. /**
209. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
210. * 实现的 SSL 的传输写入协议
211. * @param message
212. * @throws IOException
213. */
214. public void write(ByteBuffer message) throws IOException {
215. myAppData.clear();
216. myAppData.put(message);
217. myAppData.flip();
218. while (myAppData.hasRemaining()) {
219. myNetData.clear();
220. SSLEngineResult result = engine.wrap(myAppData, myNetData);
221. switch (result.getStatus()) {
222. case OK:
223. myNetData.flip();
224. while (myNetData.hasRemaining()) {
225. socketChannel.write(myNetData);
226. }
227. log.debug("写入远程的消息为: {}", message);
228. break;
229. case BUFFER_OVERFLOW:
230. myNetData = enlargePacketBuffer(engine, myNetData);
231. break;
232. case BUFFER_UNDERFLOW:
233. throw new SSLException("加密后消息内容为空.");
234. case CLOSED:
235. closeConnection();
236. return;
237. default:
238. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
239. }
240. }
241. }
243. /**
244. * 关闭连接
245. * @throws IOException
246. */
247. public void closeConnection() throws IOException {
248. engine.closeOutbound();
249. doHandshake();
250. socketChannel.close();
251. executor.shutdown();
252. }
254. /**
255. * END OF STREAM(-1)默认是关闭连接
256. * @throws IOException
257. */
258. protected void handleEndOfStream() throws IOException {
259. try {
260. engine.closeInbound();
261. } catch (Exception e) {
262. log.error("END OF STREAM 关闭失败.", e);
263. }
264. closeConnection();
265. }
267. }

以上：

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

SslSocketChannel测试服务端

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

1. @Slf4j
2. public class NioSslServer {
4. public static void main(String[] args) throws Exception {
5. NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006);
6. sslServer.start()
7. // 使用 curl -vv -k 'https://localhost:8006' 连接
8. }
10. private SSLContext context;
12. private Selector selector;
14. public NioSslServer(String hostAddress, int port) throws Exception {
15. // 初始化SSL Context
16. context = serverSSLContext();
18. // 注册监听器
19. selector = SelectorProvider.provider().openSelector();
20. ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
21. serverSocketChannel.configureBlocking(false);
22. serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));
23. serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
24. }
26. public void start() throws Exception {
28. log.debug("等待连接中.");
30. while (true) {
31. selector.select();
32. Iterator selectedKeys = selector.selectedKeys().iterator();
33. while (selectedKeys.hasNext()) {
34. SelectionKey key = selectedKeys.next();
35. selectedKeys.remove();
36. if (!key.isValid()) {
37. continue;
38. }
39. if (key.isAcceptable()) {
40. accept(key);
41. } else if (key.isReadable()) {
42. ((SslSocketChannel)key.attachment()).read(buf->{});
43. // 直接回应一个OK
44. ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n");
45. ((SslSocketChannel)key.attachment()).closeConnection();
46. }
47. }
48. }
49. }
51. private void accept(SelectionKey key) throws Exception {
52. log.debug("接收新的请求.");
54. SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept();
55. socketChannel.configureBlocking(false);
57. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);
58. if (sslSocketChannel.doHandshake()) {
59. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
60. } else {
61. socketChannel.close();
62. log.debug("握手失败，关闭连接.");
63. }
64. }
65. }

以上：

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

SslSocketChannel测试服务端

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

1. @Slf4j
2. public class NioSslServer {
4. public static void main(String[] args) throws Exception {
5. NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006);
6. sslServer.start();
7. // 使用 curl -vv -k 'https://localhost:8006' 连接
8. }
10. private SSLContext context;
12. private Selector selector;
14. public NioSslServer(String hostAddress, int port) throws Exception {
15. // 初始化SSL Context
16. context = serverSSLContext();
18. // 注册监听器
19. selector = SelectorProvider.provider().openSelector();
20. ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
21. serverSocketChannel.configureBlocking(false);
22. serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));
23. serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
24. }
26. public void start() throws Exception {
28. log.debug("等待连接中.");
30. while (true) {
31. selector.select();
32. Iterator selectedKeys = selector.selectedKeys().iterator();
33. while (selectedKeys.hasNext()) {
34. SelectionKey key = selectedKeys.next();
35. selectedKeys.remove();
36. if (!key.isValid()) {
37. continue;
38. }
39. if (key.isAcceptable()) {
40. accept(key);
41. } else if (key.isReadable()) {
42. ((SslSocketChannel)key.attachment()).read(buf->{});
43. // 直接回应一个OK
44. ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n");
45. ((SslSocketChannel)key.attachment()).closeConnection();
46. }
47. }
48. }
49. }
51. private void accept(SelectionKey key) throws Exception {
52. log.debug("接收新的请求.");
54. SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept();
55. socketChannel.configureBlocking(false);
57. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);
58. if (sslSocketChannel.doHandshake()) {
59. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
60. } else {
61. socketChannel.close();
62. log.debug("握手失败，关闭连接.");
63. }
64. }
65. }

以上：

由于是NIO，简单的测试需要用到NIO的基础组件Selector进行测试;

首先初始化ServerSocketChannel，监听8006端口;

接收到请求后，将SocketChannel封装为SslSocketChannel，注册到Selector；

接收到数据后，通过SslSocketChannel做read和write;

以上：

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

SslSocketChannel测试客户端

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

1. @Slf4j ：
2. public class NioSslClient {
4. public static void main(String[] args) throws Exception {
5. NioSslClient sslClient = new NioSslClient("httpbin.org", 443);
6. sslClient.connect();
7. // 请求 'https://httpbin.org/get'
8. }
10. private String remoteAddress;
12. private int port;
14. private SSLEngine engine;
16. private SocketChannel socketChannel;
18. private SSLContext context;
20. /**
21. * 需要远程的HOST和PORT
22. * @param remoteAddress
23. * @param port
24. * @throws Exception
25. */
26. public NioSslClient(String remoteAddress, int port) throws Exception {
27. this.remoteAddress = remoteAddress;
28. this.port = port;
30. context = clientSSLContext();
31. engine = context.createSSLEngine(remoteAddress, port);
32. engine.setUseClientMode(true);
33. }
35. public boolean connect() throws Exception {
36. socketChannel = SocketChannel.open();
37. socketChannel.configureBlocking(false);
38. socketChannel.connect(new InetSocketAddress(remoteAddress, port));
39. while (!socketChannel.finishConnect()) {
40. // 通过REACTOR，不会出现等待情况
41. //log.debug("连接中..");
42. }
44. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true);
45. sslSocketChannel.doHandshake();
47. // 握手完成后，开启SELECTOR
48. Selector selector = SelectorProvider.provider().openSelector();
49. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
51. // 写入请求
52. sslSocketChannel.write("GET /get HTTP/1.1\r\n"
53. + "Host: httpbin.org:443\r\n"
54. + "User-Agent: curl/7.62.0\r\n"
55. + "Accept: */*\r\n"
56. + "\r\n");
58. // 读取结果
59. while (true) {
60. selector.select();
61. Iterator selectedKeys = selector.selectedKeys().iterator();
62. while (selectedKeys.hasNext()) {
63. SelectionKey key = selectedKeys.next();
64. selectedKeys.remove();
65. if (key.isValid() && key.isReadable()) {
66. ((SslSocketChannel)key.attachment()).read(buf->{
67. log.info("{}", new String(buf.array(), 0, buf.position()));
68. });
69. ((SslSocketChannel)key.attachment()).closeConnection();
70. return true;
71. }
72. }
73. }
74. }
75. }

总结

以上：

• 非透明代理需要拿到完整的请求数据，可以通过 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.代理网关服务端

主。要包括

• 初始化代理网关服务端
• 初始化服务端处理器
• 服务端协议升级与处理

初始化代理网关服务

1. public void start() {
2. HookedExecutors.newSingleThreadExecutor().submit(() ->{
3. log.info("开始启动代理服务器，监听端口：{}", auditProxyConfig.getProxyServerPort());
4. EventLoopGroup bossGroup = new NioEventLoopGroup(auditProxyConfig.getBossThreadCount());
5. EventLoopGroup workerGroup = new NioEventLoopGroup(auditProxyConfig.getWorkThreadCount());
6. try {
7. ServerBootstrap b = new ServerBootstrap();
8. b.group(bossGroup, workerGroup)
9. .channel(NioServerSocketChannel.class)
10. .handler(new LoggingHandler(LogLevel.DEBUG))
11. .childHandler(new ServerChannelInitializer(auditProxyConfig))
12. .bind(auditProxyConfig.getProxyServerPort()).sync().channel().closeFuture().sync();
13. } catch (InterruptedException e) {
14. log.error("代理服务器被中断.", e);
15. Thread.currentThread().interrupt();
16. } finally {
17. bossGroup.shutdownGracefully();
18. workerGroup.shutdownGracefully();
19. }
20. });
21. }

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

bossGroup线程组，负责接收请求

workerGroup线程组，负责处理接收的请求数据，具体处理逻辑封装在ServerChannelInitializer中。

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

1. @Override
2. protected void initChannel(SocketChannel ch) throws Exception {
3. ch.pipeline()
4. .addLast(new HttpRequestDecoder())
5. .addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize()))
6. .addLast(new ServerChannelHandler(auditProxyConfig));
7. }

首先解析HTTP请求，然后做聚批的处理，最后ServerChannelHandler实现代理网关协议;

代理网关协议：

• 判定是否是CONNECT请求，如果是，会存储CONNECT请求;暂停读取，发送代理成功的响应，并在回应成功后，升级协议;
• 升级引擎，本质上是采用SslSocketChannel对原SocketChannel做透明的封装;
• 最后根据CONNECT请求连接远程服务端;

详细实现为：

1. @Override
2. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3. FullHttpRequest request = (FullHttpRequest)msg;
5. try {
6. if (isConnectRequest(request)) {
7. // CONNECT 请求，存储待处理
8. saveConnectRequest(ctx, request);
10. // 禁止读取
11. ctx.channel().config().setAutoRead(false);
13. // 发送回应
14. connectionEstablished(ctx, ctx.newPromise().addListener(future -> {
15. if (future.isSuccess()) {
16. // 升级
17. if (isSslRequest(request) && !isUpgraded(ctx)) {
18. upgrade(ctx);
19. }
21. // 开放消息读取
22. ctx.channel().config().setAutoRead(true);
23. ctx.read();
24. }
25. }));
27. } else {
28. // 其他请求，判定是否已升级
29. if (!isUpgraded(ctx)) {
31. // 升级引擎
32. upgrade(ctx);
33. }
35. // 连接远程
36. connectRemote(ctx, request);
37. }
38. } finally {
39. ctx.fireChannelRead(msg);
40. }
41. }

2.代理网关客户端

代理网关服务端需要连接远程服务，进入代理网关客户端部分。

代理网关客户端初始化：

1. /**
2. * 初始化远程连接
3. * @param ctx
4. * @param httpRequest
5. */
6. protected void connectRemote(ChannelHandlerContext ctx, FullHttpRequest httpRequest) {
7. Bootstrap b = new Bootstrap();
8. b.group(ctx.channel().eventLoop()) // use the same EventLoop
9. .channel(ctx.channel().getClass())
10. .handler(new ClientChannelInitializer(auditProxyConfig, ctx, safeCopy(httpRequest)));
12. // 动态连接代理
13. FullHttpRequest originRequest = ctx.channel().attr(CONNECT_REQUEST).get();
14. if (originRequest == null) {
15. originRequest = httpRequest;
16. }
17. ChannelFuture cf = b.connect(new InetSocketAddress(calculateHost(originRequest), calculatePort(originRequest)));
18. Channel cch = cf.channel();
19. ctx.channel().attr(CLIENT_CHANNEL).set(cch);
20. }

以上：

• 复用代理网关服务端的workerGroup线程组;
• 请求和结果的处理封装在ClientChannelInitializer;
• 连接的远程服务端的HOST和PORT在服务端收到的请求中可以解析到。

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

1. @Override
2. protected void initChannel(SocketChannel ch) throws Exception {
3. SocketAddress socketAddress = calculateProxy();
4. if (!Objects.isNull(socketAddress)) {
5. ch.pipeline().addLast(new HttpProxyHandler(calculateProxy(), auditProxyConfig.getUserName(), auditProxyConfig
6. .getPassword()));
7. }
8. if (isSslRequest()) {
9. String host = host();
10. int port = port();
11. if (StringUtils.isNoneBlank(host) && port > 0) {
12. ch.pipeline().addLast(new SslHandler(sslEngine(host, port)));
13. }
14. }
15. ch.pipeline().addLast(new ClientChannelHandler(clientContext, httpRequest));
16. }

以上：

如果下游是代理，那么会采用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中包含地域标识时，代理网关会尽力保证请求打入指定的地域代理，经由地域代理访问远程服务端

本文参考https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html实现 SslSocketChannel，以透明处理HTTP和HTTPS协议。

原文链接：https://zhuanlan.51cto.com/art/202111/692258.htm