用 Apache MINA 2 开发网络应用

jieforest

I/O 会话

I/O 会话表示一个活动的网络连接，与所使用的传输方式无关。I/O 会话可以用来存储用户自定义的与应用相关的属性。这些属性通常用来保存应用的状态信息，还可以用来在 I/O 过滤器和 I/O 处理器之间交换数据。I/O 会话在作用上类似于 Servlet 规范中的 HTTP 会话。

Apache MINA 中 I/O 会话实现的接口是 org.apache.mina.core.session.IoSession。该接口中比较重要的方法如 表 2 所示。

在介绍完 I/O 会话之后，下面介绍 I/O 过滤器。

表 2. IoSession 中的重要方法

jieforest

I/O 过滤器

从 I/O 服务发送过来的所有 I/O 事件和请求，在到达 I/O 处理器之前，会先由 I/O 过滤器链中的 I/O 过滤器进行处理。Apache MINA 中的过滤器与 Servlet 规范中的过滤器是类似的。过滤器可以在很多情况下使用，比如记录日志、性能分析、访问控制、负载均衡和消息转换等。过滤器非常适合满足网络应用中各种横切的非功能性需求。在一个基于 Apache MINA 的网络应用中，一般存在多个过滤器。这些过滤器互相串联，形成链条，称为过滤器链。每个过滤器依次对传入的 I/O 事件进行处理。当前过滤器完成处理之后，由过滤器链中的下一个过滤器继续处理。当前过滤器也可以不调用下一个过滤器，而提前结束，这样 I/O 事件就不会继续往后传递。比如负责用户认证的过滤器，如果遇到未认证的对等体发出的 I/O 事件，则会直接关闭连接。这可以保证这些事件不会通过此过滤器到达 I/O 处理器。

Apache MINA 中 I/O 过滤器都实现 org.apache.mina.core.filterchain.IoFilter接口。一般来说，不需要完整实现 IOFilter接口，只需要继承 Apache MINA 提供的适配器 org.apache.mina.core.filterchain.IoFilterAdapter，并覆写所需的事件过滤方法即可，其它方法的默认实现是不做任何处理，而直接把事件转发到下一个过滤器。

jieforest

IoFilter 接口详细说明

IoFilter接口提供了 15 个方法。这 15 个方法大致分成两类，一类是与过滤器的生命周期相关的，另外一类是用来过滤 I/O 事件的。第一类方法如 表 3 所示。

表 3. IoFilter 中与过滤器的生命周期相关的方法

jieforest

在 表 3 中给出的方法中，参数 parent 表示包含此过滤器的过滤器链，参数 name 表示过滤器的名称，参数 nextFilter 表示过滤器链中的下一个过滤器。

第二类方法如 表 4 所示。

表 4. IoFilter 中过滤 I/O 事件的方法

jieforest

对于 表 4 中给出的与 I/O 事件相关的方法，它们都有一个参数是 nextFilter，表示过滤器链中的下一个过滤器。如果当前过滤器完成处理之后，可以通过调用 nextFilter 中的方法，把 I/O 事件传递到下一个过滤器。如果当前过滤器不调用 nextFilter 中的方法的话，该 I/O 事件就不能继续往后传递。另外一个共同的参数是 session，用来表示当前的 I/O 会话，可以用来发送消息给对等体。下面通过具体的实例来说明过滤器的实现。

BlacklistFilter

BlacklistFilter是 Apache MINA 自带的一个过滤器实现，其功能是阻止来自特定地址的连接，即所谓的“黑名单”功能。BlacklistFilter继承自 IoFilterAdapter，并覆写了 IoHandler相关的方法。清单 4 中给出了部分实现。

清单 4. 阻止来自特定地址连接的 BlacklistFilterpublic void messageReceived(NextFilter nextFilter, IoSession session, Object message) {
    if (!isBlocked(session)) {
        nextFilter.messageReceived(session, message);
    } else {
        blockSession(session);
    }
}

private void blockSession(IoSession session) {
    session.close(true);
}



在 清单 4 中 messageReceived 方法的实现中，首先通过 isBlocked 来判断当前连接是否应该被阻止，如果不是的话，则通过 nextFilter.messageReceived 把该 I/O 事件传递到下一个过滤器；否则的话，则通过 blockSession 来阻止当前连接。

jieforest

使用 ProtocolCodecFilter

ProtocolCodecFilter 用来在字节流和消息对象之间互相转换。当该过滤器接收到字节流的时候，需要首先判断消息的边界，然后把表示一条消息的字节提取出来，通过一定的逻辑转换成消息对象，再把消息对象往后传递，交给 I/O 处理器来执行业务逻辑。这个过程称为“解码”。与“解码”对应的是“编码”过程。在“编码”的时候，过滤器接收到的是消息对象，通过与“解码”相反的逻辑，把消息对象转换成字节，并反向传递，交给 I/O 服务来执行 I/O 操作。

在“编码”和“解码”中的一个重要问题是如何在字节流中判断消息的边界。通常来说，有三种办法解决这个问题：
使用固定长度的消息。这种方式实现起来比较简单，只需要每次读取特定数量的字节即可。
使用固定长度的消息头来指明消息主体的长度。比如每个消息开始的 4 个字节的值表示了后面紧跟的消息主体的长度。只需要首先读取该长度，再读取指定数量的字节即可。
使用分隔符。消息之间通过特定模式的分隔符来分隔。每次只要遇到该模式的字节，就表示到了一个消息的末尾。

具体到示例应用来说，客户端和服务器之间的通信协议比较复杂，有不同种类的消息。每种消息的格式都不相同，同类消息的内容也不尽相同。因此，使用固定长度的消息头来指明消息主体的长度就成了最好的选择。

示例应用中的每种消息主体由两部分组成，第一部分是固定长度的消息类别名称，第二部分是每种消息的主体内容。图 4 中给出了示例应用中一条完整的消息的结构。

图 4. 示例应用中消息的结构

jieforest

AbstractTetrisCommand用来描述联机游戏示例应用中的消息。它是一个抽象类，是所有具体消息的基类。其具体实现如 清单 5 所示。

清单 5. 联机游戏示例应用中的消息 AbstractTetrisCommandpublic abstract class AbstractTetrisCommand implements TetrisCommand {        
    public abstract String getName();

    public abstract byte[] bodyToBytes() throws Exception;
       
    public abstract void bodyFromBytes(byte[] bytes) throws Exception;

    public byte[] toBytes() throws Exception {
        byte[] body = bodyToBytes();
        int commandNameLength = Constants.COMMAND_NAME_LENGTH;
        int len = commandNameLength + body.length;
        byte[] bytes = new byte[len];
        String name = StringUtils.rightPad(getName(), commandNameLength,
            Constants.COMMAND_NAME_PAD_CHAR);
        name = name.substring(0, commandNameLength);
        System.arraycopy(name.getBytes(), 0, bytes, 0, commandNameLength);
        System.arraycopy(body, 0, bytes, commandNameLength, body.length);
        return bytes;
    }
}

jieforest

如 清单 5 所示，AbstractTetrisCommand 中定义了 3 个抽象方法：getName、bodyToBytes 和 bodyFromBytes，分别用来获取消息的名称、把消息的主体转换成字节数组和从字节数组中构建消息。bodyToBytes对应于前面提到的“编码”过程，而 bodyFromBytes对应于“解码”过程。每种具体的消息都应该实现这 3 个方法。AbstractTetrisCommand 中的方法 toBytes 封装了把消息的主体转换成字节数组的逻辑，在字节数组中，首先是长度固定为 Constants.COMMAND_NAME_LENGTH的消息类别名称，紧接着是每种消息特定的主体内容，由 bodyToBytes 方法来生成。

在介绍完示例应用中的消息格式之后，下面将讨论具体的“编码”和“解码”过程。“编码”过程由编码器来完成，编码器需要实现 org.apache.mina.filter.codec.ProtocolEncoder 接口，一般来说继承自 org.apache.mina.filter.codec.ProtocolEncoderAdapter 并覆写所需的方法即可。清单 6 中给出了示例应用中消息编码器 CommandEncoder 的实现。

清单 6. 联机游戏示例应用中消息编码器 CommandEncoderpublic class CommandEncoder extends ProtocolEncoderAdapter {

    public void encode(IoSession session, Object message,
        ProtocolEncoderOutput out) throws Exception {
        AbstractTetrisCommand command = (AbstractTetrisCommand) message;
        byte[] bytes = command.toBytes();
        IoBuffer buf = IoBuffer.allocate(bytes.length, false);
               
        buf.setAutoExpand(true);
        buf.putInt(bytes.length);
        buf.put(bytes);
               
        buf.flip();
        out.write(buf);
    }
}

jieforest

在 清单 6 中，encode 方法封装了编码的逻辑。由于 AbstractTetrisCommand的 toBytes已经完成了到字节数组的转换，encode 方法直接使用即可。首先写入消息主体字节数组的长度，再是字节数组本身，就完成了编码的过程。

与编码过程相比，解码过程要相对复杂一些。具体的实现如 清单 7 所示。

清单 7. 联机游戏示例应用中消息解码器 CommandDecoderpublic class CommandDecoder extends CumulativeProtocolDecoder {

    protected boolean doDecode(IoSession session, IoBuffer in,
            ProtocolDecoderOutput out) throws Exception {
        if (in.prefixedDataAvailable(4, Constants.MAX_COMMAND_LENGTH)) {
            int length = in.getInt();
            byte[] bytes = new byte[length];
            in.get(bytes);
            int commandNameLength = Constants.COMMAND_NAME_LENGTH;
            byte[] cmdNameBytes = new byte[commandNameLength];
            System.arraycopy(bytes, 0, cmdNameBytes, 0, commandNameLength);
            String cmdName = StringUtils.trim(new String(cmdNameBytes));
            AbstractTetrisCommand command = TetrisCommandFactory
                .newCommand(cmdName);
            if (command != null) {
                byte[] cmdBodyBytes = new byte[length - commandNameLength];
                System.arraycopy(bytes, commandNameLength, cmdBodyBytes, 0,
                    length - commandNameLength);
                command.bodyFromBytes(cmdBodyBytes);
                out.write(command);
            }
            return true;
        } else {
            return false;
        }
    }
}

jieforest

在 清单 7 中可以看到，解码器 CommandDecoder 继承自 CumulativeProtocolDecoder。这是 Apache MINA 提供的一个帮助类，它会自动缓存所有已经接收到的数据，直到编码器认为可以开始进行编码。这样在实现自己的编码器的时候，就只需要考虑如何判断消息的边界即可。如果一条消息的后续数据还没有接收到，CumulativeProtocolDecoder会自动进行缓存。在之前提到过，解码过程的一个重要问题是判断消息的边界。对于固定长度的消息来说，只需要使用 Apache MINA 的 IoBuffer的 remaining方法来判断当前缓存中的字节数目，如果大于消息长度的话，就进行解码；对于使用固定长度消息头来指明消息主体的长度的情况，IoBuffer提供了 prefixedDataAvailable方法来满足这一需求。prefixedDataAvailable会检查当前缓存中是否有固定长度的消息头，并且由此消息头指定长度的消息主体是否已经全部在缓存中。如果这两个条件都满足的话，说明一条完整的消息已经接收到，可以进行解码了。解码的过程本身并不复杂，首先读取消息的类别名称，然后通过 TetrisCommandFactory.newCommand方法来生成一个该类消息的实例，接着通过该实例的 bodyFromBytes方法就可以从字节数组中恢复消息的内容，得到一个完整的消息对象。每次成功解码一个消息对象，需要调用 ProtocolDecoderOutput的 write把此消息对象往后传递。消息对象会通过过滤器链，最终达到 I/O 处理器，在 IoHandler.messageReceived中接收到此消息对象。如果当前缓存的数据不足以用来解码一条消息的话，doDecode只需要返回 false即可。接收到新的数据之后，doDecode会被再次调用。