简单之美 | Apache Mina通信框架架构与应用

Apache Mina Server 是一个网络通信应用框架，也就是说，它主要是对基于 TCP/IP、UDP/IP协议栈的通信框架（当然，也可以提供 JAVA 对象的序列化服务、虚拟机管道通信服务等），Mina 可以帮助我们快速开发高性能、高扩展性的网络通信应用，Mina 提供了事件驱动、异步（Mina 的异步 IO 默认使用的是 JAVA NIO 作为底层支持）操作的编程模型。
从官网文档“MINA based Application Architecture”中可以看到Mina作为一个通信层框架，在实际应用所处的位置，如图所示：
apparch_small
Mina位于用户应用程序和底层Java网络API（和in-VM通信）之间，我们开发基于Mina的网络应用程序，就无需关心复杂的通信细节。

应用整体架构

再看一下，Mina提供的基本组件，如图所示：
mina_app_arch
也就是说，无论是客户端还是服务端，使用Mina框架实现通信的逻辑分层在概念上统一的，即包含如下三层：

I/O Service – Performs actual I/O
I/O Filter Chain – Filters/Transforms bytes into desired Data Structures and vice-versa
I/O Handler – Here resides the actual business logic

想要开发基于MIna的应用程序，你只需要做如下事情：

Create an I/O service – Choose from already available Services (*Acceptor) or create your own
Create a Filter Chain – Choose from already existing Filters or create a custom Filter for transforming request/response
Create an I/O Handler – Write business logic, on handling different messages

下面看一下使用Mina的应用程序，在服务器端和客户端的架构细节：

服务器端架构
服务器端监听指定端口上到来的请求，对这些请求经过处理后，回复响应。它也会创建并处理一个链接过来的客户会话对象（Session）。服务器端架构如图所示：
Server_arch
对服务器端的说明，引用官网文档，如下所示：

IOAcceptor listens on the network for incoming connections/packets
For a new connection, a new session is created and all subsequent request from IP Address/Port combination are handled in that Session
All packets received for a Session, traverses the Filter Chain as specified in the diagram. Filters can be used to modify the content of packets (like converting to Objects, adding/removing information etc). For converting to/from raw bytes to High Level Objects, PacketEncoder/Decoder are particularly useful
Finally the packet or converted object lands in IOHandler. IOHandlers can be used to fulfill business needs.

客户端架构
客户端主要做了如下工作：

连接到服务器端
向服务器发送消息
等待服务器端响应，并处理响应

客户端架构，如图所示：
clientdiagram
对客户端架构的说明，引用官网文档内容，如下所示：

Client first creates an IOConnector (MINA Construct for connecting to Socket), initiates a bind with Server
Upon Connection creation, a Session is created and is associated with Connection
Application/Client writes to the Session, resulting in data being sent to Server, after traversing the Filter Chain
All the responses/messages received from Server are traverses the Filter Chain and lands at IOHandler, for processing

应用实例开发

下面根据上面给出的架构设计描述，看一下Mina（版本2.0.7）自带的例子，如何实现一个简单的C/S通信的程序，非常容易。
服务端
首先，服务器端需要使用的组件有IoAdaptor、IoHandler、IoFilter，其中IoFilter可选.
我们基于Mina自带的例子进行了简单地修改，实现服务端IoHandler的代码如下所示：

package org.shirdrn.mina.server;

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

public class TinyServerProtocolHandler extends IoHandlerAdapter {
    private final static Logger LOGGER = LoggerFactory.getLogger(TinyServerProtocolHandler.class);
   
    @Override
    public void sessionCreated(IoSession session) {
        session.getConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        LOGGER.info("CLOSED");
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {
        LOGGER.info("OPENED");
    }

    @Override
    public void sessionIdle(IoSession session, IdleStatus status) {
        LOGGER.info("*** IDLE #" + session.getIdleCount(IdleStatus.BOTH_IDLE) + " ***");
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) {
        session.close(true);
    }

    @Override
    public void messageReceived(IoSession session, Object message)
            throws Exception {
        LOGGER.info( "Received : " + message );
       if(!session.isConnected()) {
            session.close(true);
       }
    }
}

这个版本中，IoHandlerAdapter实现了IoHandler接口，里面封装了一组用于事件处理的空方法，其中包含服务端和客户端的事件。在实际应用中，客户端可以选择客户端具有的事件，服务器端选择服务器端具有的事件，然后分别对这两类事件进行处理（有重叠的事件，如连接事件、关闭事件、异常事件等）。
客户端的IoHandler的具体实现也是类似的，不过多累述。
下面看启动服务器的主方法类，代码如下所示：

package org.shirdrn.mina.server;

import java.net.InetSocketAddress;

import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TinyMinaServer {

	private final static Logger LOG = LoggerFactory.getLogger(TinyMinaServer.class);
	/** Choose your favorite port number. */
	private static final int PORT = 8080;

	public static void main(String[] args) throws Exception {
		SocketAcceptor acceptor = new NioSocketAcceptor();
		acceptor.setReuseAddress(true);
		acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));

		// Bind
		acceptor.setHandler(new TinyServerProtocolHandler());
		acceptor.bind(new InetSocketAddress(PORT));
		LOG.info("Listening on port " + PORT);

		LOG.info("Server started!");

		for (;;) {
			LOG.info("R: " + acceptor.getStatistics().getReadBytesThroughput() + ", W: " + acceptor.getStatistics().getWrittenBytesThroughput());
			Thread.sleep(3000);
		}
	}

}

客户端
实现客户端IoHandler的代码如下所示：

package org.shirdrn.mina.client;

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

public class TinyClientProtocolHandler extends IoHandlerAdapter {

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

	@Override
	public void sessionCreated(IoSession session) {
		LOGGER.info("CLIENT::CREATED");
	}

	@Override
	public void sessionClosed(IoSession session) throws Exception {
		LOGGER.info("CLIENT::CLOSED");
	}

	@Override
	public void sessionOpened(IoSession session) throws Exception {
		LOGGER.info("CLIENT::OPENED");
	}

	@Override
	public void sessionIdle(IoSession session, IdleStatus status) {
		LOGGER.info("CLIENT::*** IDLE #"
				+ session.getIdleCount(IdleStatus.BOTH_IDLE) + " ***");
	}

	@Override
	public void exceptionCaught(IoSession session, Throwable cause) {
		LOGGER.info("CLIENT::EXCEPTIONCAUGHT");
		cause.printStackTrace();
	}

	public void messageSent(IoSession session, Object message) throws Exception {
		LOGGER.info("CLIENT::MESSAGESENT: " + message);
	}
}

下面看启动客户端的主方法类，代码如下所示：

package org.shirdrn.mina.client;

import java.net.InetSocketAddress;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.SocketConnector;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TinyMinaClient {

	private final static Logger LOG = LoggerFactory.getLogger(TinyMinaClient.class);
	/** Choose your favorite port number. */
	private static final int PORT = 8080;

	public static void main(String[] args) throws Exception {
		SocketConnector connector = new NioSocketConnector();

		// Connect
		connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));
		connector.setHandler(new TinyClientProtocolHandler());

		for (int i = 0; i < 10; i++) {
			ConnectFuture future = connector.connect(new InetSocketAddress(PORT));
			LOG.info("Connect to port " + PORT);
			future.awaitUninterruptibly();
			future.getSession().write(String.valueOf(i));
			Thread.sleep(1500);
		}

	}
}

我们只是发送了十个数字，每发一次间隔1500ms。
测试上述服务器端与客户端交互，首先启动服务器端，监听8080端口。
接着启动客户端，连接到服务器端8080端口，然后发送消息，服务器端接收到消息后，直接将到客户端的连接关闭掉。

参考链接

参考文档

《Apache_Mina_Server_2.0中文参考手册V1.0.pdf》

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布，欢迎转载、使用、重新发布，但务必保留文章署名时延军（包含链接：http://shiyanjun.cn），不得用于商业目的，基于本文修改后的作品务必以相同的许可发布。如有任何疑问，请与我联系。

发表评论取消回复

丘比特: 请问博主，如果在窗口中用到广播状态，现在您有什么实现方案吗？
z: 寫的真好
方俊: 大佬好有耐心，从14年回复到19年哈哈
Yanjun: 图是用 Astah 和 OminiGraffle 画的
JacobZheng: 问个题外话，图是用什么工具画的啊
Derek Dekker: 感觉还挺难的
luosijie: 博主你好，请问您知道K距离方法出自哪篇文献吗，我该如何引用？
lee: 你好，点赞
郑威: 顺便咨询一下博主，我考虑在processfunction中注册定时器，来替代timewindow；connect上...
郑威: 非常感谢，百度了好久，只有您的文章，可以完全帮助我。特意评论感谢
rentao: 我测试过在Drvier端创建pool对象然后进行广播，但在e xcutor中通过broadcast.value获取的...
rentao: 很好的文章。你在第一个例子中使用单例对象来定义redisPo ol，而scala中单例对象是具有懒加载属性的，我理解...

发表评论 取消回复

发表评论取消回复