Mina框架学习笔记（五）

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

清单 6. 联机游戏示例应用中消息编码器 CommandEncoder

public 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); 
    } 
}

 

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

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

清单 7. 联机游戏示例应用中消息解码器 CommandDecoder

public 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; 
        } 
    } 
}

 

在 清单 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会被再次调用。

过滤器链

过滤器只有在添加到过滤器链中的时候才起作用。过滤器链是过滤器的容器。过滤器链与 I/O 会话是一一对应的关系。org.apache.mina.core.filterchain.IoFilterChain是 Apache MINA 中过滤器链的接口，其中提供了一系列方法对其中包含的过滤器进行操作，包括查询、添加、删除和替换等。如 表 5 所示。

表 5. IoFilterChain 接口的方法

方法 说明

addFirst(String name, IoFilter filter)	将指定名称的过滤器添加到过滤器链的开头。
addLast(String name, IoFilter filter)	将指定名称的过滤器添加到过滤器链的末尾。
contains(String name)	判断过滤器链中是否包含指定名称的过滤器。
get(String name)	从过滤器链中获取指定名称的过滤器。
remove(String name)	从过滤器链中删除指定名称的过滤器。
replace(String name, IoFilter newFilter)	用过滤器 newFilter替换掉过滤器链中名为 name的过滤器。
getSession()	获取与过滤器链一一对应的 I/O 会话。

在介绍完 I/O 过滤器和过滤器链之后，下面介绍 I/O 处理器。

 

回页首

I/O 处理器

I/O 事件通过过滤器链之后会到达 I/O 处理器。I/O 处理器中与 I/O 事件对应的方法会被调用。Apache MINA 中 org.apache.mina.core.service.IoHandler是 I/O 处理器要实现的接口，一般情况下，只需要继承自 org.apache.mina.core.service.IoHandlerAdapter并覆写所需方法即可。IoHandler接口的方法如 表 6 所示。

表 6. IoHandler 接口的方法

方法 说明

sessionCreated(IoSession session)	当有新的连接建立的时候，该方法被调用。
sessionOpened(IoSession session)	当有新的连接打开的时候，该方法被调用。该方法在 sessionCreated之后被调用。
sessionClosed(IoSession session)	当连接被关闭的时候，此方法被调用。
sessionIdle(IoSession session, IdleStatus status)	当连接变成闲置状态的时候，此方法被调用。
exceptionCaught(IoSession session, Throwable cause)	当 I/O 处理器的实现或是 Apache MINA 中有异常抛出的时候，此方法被调用。
messageReceived(IoSession session, Object message)	当接收到新的消息的时候，此方法被调用。
messageSent(IoSession session, Object message)	当消息被成功发送出去的时候，此方法被调用。

对于 表 6 中的方法，有几个需要重点的说明一下。首先是 sessionCreated 和 sessionOpened 的区别。sessionCreated方法是由 I/O 处理线程来调用的，而 sessionOpened是由其它线程来调用的。因此从性能方面考虑，不要在 sessionCreated 方法中执行过多的操作。对于 sessionIdle，默认情况下，闲置时间设置是禁用的，也就是说sessionIdle 并不会被调用。可以通过 IoSessionConfig.setIdleTime(IdleStatus, int) 来进行设置。

Apache MINA 中的基本概念已经介绍完了，下面介绍状态机的使用。

 

回页首

使用状态机

在 I/O 处理器中实现业务逻辑的时候，对于简单的情况，一般只需要在 messageReceived 方法中对传入的消息进行处理。如果需要写回数据到对等体的话，用IoSession.write 方法即可。在另外的一些情况下，客户端和服务器端的通信协议比较复杂，客户端其实是有状态变迁的。这个时候可以用 Apache MINA 提供的状态机实现，可以使得 I/O 处理器的实现更加简单。

状态机中两个重要的元素是状态以及状态之间的迁移。示例应用中客户端的状态以及迁移如 图 5 所示。

图 5. 联机游戏示例应用中客户端的状态以及迁移

客户端初始化的时候，其状态为“未连接”，表示客户端还没有在服务器上面注册，此时还不能进行游戏；接着用户需要输入一个昵称来注册到服务器上面，完成之后状态迁移到“闲置”。此时客户端会接收到当前在线的所有其它用户的列表。当前用户可以邀请其它用户和他一块游戏，也可以接收来自其它用户的邀请。邀请发送出去之后，客户端的状态迁移到“邀请已发送”。如果接受了其它用户的邀请，客户端的状态迁移到“邀请已接收”。如果某个用户的邀请被另外一个用户接受的话，两个客户端的状态都会迁移到“游戏中”。

要实现这样较为复杂的状态机的话，只需要在 I/O 处理器中以声明式的方式定义状态和迁移条件就可以了。首先需要声明状态机中状态，如 清单 8 所示。

清单 8. 联机游戏示例应用中的状态声明

@State public static final String ROOT = "Root"; 
@State(ROOT) public static final String NOT_CONNECTED = "NotConnected"; 
@State(ROOT) public static final String IDLE = "Idle"; 
@State(ROOT) public static final String INVITATION_SENT = "InvitationSent"; 
@State(ROOT) public static final String INVITATION_ACCEPTED = "InvitationAccepted"; 
@State(ROOT) public static final String PLAYING = "Playing"; 

 

如 清单 8 所示，上面定义了一共六个状态。通过标注 @State就声明了一个状态。需要注意的是状态之间是可以继承的。如果状态机接收到一个事件的时候，在当前状态中找不到对应的迁移，就会在其父状态上继续查找。状态的继承在某些情况下是很有用的，比如希望为所有的状态都增加同样的迁移逻辑，就可以直接把迁移条件添加在父状态上面。一个典型的场景就是错误处理，一般来说，所有的状态都需要错误处理，而错误处理的逻辑一般都是相同的。把发生错误时候的迁移放在父状态中，可以简洁的描述这一场景。

定义了状态之后，下面应该声明状态之间的迁移。如 清单 9 所示。

清单 9. 联机游戏示例应用中的状态迁移声明

@IoHandlerTransition(on = MESSAGE_RECEIVED, in = NOT_CONNECTED, next = IDLE) 
public void login(TetrisServerContext context, IoSession session, LoginCommand cmd) { 
    String nickName = cmd.getNickName(); 
    context.nickName = nickName; 
    session.setAttribute("nickname", nickName); 
    session.setAttribute("status", UserStatus.IDLE); 
    sessions.add(session); 
    users.add(nickName); 
		
    RefreshPlayersListCommand command = createRefreshPlayersListCommand(); 
    broadcast(command); 
} 
  
@IoHandlerTransition(on = EXCEPTION_CAUGHT, in = ROOT, weight = 10) 
public void exceptionCaught(IoSession session, Exception e) { 
    LOGGER.warn("Unexpected error.", e); 
    session.close(true); 
} 

@IoHandlerTransition(in = ROOT, weight = 100) 
public void unhandledEvent() { 
    LOGGER.warn("Unhandled event."); 
} 

 

清单 9 中，使用标注 @IoHandlerTransition声明了一个状态迁移。每个状态迁移可以有四个属性：on、in、next和 weight，其中属性 in是必须的，其余是可选的。属性on表示触发此状态迁移的事件名称，如果省略该属性的话，则默认为匹配所有事件的通配符。该属性的值可以是表中给出的 I/O 处理器中能处理的七种事件类型。属性 in表示状态迁移的起始状态。属性 next表示状态迁移的结束状态，如果省略该属性的话，则默认为表示当前状态 的 _self_。属性 weight用来指明状态迁移的权重。一个状态的所有迁移是按照其权重升序排列的。对于当前状态，如果有多个可能的迁移，排序靠前的迁移将会发生。代码中的第一个标注声明了如果当前状态是“未连接”，并且接收到了 MESSAGE_RECEIVED事件，而且消息的内容是一个 LoginCommand对象的话，login方法会被调用，调用完成之后，当前状态迁移到“闲置”。第二个标注声明了对于任何的状态，如果接收到了 EXCEPTION_CAUGHT事件，exceptionCaught方法会被调用。最后一个标注声明了一个状态迁移，其起始状态是 ROOT，表示该迁移对所有的事件都起作用。不过它的 weight是 100，优先级比较低。该状态迁移的作用是处理其它没有对应状态迁移的事件。

使用了 Apache MINA 提供的状态机之后，创建 I/O 处理器的方式也发生了变化。I/O 处理器的实例由状态机来创建，如 清单 10 所示。

清单 10. 在状态机中创建 I/O 处理器

private static IoHandler createIoHandler() { 
    StateMachine sm = StateMachineFactory.getInstance( 
        IoHandlerTransition.class).create(ServerHandler.NOT_CONNECTED, 
        new ServerHandler()); 
    return new StateMachineProxyBuilder().setStateContextLookup( 
        new IoSessionStateContextLookup(new StateContextFactory() { 
            public StateContext create() { 
                return new ServerHandler.TetrisServerContext(); 
            } 
    })).create(IoHandler.class, sm); 
}

 

在 清单 10 中，TetrisServerContext是提供给状态机的上下文对象，用来在状态之间共享数据。当然用 IoSession也是可以实现的，不过上下文对象的好处是类型安全，不需要做额外的类型转换。

在介绍完状态机后，下面介绍一些高级话题，包括异步操作以及 Apache MINA 与 JMX 和 Spring 的集成。