工程目錄結構

程式碼：NioServer main類

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;


import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
/** 
 * Netty 服務端程式碼 
 *  
 */  
public class NioServer {  
  
/**
* 日誌物件
*/
//protected static Logger logger = LoggerFactory.getLogger(NioServer.class);
private static int port = 9999;
    public static void main(String args[])  {  
   
        // Server服務啟動器  
        ServerBootstrap bootstrap = new ServerBootstrap();  
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
        bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel  ch) throws Exception {
//可以在socket接上來的時候新增很多指定義邏輯
ch.pipeline().addLast("encode",new StringEncoder());      
ch.pipeline().addLast("decode",new StringDecoder());  
ch.pipeline().addLast("ping", new IdleStateHandler(25, 15, 10,TimeUnit.SECONDS));
ch.pipeline().addLast(new  NioServerHandler());
}
}).option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
        // Bind and start to accept incoming connections. 
        ChannelFuture future = bootstrap.bind(port).sync();
//        logger.info("server started ,listen {}" ,port);
        System.out.println("開始監聽 "+ port);
        //啟動一個執行緒 來給客戶端發訊息
        //new Thread(new ServerTask()).run();
           // Wait until the server socket is closed.
           // In this example, this does not happen, but you can do that to gracefully
           // shut down your server. 呼叫實現優雅關機
           future.channel().closeFuture().sync();
    } catch (InterruptedException e) {
e.printStackTrace();
} finally {
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
    }
    }  
}  
class ServerTask implements Runnable{
/**
* 日誌物件
*/
//protected Logger logger = LoggerFactory.getLogger(ServerTask.class);
public void run() {
Channel channel = null;
Map.Entry<String, Channel>  entry = null;
while (true) {
Iterator<Entry<String, Channel>> it = GatewayService.map.entrySet().iterator();
while (it.hasNext()) {
entry = it.next();
channel = entry.getValue();
if (channel.isActive() && channel.isWritable()) {
entry.getValue().writeAndFlush(new Date() + "我是測試的伺服器端向客戶端發資料");
} else {
channel.close();
it.remove();
System.out.println("通道不能連線uid:{ },伺服器關閉和刪除它 "+ entry.getKey());
//logger.info("channel cannot connect uid : {}, server close and remove it" ,entry.getKey());
}
}
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}


}

 //GatewayService.java

import io.netty.channel.Channel;


import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class GatewayService {
/**
* 日誌物件
*/
//protected static Logger logger = LoggerFactory.getLogger(GatewayService.class);
/**
* 存放客戶端連線
*/
public static Map<String,Channel> map = new ConcurrentHashMap<String, Channel>();
/**
* 新增一個客戶端連線
* @param uid
* @param socketChannel
*/
public static void add(String uid,Channel channel){
if (map.containsKey(uid)) {
Channel temp =map.get(uid);
if (!(temp.isActive() && temp.isWritable())){
map.remove(uid);
map.put(uid, channel);
}
} else {
map.put(uid, channel);
}
  System.out.println("新增一個客戶端uid : " +uid + "當前客戶端數量  :"+getClientSize());
 
}
/**
* 移除一個客戶端連線
* @param uid
*/
public static void remove(String uid){
Channel channel = map.remove(uid);
channel.close();
 
 System.out.println("刪除一個客戶端uid : " +uid + "  當前客戶端數量  :"+ getClientSize());
}
/**
* 當前客戶端數量
* @return
*/
public static int getClientSize(){
return map.size();
}
}

NioServerHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;


/**
 * 實際的業務處理類
 * 
 * @liuguodong
 *
 */
public class NioServerHandler extends SimpleChannelInboundHandler<String> {


/**
* 日誌物件
*/
//protected Logger logger = LoggerFactory.getLogger(NioServerHandler.class);


@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {


super.channelRegistered(ctx);


System.out.println("--------服務註冊 通道 -------");
}


@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg)
throws Exception {

System.out.println("server--->收到客戶端發的訊息:" + msg);
ctx.writeAndFlush("Received your message !\n"+msg+'\n');
System.out.println("server--->服務端返回的資訊:" + msg+" # ");
GatewayService.add("11", ctx.channel());
}


@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//  ctx.writeAndFlush("Received your message !\n");
//  ctx.writeAndFlush("你好 !\n");
super.channelActive(ctx);



 System.out.println("--------通道啟用-------");
}


/**
* 響應netty 心跳  這個可以給客戶端傳送一些特定包 來標識
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
throws Exception {
super.userEventTriggered(ctx, evt);
if (evt instanceof IdleStateEvent) {
IdleStateEvent e = (IdleStateEvent) evt;
//多少秒沒有讀
if (e.state() == IdleState.READER_IDLE) {


//System.out.println("讀超時.....");
} else if (e.state() == IdleState.WRITER_IDLE) {//多少秒沒有寫
//ctx.close();

//System.out.println("寫超時.....");
}else if (e.state() == IdleState.ALL_IDLE) { //總時間
          
//                System.out.println("總超時.....");
                //ctx.close();
                //可以發包了， 然後發出去後  客戶單迴應一個 這時就不會進入read了 來判斷心跳
            }
}
}


@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
try {
super.exceptionCaught(ctx, cause);
} catch (Exception e) {
  System.out.println("服務異常 ....關閉連線");
 
ctx.close();
}
}


}

原始碼地址：http://download.csdn.net/download/liu3364575/9966543