V2AS问路
Java网关服务-AIO（二）
阅读原文时间：2023年07月09日阅读：3

Java网关服务-AIO（二）

AIO的特点就是用户程序注册一个事件后就可以做其他事情，当事件被内核执行并得到结果后，我们的CompletionHandler会在I/O回调线程中被自动调用，有点类似观察者模式；因此我们的服务端会有很多个CompletionHandler

接口定义

为了区别一个Handler是从接收缓冲区读入数据到ByteBuffer，还是将ByteBuffer中的内容写入发送缓冲区，我们定义了两个接口

InboundHandler

/**
 * server端处理请求的handler
 */
public interface InboundHandler {

    /**
     * 从通道接收缓冲区读取数据
     * @param socketChannel client的channel
     * @param in 分配的byteBuffer缓冲
     * @return
     */
    Object read(AsynchronousSocketChannel socketChannel, ByteBuffer in);
}

OutboundHandler

/**
 * 对外输出的handler
 */
public interface OutboundHandler {

    /**
     * 向通道写输出的数据
     *
     * @param socketChannel client对应的channel
     * @param out           向通道输出的数据
     * @return
     */
    Object write(AsynchronousSocketChannel socketChannel, ByteBuffer out);

}

ChannelAcceptedHandler

/**
 * accept完成时的handler
 * deocder:  4 + body
 * 前4个字节为一个int，值为body.length,后面紧跟body
 * BEFORE DECODE (16 bytes)         AFTER DECODE (12 bytes)
 * +--------+----------------+      +----------------+
 * | Length | Actual Content |----->| Actual Content |
 * | 0x000C | "HELLO, WORLD" |      | "HELLO, WORLD" |
 * +--------+----------------+      +----------------+
 * <p/>
 * 无论成功还是失败，都要继续使server accept请求
 * 不要再AcceptHandler中同步的读取数据
 */
public class ChannelAcceptedHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel>, InboundHandler {

    private final static Logger LOGGER = LoggerFactory.getLogger(ChannelAcceptedHandler.class);

    /**
     * 仅接受localhost请求
     */
    private boolean onlyAcceptLocalhost;

    public ChannelAcceptedHandler(boolean onlyAcceptLocalhost) {
        this.onlyAcceptLocalhost = onlyAcceptLocalhost;
    }

    public void completed(AsynchronousSocketChannel socketChannel, AsynchronousServerSocketChannel attachment) {
        //继续接受其他客户端的连接
        attachment.accept(attachment, this);

        try {
            InetSocketAddress inetSocketAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
            String host = inetSocketAddress.getAddress().getHostAddress();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("remote host:{}", host);
            }

            if (onlyAcceptLocalhost && !"127.0.0.1".equals(host)) {
                LOGGER.warn("拒绝ip:{}的连接", host);
                socketChannel.close();
                return;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        //读取前4个字节，也就是body的长度
        ByteBuffer buffer = ByteBuffer.allocate(4);

        read(socketChannel, buffer);

    }

    @Override
    public Object read(AsynchronousSocketChannel socketChannel, ByteBuffer in) {
        //开始监听读头部buffer
        socketChannel.read(in, in, new ChannelReadHeadHandler(socketChannel));
        return null;
    }

    public void failed(Throwable exc, AsynchronousServerSocketChannel attachment) {
        attachment.accept(attachment, this);
        LOGGER.error("连接错误", exc);
    }
}

ChannelAcceptedHandler实现了InboundHandler，表明它是接收请求信息的CompletionHandler

onlyAcceptLocalhost

onlyAcceptLocalhost是用来进行ip过滤的，是否只接受127.0.0.1的连接，如果连接应该被拒绝，则将socketChannel.close()调，不再向下继续处理

继续接受其他客户端的连接

attachment.accept(attachment, this);

我们在AsyncServer执行了serverChannel.accept方法，为什么这里要继续执行呢？

答：执行serverChannel.accept仅注册了一次监听连接的事件，当服务端在accept完一个连接后，如果不继续注册，则服务端就无法继续接受请求了

需要非常注意的是，很多人只知道accept方法需要在每次accept成功后都要接着调用一次，却忽略了在accept失败时也需要继续调用。试想如果一个客户端在connect到server后，立即将连接close掉，而这个时候，server正准备处理这个尝试建立连接的请求，却发现已经不能用了，此时会触发failed方法，而非completed方法。网络情况是及其复杂的，这样的情况不可谓不常见。在实际测试中，我有过这样的测试用例，这也是在被坑后才注意到的。

协议

如果继续网下面讲，就需要聊到协议了。通过查看socketChannel.read方法，你会发现AIO是先声明一个ByteBuffer，让内核往里面读入对应个数的byte，也就是说，需要先确定缓冲区大小，再调用read方法，注册回调事件。

纳尼，为什么不让我有机会先知道一下总的长度呢，例如Http协议有一个请求头，可以通过请求头中的信息，知道整个请求的大小。

这也是AIO看似编码简单，但是却很麻烦的地方。

我们采用了一个极其简单的协议：int(4 byte) + realbody的协议，也就是说先读到4个字节，解析成int类型，就知道后面的realbody有多少个字节，和客户端提前约定好；

    ByteBuffer buffer = ByteBuffer.allocate(4);

    read(socketChannel, buffer);

先分配4个字节到buffer中，再继续注册read事件

     socketChannel.read(in, in, new ChannelReadHeadHandler(socketChannel));

ChannelReadHeadHandler

ChannelReadHeadHandler将会成功读到4个字节，从而获取body的长度

/**
 * 读取请求的内容，业务处理,取前4个字节，获取body的具体长度
 */
public class ChannelReadHeadHandler implements CompletionHandler<Integer, ByteBuffer>, InboundHandler {

    private final static Logger LOGGER = LoggerFactory.getLogger(ChannelReadHeadHandler.class);

    private AsynchronousSocketChannel channel;

    private static int BODY_PAYLOAD = Constants.MB_1;

    static {
        String maxBodySize = System.getProperty("max_body_size");
        if (maxBodySize != null && maxBodySize.trim().equals("")) {
            try {
                BODY_PAYLOAD = Integer.valueOf(maxBodySize);
            } catch (Exception e) {
                LOGGER.warn("环境变量max_body_size转换int失败:{}", maxBodySize);
            }
        }

        if (BODY_PAYLOAD < 1) {
            BODY_PAYLOAD = Constants.MB_1;
        }

        LOGGER.info("body-payload:{}", BODY_PAYLOAD);
    }

    public ChannelReadHeadHandler(AsynchronousSocketChannel channel) {
        this.channel = channel;
    }

    public void completed(Integer result, ByteBuffer attachment) {
        //如果条件成立，说明客户端主动终止了TCP套接字，这时服务端终止就可以了
        if (result == -1 || result < 4) {
            System.out.println("remote is close");
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }

            return;
        }

        attachment.flip();
        //获取实际body的长度
        Integer bodyLen = attachment.getInt();
        LOGGER.info("bodyLen:" + bodyLen);
        if (bodyLen > BODY_PAYLOAD) {
            LOGGER.warn("请求体过大，直接丢弃,currentBodyLen:{}, maxLen:{}", bodyLen, BODY_PAYLOAD);
//                result.read()
            releaseConnection(channel);
            return;
        }
        //为body生成一个buffer
        ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen);

        read(channel, bodyBuffer);
    }

    @Override
    public Object read(AsynchronousSocketChannel socketChannel, ByteBuffer in) {
        //开始监听读buffer
        socketChannel.read(in, in, new ChannelServerHandler(socketChannel));
        return null;
    }

    public void failed(Throwable exc, ByteBuffer attachment) {
        releaseConnection(this.channel);
    }

    private void releaseConnection(Channel channel) {
        try {
            channel.close();
            LOGGER.warn("server close client channle");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

BODY_PAYLOAD

约定一个body的最大长度，我们的服务端不可能接受无止境的body，一定需要设置一个最大长度，超过这个长度的，就直接拒绝，如1MB

长度判断

        //如果条件成立，说明客户端主动终止了TCP套接字，这时服务端终止就可以了
        if (result == -1 || result < 4) {
            System.out.println("remote is close");
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }

            return;
        }

result是成功读取到的byte个数

  • 如果为-1，标识没有读到任何东西，这次socket传输已经到达了end of stream,这个我们预期或者约定的有出入，直接关闭连接
  • 如果<4,异常情况，server无法处理一个3个字节的长度，这也和约定不符合，关闭连接，结束

获取body长度

    attachment.flip();
    //获取实际body的长度
    Integer bodyLen = attachment.getInt();
    LOGGER.info("bodyLen:" + bodyLen);
    if (bodyLen > BODY_PAYLOAD) {
        LOGGER.warn("请求体过大，直接丢弃,currentBodyLen:{}, maxLen:{}", bodyLen, BODY_PAYLOAD);
        releaseConnection(channel);
        return;
    }

ByteBuffer的常规操作，要读时先调用flip()方法，切换为读模式

attachment.getInt()从buffer中读出一个int，此处一共4个字节，恰好可以读到一个int，也就是body的长度，成功获取长度后立即判断长度是否过大

继续读body内容

 socketChannel.read(in, in, new ChannelServerHandler(socketChannel));

既然已经确定了长度，那么我们可以分配指定长度大小的Buffer继续从通道里面取数据了

ChannelServerHandler

下一节将讲解具体处理请求Handler

手机扫一扫

移动阅读更方便

阿里云服务器
腾讯云服务器
七牛云服务器

你可能感兴趣的文章

【RocketMQ】主从同步实现原理
Java网关服务-AIO（三）