最近自己在工作之余做一个金融类的项目（类似股票），采用mina开源框架进行服务端和客户端之间交互。开始学学mina。转入正题...
（1）首先需要引入以下几个包：
mina-core-2.0.x.jar，slf4j-api-1.6.x.jar，slf4j-log4j12-1.6.x.jar
这里需要注意下：slf4j-api-1.6.x.jar，slf4j-log4j12-1.6.x.jar 必须是同样二级版本号，即1.6版本，如果一个是1.5版本、一个是1.6版本，运行时会报错：java.lang.NoSuchMethodError: org.slf4j.helpers.MessageFormatter.format，原因是包冲突。
（2）创建一个主服务类：TradeSocket
public class TradeSocket extends SocketBase implements ISocket {
.....
@Override
public boolean initSocket(Config config) {
logger.info("正在初始化 tradeSocket");
TradeSocketConfig tradeConfig = (TradeSocketConfig) config;
this.setSocket_IP(tradeConfig.getIp());
logger.info("正在初始化 tradeSocket的Ip");
this.setSocket_port(tradeConfig.getPost());
logger.info("正在初始化 tradeSocket的端口");
this.setHandlerAdapter(new TradeSocketHandler());
logger.info("正在初始化 tradeSocket的处理器");
this.setAcceptor(new NioSocketAcceptor());
logger.info("正在初始化 tradeSocket的接收器");
logger.info("成功初始化 tradeSocket");
logger.info("tradeSocket服务IP:" + this.getSocket_IP());
logger.info("tradeSocket服务端口号:" + this.getSocket_port());
return true;
}
@Override
public boolean startSocket() {
boolean result = false;
if (this.isStart() == false) {
initSocket(this.tradeSocketConfig);
// 设置过滤器，ObjectSerializationCodecFactory可以对象序列化方式在服务端和客户端之间传递数据
this.getAcceptor().getFilterChain().addLast(DGConstants.SERVER_SOCKET_FILTER_CHIN, new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
// 设置读取数据的缓冲区大小
this.getAcceptor().getSessionConfig().setReadBufferSize(2048);
// 设置长连接
this.getAcceptor().getSessionConfig().setKeepAlive(true);
// 读写通道10秒内无操作进入空闲状态
this.getAcceptor().getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
this.getAcceptor().setHandler(this.getHandlerAdapter());
try {
this.getAcceptor().bind(new InetSocketAddress(this.getSocket_IP(), this.getSocket_port()));
this.setStart(true);
result = true;
logger.info("成功启动 tradeSocket");
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
return result;
}
@Override
public boolean closeSocket() {
this.getAcceptor().dispose();
this.setStart(false);
logger.error("关闭了交易Socket");
return false;
}
（3）创建一个handler来管理事件，是业务处理关注点
public class TradeSocketHandler extends IoHandlerAdapter {
public static Logger logger = Logger.getLogger(TradeSocketHandler.class);
@Override
public void sessionCreated(IoSession session) throws Exception {
super.sessionCreated(session);
logger.info("服务端与客户端创建连接...");
}
@Override
public void sessionOpened(IoSession session) throws Exception {
InetSocketAddress remoteAddress = (InetSocketAddress) session.getRemoteAddress();
String clientIp = remoteAddress.getAddress().getHostAddress();
logger.info("服务器打开Session ID=" + session.getId());
logger.info( "服务端与客户端["+session.getRemoteAddress()+"]连接打开...");
// ConnectResponseMessage connectResponseMessage = new ConnectResponseMessage();
// connectResponseMessage.setConnectResponse(ConnectResponseMessage.ConnectSuccess);
// session.write(connectResponseMessage);
}
@Override
public void sessionClosed(IoSession session) throws Exception {
super.sessionClosed(session);
logger.info("服务端与客户端断开连接");
}
@Override
public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
super.sessionIdle(session, status);
logger.info("服务端进入空闲状态...");
}
@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
logger.error(cause.getMessage(), cause);
super.exceptionCaught(session, cause);
}
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
if (message instanceof BaseMessage) {
BaseMessage baseMessage = (BaseMessage) message;
logger.info("服务器收到的消息为：" + baseMessage);
if (baseMessage.getMsgType() == MessageType.LoginMessage_Type) {
LoginMessage loginMessage = (LoginMessage) baseMessage;
UserConnect userConnect = new UserConnect();
userConnect.setTradeSession(session);
loginMessage.setUserid(loginMessage.getUserid() + session.hashCode());
ServerCache cache = ServerCache.getInstance();
cache.addUserConnect(loginMessage.getUserid(), userConnect);
userConnect.getTradeSession().setAttribute("userID", loginMessage.getUserid());
userConnect.getTradeSession().setAttribute("sessionID", session.hashCode());
//这里主要是处理各种业务
ServerReceivedMessageProcess.getInstance().processMessage(loginMessage);
}
}
}
@Override
public void messageSent(IoSession session, Object message) throws Exception {
super.messageSent(session, message);
}
ServerReceivedMessageProcess.getInstance().processMessage(.....);方法是根据从客户端传过来不同message，实例化出不同处理类。
（4）创建客户端主类
public class ClientTradeSocket extends ClientSocketBase {
private static Logger logger = Logger.getLogger(ClientTradeSocket.class);
private static ClientTradeSocket instance;
public static ClientTradeSocket getInstance() {
if (instance == null) {
instance = new ClientTradeSocket();
}
return instance;
}
private ClientTradeSocket() {
}
@Override
public boolean closedSocket() {
this.getConnector().dispose();
this.setStart(false);
logger.info("客户关闭了与服务器的连接");
return true;
}
@Override
public void initSocket() {
this.setHandlerAdapter(new ClientTradeHandler());
this.setConnector(new NioSocketConnector());
logger.info("客户初始化了与服务器的连接");
}
@Override
public void startSocket() {
this.initSocket();
try {
if (this.isStart() == false) {
/**
* 创建接收数据的过滤器
*/
DefaultIoFilterChainBuilder chain = this.getConnector().getFilterChain();
chain.addLast(DGConstants.SERVER_SOCKET_FILTER_CHIN, new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
this.getConnector().setHandler(this.getHandlerAdapter());
/**
* 设置连接超时的时间 为：一分钟 这个超时不可以过小
*/
this.getConnector().setConnectTimeoutMillis(ClientSocketBase.timeOutMillis);
logger.info("ip/port:" + this.getServer_IP_Port() + "/" + this.getServer_Socket_Port());
InetSocketAddress socketAddress = new InetSocketAddress(this.getServer_IP_Port(), this.getServer_Socket_Port());
/**
* 建立连接
*/
this.connectFuture = this.getConnector().connect(socketAddress);
//等待连接，这句话比较重要，如果没有，则后面的this.connectFuture.isConnected()为false的。
//原因在于：connect方法是异步的，awaitUninterruptibly方式阻塞主线程，等待服务端返回，从而实现同步
//如果不用awaitUninterruptibly，可以采用this.connectFuture.addListener方法进行回调
this.connectFuture.awaitUninterruptibly();
//连接后，获得session
if (this.connectFuture.isConnected()) {
logger.info("已经建立用户连接");
this.setStart(true);
logger.info("客户开启了与服务器的连接");
ConnectSession session = ConnectSession.getInstance();
//将session保存到缓存中
session.setTradeSession(this.connectFuture.getSession());
} else {
logger.info("不能建立用户连接");
this.setStart(false);
ConnectSession session = ConnectSession.getInstance();
session.setTradeSession(null);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
this.setStart(false);
}
}
}
（5）客户端业务处理类handler
public class ClientTradeHandler extends IoHandlerAdapter {
private static Logger logger = Logger.getLogger(ClientTradeHandler.class);
private String user;
public void setUser(String user) {
this.user = user;
}
public String getUser() {
return this.user;
}
/**
*
* 当接口中其他方法抛出异常未被捕获时触发此方法
*/
@Override
public void exceptionCaught(IoSession ioSession, Throwable throwable) throws Exception {
}
/**
* 当接收到消息后被触发
*
*/
@Override
public void messageReceived(IoSession ioSession, Object msg) throws Exception {
BaseMessage message = (BaseMessage) msg;
logger.info("接受到得消息为：" + message);
if(message!=null) {
if(message instanceof ConnectResponseMessage) {
ConnectResponseMessage connectResponseMessage = (ConnectResponseMessage)message;
if(connectResponseMessage.getConnectResponse() == ConnectResponseMessage.ConnectSuccess) {
ConnectSession clientConnect = ConnectSession.getInstance();
clientConnect.setTradeSession(ioSession);
}
} else {
//根据服务端返回的message，实例化不同处理类
ClientReceivedMessageProcess messageProcess = new ClientReceivedMessageProcess();
messageProcess.processMessage(message);
}
}
}
/**
* 当发送消息后被触发
*
*/
@Override
public void messageSent(IoSession ioSession, Object msg) throws Exception {
}
/**
* 当会话关闭时被触发
*
*/
@Override
public void sessionClosed(IoSession ioSession) throws Exception {
logger.info("与服务器断开连接！！");
// EIMTrayIcon trayIcon = EIMTrayIcon.getInStance();
// trayIcon.showIcon(EIMClientConfig.OffLineTryIcon_Type);
}
/**
* 当会话创建时被触发
*
*/
@Override
public void sessionCreated(IoSession ioSession) throws Exception {
}
/**
*
* 当会话空闲时被触发
*/
@Override
public void sessionIdle(IoSession ioSession, IdleStatus msg) throws Exception {
}
/**
*
* 当会话开始时被触发
*/
@Override
public void sessionOpened(IoSession ioSession) throws Exception {
logger.info("会话已经打开");
}
}
最后，可以进行运行测试了。。。。
相关推荐
mina的入门例子
pache Mina Server 是一个网络通信应用框架，也就是说，它主要是对基于TCP/IP、UDP/IP协议栈的通信框架（然，也可以提供JAVA 对象的序列化服务、虚拟机管道通信服务等），Mina 可以帮助我们快速开发高性能、高扩展性...
mina 基础的入门实例，使用maven+jdk1.7进行构建，使用客户端直接可以进行测试
MINA2.0入门教程（适合刚接触的人）
这个网上一个牛人写的mina入门知识， 里面有基础介绍和项目开发步骤。 入门还是不错的，写的挺详细。 我把项目也加进来了。
该文档较好的介绍了，apache mina框架的使用方法。
NULL 博文链接：https://mina-java.iteye.com/blog/1115429
Mina2.0快速入门与源码剖析（个人经验文档）
ApacheMina入门 ApacheMina入门 ApacheMina入门 ApacheMina入门 ApacheMina入门
mina的使用初步入门mina的使用初步入门mina的使用初步入门
Apache_MINA_快速入门,开发中用到，自己想学的，可以看看。
mina框架的demo 入门，开发 学习
Mina 2.0快速入门与源码解析 很适合初学者使用!!!
里面包含mina2.0的api（英文）和mina自学手册，还有mina的开发指导
mina 入门 教程
mina-core-2.0.0-M6.jar mina-example-2.0.0-M6.jar mina-filter-codec-netty-2.0.0-M6.jar mina-filter-compression-2.0.0-M6.jar mina-integration-beans-2.0.0-M6.jar mina-integration-jmx-2.0.0-M6.jar mina-...
mina连接，mina心跳连接，mina断线重连。其中客户端可直接用在android上。根据各方参考资料，经过自己的理解弄出来的。CSDN的资源分太难得了。
Mina开发之客户端的代码，详情请查看：http://www.cnblogs.com/getherBlog/p/3937196.html
Mina开发之服务器的代码，详情请查看：http://www.cnblogs.com/getherBlog/p/3937196.html Mina开发之客户端的代码，详情请查看：http://www.cnblogs.com/getherBlog/p/3937196.html