NIO实现的Http正向代理

package org.example; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.example.common.HttpRequestParser; import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; public class ServerEventLoop extends AbstractEventLoop { private final Logger logger = LogManager.getLogger(ServerEventLoop.class); public ServerEventLoop(Bootstrap bootstrap) { super(bootstrap); } @Override protected void processSelectedKey(SelectionKey key) { if (key.isValid() && key.isAcceptable()) { if (key.attachment() instanceof Acceptor acceptor) { acceptor.accept(); } } if (key.isValid() && key.isReadable()) { if (key.attachment() instanceof ChannelHandler channelHandler) { channelHandler.handleRead(); } } if (key.isValid() && key.isConnectable()) { key.interestOpsAnd(~SelectionKey.OP_CONNECT); if (key.attachment() instanceof ChannelHandler channelHandler) { channelHandler.handleConnect(); } } if (key.isValid() && key.isWritable()) { key.interestOpsAnd(~SelectionKey.OP_WRITE); if (key.attachment() instanceof ChannelHandler channelHandler) { channelHandler.handleWrite(); } } } @Override public void bind(int port) throws Exception { ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); SelectionKey key = serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT); key.attach(new Acceptor(serverSocketChannel)); serverSocketChannel.bind(new InetSocketAddress(port)); } class Acceptor { ServerSocketChannel ssc; public Acceptor(ServerSocketChannel ssc) { this.ssc = ssc; } public void accept() { try { SocketChannel socketChannel = ssc.accept(); socketChannel.configureBlocking(false); socketChannel.register(selector, SelectionKey.OP_READ, new ClientChannelHandler(socketChannel)); logger.info("accept client connection"); } catch (IOException e) { logger.error("accept error"); } } } abstract class ChannelHandler { Logger logger; SocketChannel channel; ByteBuffer writeBuffer; public ChannelHandler(SocketChannel channel) { this.logger = LogManager.getLogger(this.getClass()); this.channel = channel; this.writeBuffer = null; } abstract void handleRead(); public void handleWrite() { doWrite(); } public abstract void onChannelClose(); public ByteBuffer doRead() { ByteBuffer buffer = ByteBuffer.allocate(4096); try { int len = channel.read(buffer); if (len == -1) { logger.info("read end-of-stream, close channel {}", channel); channel.close(); onChannelClose(); } if (len > 0) { buffer.flip(); } } catch (IOException e) { logger.error("read channel error"); try { channel.close(); onChannelClose(); } catch (IOException ex) { logger.error("close channel error."); } } return buffer; } public void doWrite() { if (writeBuffer != null) { try { while (writeBuffer.hasRemaining()) { channel.write(writeBuffer); } } catch (IOException e) { logger.error("write channel error."); try { channel.close(); onChannelClose(); } catch (IOException ex) { logger.error("close channel error"); } } writeBuffer = null; } } public void handleConnect() { } } class ClientChannelHandler extends ChannelHandler { HttpRequestParser requestParser; private SelectableChannel proxyChannel; public ClientChannelHandler(SocketChannel sc) { super(sc); this.channel = sc; this.requestParser = new HttpRequestParser(); this.proxyChannel = null; } @Override public void handleRead() { if (requestParser.isParsed()) { if (proxyChannel != null) { SelectionKey proxyKey = proxyChannel.keyFor(selector); if (proxyKey != null && proxyKey.isValid() && proxyKey.attachment() instanceof ProxyChannelHandler proxyHandler) { //需要等待ProxyHandler的写入缓存为空后才可读取客户端的数据 if (proxyHandler.writeBuffer == null) { ByteBuffer buffer = doRead(); if (buffer.hasRemaining() && proxyKey.isValid()) { proxyHandler.writeBuffer = buffer; proxyKey.interestOpsOr(SelectionKey.OP_WRITE); } } } } } else { ByteBuffer buffer = doRead(); requestParser.putFromByteBuffer(buffer); if (requestParser.isParsed()) { //连接到目标服务器 ByteBuffer buf = null; if (requestParser.getMethod().equals(HttpRequestParser.HTTP_METHOD_CONNECT)) { //回写客户端连接成功 SelectionKey clientKey = channel.keyFor(selector); if (clientKey != null && clientKey.isValid() && clientKey.attachment() instanceof ClientChannelHandler clientHandler) { clientHandler.writeBuffer = ByteBuffer.wrap((requestParser.getProtocol() + " 200 Connection Established\r\n\r\n").getBytes()); clientKey.interestOpsOr(SelectionKey.OP_WRITE); } } else { //将缓存的客户端的数据通过代理转发 byte[] allBytes = requestParser.getAllBytes(); buf = ByteBuffer.wrap(allBytes); } this.proxyChannel = connect(requestParser.getAddress(), buf); } } } @Override public void onChannelClose() { try { if (proxyChannel != null) { proxyChannel.close(); } } catch (IOException e) { logger.error("close channel error"); } } private SocketChannel connect(String address, ByteBuffer buffer) { String host = address; int port = 80; if (address.contains(":")) { host = address.split(":")[0].trim(); port = Integer.parseInt(address.split(":")[1].trim()); } SocketA