netty4.0学习笔记

1.初步使用netty框架创建服务端


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.Map;
import java.util.concurrent.TimeUnit;
 
public class NettyServer { 
private int port;
private String ipAddress;
private EventLoopGroup bossGroup = null;
private EventLoopGroup workerGroup = null;
private ServerDecoder serverDecoder = null;
private ServerEncoder serverEncoder = null;
private BusinessHandler businessHandler = null;
public NettyServer(int port,String ipAddress){
this.port = port;
this.ipAddress = ipAddress;
}

public void start() throws Exception {
ServerBootstrap b = new ServerBootstrap();
bossGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors()*2);
workerGroup = new NioEventLoopGroup();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("idleStateHandler", new IdleStateHandler(60, 60, 90,
                TimeUnit.SECONDS));
serverDecoder = new ServerDecoder();//当服务端接收数据时自己的业务处理类
serverEncoder = new ServerEncoder();//当服务端发送数据时自己的业务处理类
businessHandler = new BusinessHandler();//发送和接收handler

pipeline.addLast("encoder",serverEncoder);
pipeline.addLast("decoder",serverDecoder);
pipeline.addLast("handler",businessHandler);
}
});
b.bind(this.port).sync();
System.out.println("TCP服务器已启动。。。");
}

public void shutdown() {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
         //下面两个方法是我需要下发的两条命令提供的方法
//提供外部调用的方法根据实际情况传参数
public boolean sendSetParameter(Map<String,Object> obj,String ipAddress) {
return this.businessHandler.sendSetParameter(obj,ipAddress);
}
public Object sendCopyData(Map<String,Object> obj,String ipAddress) {
return this.businessHandler.sendcopyData(obj, ipAddress);
}
} 

//-----------------------------

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;


/**
* 读取数据是处理业务类
* @Description 模块描述
* @version 1.00
* @see 参考类1
* @Date 2014-10-27 下午3:09:14
*/
public class ServerDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in,
List<Object> out) throws Exception {
in.resetReaderIndex();
ByteBufToBytes reader = new ByteBufToBytes();
byte[] b = reader.read(in);
         //根据自己的业务逻辑需要处理接收到的byte数组
                      //.....
out.add("处理后的数据");
}
}

//-------------------------

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

import java.util.Map;


/**
* 下发命令时处理业务
* @Description 模块描述
* @version 1.00
* @see 参考类1
* @Date 2014-10-8 上午11:25:25
*/
//注意根据你通过handler类下发的数据类型需要和 Encoder类数据类型保持一致 我这里使用的 Map<String, Object>
public class ServerEncoder extends MessageToByteEncoder<Map<String, Object>> {
@Override
protected void encode(ChannelHandlerContext ctx, Map<String, Object> map,ByteBuf out) throws Exception {
        //需要处理自己的业务逻辑然后下发命令
          // 比如由此需求 取出MAP对象里面的数据然后按16进制发送
out.writeBytes("处理后的数据");
}
}

//------------

package com.biiway.gmrc.netty.server;

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.Map;
//我使用的是ChannelDuplexHandler 他包含了读和写的操作实际上它实现了
//ChannelHandler, ChannelInboundHandler, ChannelOutboundHandler 这三个接口 --具体详解请查阅API
public class BusinessHandler extends ChannelDuplexHandler {
         //记录每个连接的客户端使用IP地址作为KEY
public Map<String, ChannelHandlerContext> channelMap = new HashMap<String, ChannelHandlerContext>();
         //得到返回的参数存放的量
public Object dataObj = new Object();

         //读取客户端反馈的信息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
System.out.println("读取返回信息。。。。"+msg);
dataObj =  msg;
}
//缓冲机制
@Override
public void flush(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}

/**
*监听当有终端连接的时候记录连接
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
String ipOrPort = ctx.channel().remoteAddress().toString();//得到IP地址和端口因为我的使用情况是端口号随机所以就使用IP来区分
System.out.println("已连接...."+ipOrPort);
channelMap.put(ipOrPort.split(":")[0], ctx);
super.channelActive(ctx);
}

/**
*监听当有终端连接断开时删除记录连接
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// TODO Auto-generated method stub
String ipOrPort = ctx.channel().remoteAddress().toString();
                  //当客户端连接中断时清除自己存放的客户端连接
channelMap.remove(ipOrPort.split(":")[0]);
System.out.println("断开连接....");
super.channelInactive(ctx);
}

/**
* 心跳机制
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
throws Exception {
if (evt instanceof IdleStateEvent) {
            IdleStateEvent _evt = (IdleStateEvent) evt;
            //当状态正常时
            if (_evt.state() == IdleState.WRITER_IDLE) {
            ctx.writeAndFlush("根据需求定义自己的心跳协议");
            } else if (_evt.state() == IdleState.READER_IDLE) {
            //当超过自定义的响应时间，如果客户端还没响应，则服务端主动断开与此客户端的连接
                ctx.fireExceptionCaught(new SocketTimeoutException(
                        "force to close channel("
                                + ctx.channel().remoteAddress()
                                + "), reason: time out."));
                ctx.channel().close();
            }
        }
super.userEventTriggered(ctx, evt);
}

/**
* 设置参数
* @throws Exception
* @Date 2014-10-9 下午5:26:18
*/
public boolean sendSetParameter(Map<String, Object> obj,String ipOrPort){
Map<String, ChannelHandlerContext> map = this.channelMap;
//找到需要下发命令的客户端连接
ChannelHandlerContext cxt = map.get(ipOrPort);
cxt.writeAndFlush(obj);
return null;
}
/**
* 抄收数据
* @Date 2014-10-9 下午5:26:33
*/
public Object sendcopyData(Map<String, Object> obj, String ipAndPort){
Map<String, ChannelHandlerContext> map = this.channelMap;
ChannelHandlerContext cxt =  map.get(ipAndPort);
try {
cxt.writeAndFlush(obj);
if(this.dataObj != null){
return this.dataObj;
}
} catch (Exception e) {
return this.dataObj;
}
return this.dataObj;
}

}

//---------把得到的byteBuf转为BYTE数组工具类

import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
 
public class ByteBufToBytes { 
    private ByteBuf temp; 
 
    private boolean end = true; 
 
    public ByteBufToBytes() {} 
 
    public ByteBufToBytes(int length) { 
        temp = Unpooled.buffer(length); 
    } 
 
    public void reading(ByteBuf datas) { 
        datas.readBytes(temp, datas.readableBytes()); 
        if (this.temp.writableBytes() != 0) { 
            end = false; 
        } else { 
            end = true; 
        } 
    } 
 
    public boolean isEnd() { 
        return end; 
    } 
 
    public byte[] readFull() { 
        if (end) { 
            byte[] contentByte = new byte[this.temp.readableBytes()]; 
            this.temp.readBytes(contentByte); 
            this.temp.release(); 
            return contentByte; 
        } else { 
            return null; 
        } 
    } 
 
    public byte[] read(ByteBuf datas) { 
        byte[] bytes = new byte[datas.readableBytes()]; 
        datas.readBytes(bytes); 
        return bytes; 
    } 
} 

----netty框架很牛掰，学习后的一点经验，如果有用过的大牛给点意见。。