Mina学习笔记二_基础

bsr1983

浏览: 1100479 次
性别:
来自: 北京

最近访客更多访客>>

zxpang

dyh_it

learning_han

lobskiwi

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Mina

第二章基础
在第一章中，我们已对MINA有了一个简要的了解，在本章中，我们将介绍客户端/服务器架构，并详细规划基于MINA的服务器和客户端。
我们也会介绍一下基于TCP和UDP的简单的服务器和客户端的例子。
1.MINA的基础应用架构
这是一个经常会问到的问题：MINA的基础应用时什么样子的，在这篇文章中让我们一起看看MINA基础应用的结构。我们会尝试从基于MINA的演示中收集信息。
一个鸟瞰图：

从这个图里面可以看到，MINA是你的应用（作为一个客户端或服务器）和下方的网络层之间的粘合剂，网络层可以是基于TCP、UDP、VM内部通讯或类似RS-232的串行通讯协议。
你仅需要基于MINA设计你的应用而不需要处理网络层的所有复杂性。
让我们进一步深入了解一下细节，下面的图片展示了一些MINA的内部结构，以及每个MINA的组件所做的事情：

概况来讲，基于MINA的应用被分为3层：
（1）I/O服务：进行实际的I/O操作
（2）I/O过滤器链：过滤/转换字节为所需的数据结构，及将数据结构转换为字节
（3）I/O 处理器：在这里实现实际的业务逻辑
所以，为了创建一个基于MINA的应用，你需要：
（1）创建一个I/O服务：选择一个已经可用的服务（*Acceptor）或创建一个你自己的
（2）创建一个过滤器链：选择一个已存在的过滤器或创建一个自定义的过滤器来转换请求及响应
（3）创建一个I/O处理器：编写业务逻辑，处理不同的消息。
差不多就是这样。
服务器架构
从根本上说，一个服务监听一个端口来获取到来的请求，处理他们，然后发送回复。他也能够创建并处理与每一个客户端的会话（无论是基于TCP的还是UDP的），这些将在第四章中做详细介绍。

（1）IOAcceptor监听网络获取到来的连接或数据包
（2）一个新的会话将为一个新的连接创建，随后的所有来自IP地址/端口的请求都将在这个会话中进行处理。
（3）一个会话所接收到的所有包，按照图示遍历过滤器链。过滤器可以被用来修改报的内容（例如转化成对象，增加、删除信息等）。从原始字节到高等级对象的互相转化、包编码、解码器是非常有用的。
（4）最终保或转换的对象加载到IOHandle中，IOHandle被用来实现业务需求。
会话创建
每当一个客户端连接到一个MINA服务器上时，我们将创建一个会话用于存储持久性数据。即使协议时非连接协议，这个会话也会被创建。
传入的消息处理
现在我们来解释MINA如何处理传入的消息。
如果一个会话已经建立，任何新传入的消息将唤醒一个选择器。
（2）客户端结构
客户端需要一个到服务器的连接，发送信息并处理回复。

（1）客户端首先创建一个IOConnector（MINA构造用于连接到Socket），启动一个与服务器的绑定。
（2）在连接创建后，一个会话被创建并关联到连接
（3）应用/客户端写入会话，在经过过滤器链转换后，作为结果数据被发送给服务器。
（4）所有来自服务器的回复/消息经过过滤器链转换后，加载到IOHandler等待处理。
TCP 服务器示例
本教程将指导您构建一个基于MINA的程序。这个教程将指导您构建一个时间服务器。下面是本教程所需的先决条件。
1）MINA 2.x Core
2）JDK5或以上版本
3）SLF4J1.3.0或以上版本
我们在Windows2000专业版和linux上进行了这个程序的测试。如果您在运行该程序的过程中有任何疑问，请不要犹豫，马上联系MINA的开发者。另外，这个教程也尝试着与开发环境独立（IDE，编辑器等等）。这个教程可在您喜欢的任何运行环境中执行。为了保持教程的简洁，编译命令和程序的执行步骤已经被移除了。如果你需要学习如何编译和运行java程序，请查阅java教程。
编写MINA时间服务器
创建一个名为MinaTimeServer.java的文件，初始的代码如下：

public class MinaTimeServer {
    public static void main(String[] args) {
        // code will go here next
    }
}

上面的代码很容易看懂，我们定义了一个用来启动应用的main方法。接着，我们将开始添加代码来晚上我们的服务器。首先我们需要一个用来监听进入的连接的对象，因为这个程序是基于TCP/IP的，所以我们添加一个SocketAcceptor 。

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    public static void main( String[] args )
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
    }
}

定义NioSocketAcceptor后，我们将定义一个处理器类并将NioSocketAcceptor绑定到一个端口：

import java.net.InetSocketAddress;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    private static final int PORT = 9123;
    public static void main( String[] args ) throws IOException
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

正如你所看到的，这里调用了acceptor.setLocalAddress( new InetSocketAddress(PORT) );这个方法的功能是定义服务器将要监听的主机和端口。 IoAcceptor.bind()方法将绑定一个指定的端口，然后启动对远程客户端的处理。
接下来我们将给服务器配置一个过滤器。这个过滤器将记录所有信息，包括会话创建，接受到的消息、发送的消息以及会话关闭。另一个过滤器是ProtocolCodecFilter，这个过滤器将二进制或协议特有的数据与消息对象做相互转换。我们使用现有的TextLine 工厂来处理基于文本的消息（我们不必编写编码器部分）。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    public static void main( String[] args )
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

接下来，我们将定义一个用来为客户端提供服务并获取当前时间的处理器，这个处理器是一个必须实现IoHandler接口的类。对于所有使用MINA的程序而言，这是程序的核心部分。因为他要为所有到来的请求提供服务。对于这篇教程而言，我们使用集成IoHandlerAdapter的方式实现。这个一个使用了适配器模式的类，这大大简化了一个实现IoHandler接口的类所必须编写的代码。

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    public static void main( String[] args ) throws IOException
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));
        acceptor.setHandler(  new TimeServerHandler() );
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

现在我们将处理器配置到NioSocketAcceptor ，这样我们就可以为那些被用来接受客户端请求的socket进行特殊设置。

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    public static void main( String[] args ) throws IOException
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));
        acceptor.setHandler(  new TimeServerHandler() );
        acceptor.getSessionConfig().setReadBufferSize( 2048 );
        acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10 );
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

在MinaTimeServer类中新增了两行代码。这两个方法用来设置处理器的输入缓冲区的大小和会话的空闲属性。设置缓冲区的大小是为了告知底层的操作系统为接收到的数据分配多少内存空间。第二行规定了对空闲会话的检测策略，setIdleTime方法的第一个参数定义了如何处理检测到的空闲会话，第二个参数定义了会话被视为是空闲时所需的时间，单位为秒。
处理器的代码如下：

import java.util.Date;

import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;

public class TimeServerHandler extends IoHandlerAdapter
{
    @Override
    public void exceptionCaught( IoSession session, Throwable cause ) throws Exception
    {
        cause.printStackTrace();
    }
    @Override
    public void messageReceived( IoSession session, Object message ) throws Exception
    {
        String str = message.toString();
        if( str.trim().equalsIgnoreCase("quit") ) {
            session.close();
            return;
        }
        Date date = new Date();
        session.write( date.toString() );
        System.out.println("Message written...");
    }
    @Override
    public void sessionIdle( IoSession session, IdleStatus status ) throws Exception
    {
        System.out.println( "IDLE " + session.getIdleCount( status ));
    }
}

这个类中用到了3个方法：exceptionCaught、 messageReceived 、 sessionIdle。exceptionCaught通常在处理器中实现用来处理由远端连接引起的一次，如果不定义这个方法，那么处理器可能不能合理报告异常。
exceptionCaught方法会简单地输出错误的堆栈信息并关闭会话。对于大多数程序而言，这是常规的处理方式，除非处理器可以从异常条件中恢复。
messageReceived方法被用来接收来自客户端的数据，并且将当前的时间返回给客户端。如果当前客户端发送的消息是”quit”，那么连接将会被关闭。这个方法也会输出当前日期到客户端。依赖于所使用的协议编码器，被传入到方法中的对象（第二个参数）将是不同的，就如同你写入到方法 session.write(Object)中的对象一样。如果你没有指定一个特殊的协议编码器，你非常可能接收到一个IoBuffer对象，而且也被要求输出一个IoBuffer对象。
sessionIdle方法将在一个会话空闲的时间达到acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10 );.方法中指定的时间时被调用。
接下来要做的就是定义服务器要监听的socket端口，然后启动服务器。具体的代码为：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

public class MinaTimeServer
{
    private static final int PORT = 9123;
    public static void main( String[] args ) throws IOException
    {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));
        acceptor.setHandler( new TimeServerHandler() );
        acceptor.getSessionConfig().setReadBufferSize( 2048 );
        acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10 );
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

试用时间服务器
接下来，我们编译这个程序，编译万仇，你就可以运行这个程序来测试发生了是吗。最简单的测试方法是启动程序，然后使用telnet。
TCP客户端示例
我们已经看过了客户端的架构，接下来看一下客户端的实现示例。
我们将Sumup 客户端作为参照实现。
我们删除了一些冗杂的代码，提取了重要的实现部分代码。

public static void main(String[] args) throws Throwable {
    NioSocketConnector connector = new NioSocketConnector();
    connector.setConnectTimeoutMillis(CONNECT_TIMEOUT);

    if (USE_CUSTOM_CODEC) {
    connector.getFilterChain().addLast("codec",
        new ProtocolCodecFilter(new SumUpProtocolCodecFactory(false)));
    } else {
        connector.getFilterChain().addLast("codec",
            new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
    }

    connector.getFilterChain().addLast("logger", new LoggingFilter());
    connector.setHandler(new ClientSessionHandler(values));
    IoSession session;

    for (;;) {
        try {
            ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT));
            future.awaitUninterruptibly();
            session = future.getSession();
            break;
        } catch (RuntimeIoException e) {
            System.err.println("Failed to connect.");
            e.printStackTrace();
            Thread.sleep(5000);
        }
    }

    // wait until the summation is done
    session.getCloseFuture().awaitUninterruptibly();
    connector.dispose();
}

为了构造一个客户端，我们需要以下步骤：
1）创建一个Connector
2）创建一个过滤器链
3）创建一个IO处理器并将其添加到Connector
4）绑定到服务器
让我们一起看一下具体步骤
创建一个Connector

	NioSocketConnector connector = new NioSocketConnector();

这里创建了一个非阻塞的Socket连接
创建一个过滤器链

if (USE_CUSTOM_CODEC) {
    connector.getFilterChain().addLast("codec",
        new ProtocolCodecFilter(new SumUpProtocolCodecFactory(false)));
} else {
    connector.getFilterChain().addLast("codec",
        new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
}

我们为Connector添加过滤器到过滤器链，在这我们添加了一个ProtocolCodec
创建IO处理器
connector.setHandler(new ClientSessionHandler(values));
我们创建了ClientSessionHandler的一个实例，并将其设置为Connector的处理器
绑定到服务器

IoSession session;

for (;;) {
    try {
        ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT));
        future.awaitUninterruptibly();
        session = future.getSession();
        break;
    } catch (RuntimeIoException e) {
        System.err.println("Failed to connect.");
        e.printStackTrace();
        Thread.sleep(5000);
    }
}

这里是最重要的部分。我们连接到了元朝服务，因此，连接是一个异步任务，我们使用ConnectFuture类来获知到连接完成。一旦连接完成，我们将获得相应的IoSession。我们给服务器的任何信息都通过会话来发送。所有来自服务器的消息和回复都将经过过滤器链的转换，最终在IoHandler中进行处理。
UDP服务器示例
我们将通过查看org.apache.mina.example.udp包中的代码开始我们的教程。为了保持教程的简洁性，我们只关注与MINA有关的部分。
为了构造一个服务器，我们需要做的是：
1.创建一个用来监听从客户端发送的请求的数据报Socket。（参照 MemoryMonitor.java)
2.创建一个IoHandler来处理MINA框架产生的事件（参照 MemoryMonitorHandler.java）
下面的代码片段是上述的第一点：

NioDatagramAcceptor acceptor = new NioDatagramAcceptor();
acceptor.setHandler(new MemoryMonitorHandler(this));

我们创建了一个NioDatagramAcceptor 来监听客户端发送的请求，并为其设置了IoHandler。变量‘PORT’只能是一个整数。下一步是给DatagramAcceptor添加一个日志过滤器到过滤器链。LoggingFilter 查看当前MINA运行状态的一个很好的途径。它在不同阶段产生MINA的状态日志，给我们提供了一个查看MINA的运行状态的途径。

DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
chain.addLast("logger", new LoggingFilter());

接着我们看一下UDP传输所特有的代码，我们对接收者进行设置，以便地址重用。

DatagramSessionConfig dcfg = acceptor.getSessionConfig();
dcfg.setReuseAddress(true);
acceptor.bind(new InetSocketAddress(PORT));

当然，最后需要做的是调用bind()来绑定端口。
IoHandler的实现
我们的服务实现的3个主要事件是
1.会话创建
2.接收到消息
3.会话关闭
会话创建事件

@Override
public void sessionCreated(IoSession session) throws Exception {
    SocketAddress remoteAddress = session.getRemoteAddress();
    server.addClient(remoteAddress);
}

在会话创建事件中，我们仅仅调用addClicent方法给UI添加一个标签页。
接收到消息事件

@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    if (message instanceof IoBuffer) {
        IoBuffer buffer = (IoBuffer) message;
        SocketAddress remoteAddress = session.getRemoteAddress();
        server.recvUpdate(remoteAddress, buffer.getLong());
    }
 }

在接收到消息事件中，我们仅仅转存接收到的消息中的数据，那些需要发送回复的应用，可以在这个方法中处理消息，并将回复写入到会话中。
会话关闭事件

@Override
public void sessionClosed(IoSession session) throws Exception {
    System.out.println("Session closed...");
    SocketAddress remoteAddress = session.getRemoteAddress();
    server.removeClient(remoteAddress);
}

在会话关闭事件中，我们仅仅从UI中移除了标签页。
UDP客户端示例
为了实现一个客户端所需要做的是：
1.创建Socket及到服务器的连接
2.设置Io处理器
3.收集空闲内存
4.发送数据给服务器
我们通过查看MemMonClient.java来开始我们的教程。这个类的代码可以在

org.apache.mina.example.udp.client包中找到，代码的开始部分是很简单明了的：
connector = new NioDatagramConnector();
connector.setHandler( this );
ConnectFuture connFuture = connector.connect( new InetSocketAddress("localhost", MemoryMonitor.PORT ));

这里创建了一个NioDatagramConnector，设置了Io处理器，然后连接到服务器。你必须在InetSocketAddress 对象中设置服务器主键地址，否则无法正常运行。这个例子通常是在Window xp机器上编写和测试的，在其他运行环境中可能有一些不一样的地方。接下来我们等待，确认客户端已经连接到服务器。一旦客户端连接到服务器，我们就开始向服务器发送数据。下面是具体代码：

connFuture.addListener( new IoFutureListener(){
            public void operationComplete(IoFuture future) {
                ConnectFuture connFuture = (ConnectFuture)future;
                if( connFuture.isConnected() ){
                    session = future.getSession();
                    try {
                        sendData();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    log.error("Not connected...exiting");
                }
            }
        });

在这里我们在ConnectFuture 对象上添加了一个监听器，当收到一个客户端已经连接的回调后，我们将开始发送数据。发送给服务器的数据将通过sendData方法来完成。具体代码为：

private void sendData() throws InterruptedException {
    for (int i = 0; i < 30; i++) {
        long free = Runtime.getRuntime().freeMemory();
        IoBuffer buffer = IoBuffer.allocate(8);
        buffer.putLong(free);
        buffer.flip();
        session.write(buffer);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new InterruptedException(e.getMessage());
        }
    }
}

这个方法将在30秒内以每秒钟一次的频率发送空闲内存的统计数给服务器。你可以看到这里给IoBuffer分配了一个足够存储一个长整型变量的空间来在缓存中存储的空闲内存数量。这个缓存将会被很快地清空并把数据发送给服务器。