Java网关服务-AIO（二）

Java网关服务-AIO（二）

概述

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

Handlers

接口定义

为了区别一个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