Mina自定义协议简单实现
因公司需要做个电子秤自动称重系统，需要自定义协议实现，所以就用Mina简单实现了一下，有时间改成Netty版
服务端
package net.heartma.server;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.logging.Logger;
import net.heartma.protocol.CustomProtocolCodecFactory;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.SocketSessionConfig;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
/**
* 服务端
* @author heartma
*
*/
public class BalanceServer {
private static SocketAcceptor acceptor;
private static DefaultIoFilterChainBuilder filter;
private static SocketSessionConfig config;
public static Logger logger = Logger.getLogger("BalanceServer");
private static Executor threadPool = Executors.newFixedThreadPool(20);
public static void main(String[] args) {
// 1、创建服务器端监听
acceptor = new NioSocketAcceptor();
// 2、添加日志过滤和编码过滤
filter = acceptor.getFilterChain();
filter.addLast("threadPool", new ExecutorFilter(threadPool));
filter.addLast("logger", new LoggingFilter());
// ----编码过滤：将二进制或者协议相关数据转换成一个对象。TextLine工厂类可以处理基于文字的信息
filter.addLast("codec", new ProtocolCodecFilter(new CustomProtocolCodecFactory()));
// 3、绑定handler到acceptor
acceptor.setHandler(new BalanceServerHandler());
// 4、设置socket属性
// 获取socket的连接参数
config = acceptor.getSessionConfig();
// 设置socket的缓冲区大小为2M
config.setReadBufferSize(2048);
/**
* @params IdleStatus arg0 :在未成为idle状态前应该关心的状态(READ_IDLE或者WRITE_IDLE)
* @params @params IdleStatus arg1 : 变成IDLE状态所需要的时间(超时时间)
*
* 如果session持续idle的时间等于arg1时,将会触发handler中的sessionIdle方法
*/
// 设置空闲状态持续时间：1、这里的状态可以自己设置成只为读取设置空闲状态持续时间,只为写入设置空闲状态等待时间,或者为两者都设置空闲状态等待时间。后面的时间是两次触发handler中的sessionIdel方法的间隔时间。
config.setIdleTime(IdleStatus.BOTH_IDLE, 10);
try {
// 为服务器socket绑定端口
acceptor.bind(new InetSocketAddress(8081));
logger.info("服务已经启动... ...");
} catch (IOException e) {
logger.info("服务启动异常 ：");
e.printStackTrace();
}
}
}
服务端Handler
package net.heartma.server;
import net.heartma.pojo.Message;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
/**
* 服务端Handler
* @author heartma
*
*/
public class BalanceServerHandler extends IoHandlerAdapter {
@Override
public void exceptionCaught(IoSession ioSession, Throwable e)
throws Exception {
System.out.println("exceptionCaught");
}
@Override
public void messageReceived(IoSession ioSession, Object obj) throws Exception {
System.out.println("messageReceived");
Message message = (Message)obj;
System.out.println("服务器端接收到的消息：" +message);
}
@Override
public void messageSent(IoSession ioSession, Object obj) throws Exception {
System.out.println("messageSent");
}
@Override
public void sessionClosed(IoSession ioSession) throws Exception {
System.out.println("sessionClosed");
}
@Override
public void sessionCreated(IoSession ioSession) throws Exception {
System.out.println("sessionCreated");
}
@Override
public void sessionIdle(IoSession ioSession, IdleStatus arg1) throws Exception {
System.out.println("sessionIdle");
}
@Override
public void sessionOpened(IoSession ioSession) throws Exception {
System.out.println("sessionOpened");
}
}
客户端
23230E363932383834363630333738370D303030302E3434 为定义的协议报文
package net.heartma.client;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import net.heartma.protocol.CustomProtocolCodecFactory;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
/**
* Mina客户端
* @author heartma
*
*/
public class BalanceClient {
public static void main(String[] args) {
// 创建客户端连接器 基于tcp/ip
NioSocketConnector connector = new NioSocketConnector();
// 连接的地址和端口
SocketAddress address = new InetSocketAddress("localhost",8081);
// 获取过滤器链
DefaultIoFilterChainBuilder chain = connector.getFilterChain();
// 配置日志过滤器和自定义编解码器
chain.addLast("logger", new LoggingFilter());
chain.addLast("mycodec",new ProtocolCodecFilter(new CustomProtocolCodecFactory()));
// 添加处理器
connector.setHandler(new BalanceClientHandler());
// 连接到服务器
ConnectFuture future = connector.connect(address);
// 等待连接创建完成
future.awaitUninterruptibly();
// 会话创建后发送消息到服务器
future.getSession().write("23230E363932383834363630333738370D303030302E3434");
// 等待28000毫秒后连接断开
future.getSession().getCloseFuture().awaitUninterruptibly(28000);
// 关闭连接
connector.dispose();
}
}
客户端Handler
package net.heartma.client;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
/**
* 客户端Handler
* @author heartma
*
*/
public class BalanceClientHandler extends IoHandlerAdapter {
public void exceptionCaught(IoSession ioSession, Throwable e)
throws Exception {
System.out.println("exceptionCaught");
}
public void messageReceived(IoSession ioSession, Object obj) throws Exception {
System.out.println("messageReceived");
}
public void messageSent(IoSession ioSession, Object obj) throws Exception {
System.out.println("客户端发送消息...");
//super.messageSent(ioSession, obj);
}
public void sessionClosed(IoSession ioSession) throws Exception {
System.out.println("sessionClosed");
}
public void sessionCreated(IoSession ioSession) throws Exception {
System.out.println("sessionCreated");
}
public void sessionIdle(IoSession ioSession, IdleStatus idle) throws Exception {
System.out.println("sessionIdle");
}
public void sessionOpened(IoSession ioSession) throws Exception {
System.out.println("sessionOpened");
}
}
实体消息体Message
package net.heartma.pojo;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
/**
* 定义消息体属性
* @author heartma
*
*/
public class Message {
private String header; //头
private int length; //卡号长度
private String card; //卡号
private double weight; //重量
public String getHeader() {
return header;
}
public void setHeader(String header) {
this.header = header;
}
public int getLength() {
return length;
}
public void setLength(int length) {
this.length = length;
}
public String getCard() {
return card;
}
public void setCard(String card) {
this.card = card;
}
public double getWeight() {
return weight;
}
public void setWeight(double weight) {
this.weight = weight;
}
/**
* 解析字节数组
* @param messageBytes
*/
public final boolean ReadFromBytes(byte[] messageBytes) {
//获取头部
byte[] head = new byte[2];
System.arraycopy(messageBytes, 0, head, 0,2);
setHeader(new String(head));
//获取长度
byte[] len = new byte[1];
System.arraycopy(messageBytes, 2, len, 0,1);
setLength(len[0]);
//判断卡号长度是否为0，为0则说明为心跳信息，否则为真实数据
if((int)len[0]>0){
//卡号解析
byte[] cardDest = new byte[len[0]];
System.arraycopy(messageBytes, 3, cardDest, 0,len[0]);
setCard(new String(cardDest));
byte[] weightDest = new byte[7];
System.arraycopy(messageBytes, 3+length, weightDest, 0,7);
setWeight(Double.parseDouble(new String(weightDest)));
return true;
}
return false;
}
@Override
public String toString() {
return "Message [header=" + header + ", length=" + length + ", card="
+ card + ", weight=" + weight + "]";
}
}
自定义协议工厂类
package net.heartma.protocol;
import java.nio.charset.Charset;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolEncoder;
/**
* 自定义协议工厂类
* @author heartma
*/
public class CustomProtocolCodecFactory implements ProtocolCodecFactory{
private final CustomProtocolDecoder decoder;
private final CustomProtocolEncoder encoder;
public CustomProtocolCodecFactory(){
this.decoder = new CustomProtocolDecoder(Charset.forName("utf-8"));
this.encoder = new CustomProtocolEncoder(Charset.forName("utf-8"));
}
@Override
public ProtocolDecoder getDecoder(IoSession ioSession) throws Exception {
return decoder;
}
@Override
public ProtocolEncoder getEncoder(IoSession ioSession) throws Exception {
return encoder;
}
}
自定义协议编码
package net.heartma.protocol;
import java.nio.charset.Charset;
import net.heartma.tools.ByteUtil;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoderAdapter;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
/**
* 自定义协议编码
* @author heartma
*/
public class CustomProtocolEncoder extends ProtocolEncoderAdapter {
private Charset charset;
public CustomProtocolEncoder(Charset charset) {
this.charset = charset;
}
@Override
public void encode(IoSession session, Object message,
ProtocolEncoderOutput out) throws Exception {
byte[] hexStrToByte = ByteUtil.hexStr2ByteArray(message.toString());
IoBuffer buf = IoBuffer.allocate(hexStrToByte.length).setAutoExpand(false);
//byte[] content = Tools.HexString2Bytes(message.toString());
buf.put(hexStrToByte);
buf.flip();
out.write(buf);
out.flush();
buf.free();
}
}
自定义协议解码
package net.heartma.protocol;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import net.heartma.pojo.Message;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoderAdapter;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
/**
* 自定义协议解码
* @author heartma
*/
public class CustomProtocolDecoder extends ProtocolDecoderAdapter {
private CharsetDecoder decoder;
public CustomProtocolDecoder(Charset charset){
this.decoder = charset.newDecoder();
}
@Override
public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out)
throws Exception {
int limit = in.limit();
byte[] bytes = new byte[limit];
in.get(bytes);
Message message = new Message();
message.ReadFromBytes(bytes);
out.write(message);
}
}
代码下载:
版权声明：本文为costoalex原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。