Netty TCP协议简单实现

Netty的强大之处在于它的高度抽象和封装，对于使用者来说不必过多关心内部实现。当需要有新的需求时，只需简单的添加或者修改相关的Handler类即可。

本章将使用Netty 实现TCP协议，以下为具体实现。

 

1、服务端实现

 

TcpServer.java

package emulator.netty5;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;


public class TcpServer {

	public void bind(int port)throws Exception{
		
		//配置服务端Nio线程组
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try{
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workerGroup)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 1024)
				.childHandler(new ChildChannelHandler());
			//绑定端口，同步等待成功
			ChannelFuture f = b.bind(port).sync();
			//等待服务端监听端口关闭
			f.channel().closeFuture().sync();
			
		}finally{
			//退出时释放资源
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}		
	}
	

	private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
		@Override
		protected void initChannel(SocketChannel channel) throws Exception {
			ChannelPipeline pipeline = channel.pipeline();
			pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
			pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
			pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
			pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
			pipeline.addLast(new TcpServerHandler());
		}		
	}
		
	public static void main(String[] args) throws Exception{
		int port = 8083;
		new TcpServer().bind(port);		
	}
}

 

TcpServerHandler.java

package emulator.netty5;


import java.io.File;
import java.io.IOException;

import org.apache.commons.io.FileUtils;

import emulator.Constants;
import emulator.util.Dom4JUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.CharsetUtil;

public class TcpServerHandler extends ChannelHandlerAdapter {

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		String body = (String)msg;
		System.out.println("request content:\n"+ body);
		//响应
		resp(ctx, body);		
	}
	
	
	/**
	 * 
	 * @param xml
	 */
	private void resp(ChannelHandlerContext ctx, String xml){
		String transCode = Dom4JUtil.header(xml, "TransCode");
		String retUrl = "D:\\workspaces\\eclipse-huifu\\emulator\\xml\\error.xml";
		String retCtt = null;
		System.out.println("交易代码："+transCode);		
		if(equal(transCode, Constants.TC_DZZH)){//电子账户			
			retUrl = "D:\\workspaces\\eclipse-huifu\\emulator\\xml\\account\\manage\\resp.xml";
		}
		try {
			retCtt = FileUtils.readFileToString(new File(retUrl));
		} catch (IOException e) {
			e.printStackTrace();
		}
		ByteBuf resp = Unpooled.copiedBuffer(retCtt.getBytes());
		ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
		
	}
	
	public static boolean equal(String var, String cons){
		return isNotEmpty(var) && cons.equals(var);		
	}
	
	private static boolean isNotEmpty(String s){
		return (null != s && !"".equals(s));
	}

	
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		super.channelReadComplete(ctx);
		//ctx.flush();
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		//super.exceptionCaught(ctx, cause);
		ctx.close();
	}
	
}

 

2、客户端实现

 

TcpClient.java

package emulator.netty5;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class TcpClient {
	
	public void connect(int port,String host)throws Exception{
		
		//配置客户端NIO线程组
		EventLoopGroup group = new NioEventLoopGroup();
		
		try{
			Bootstrap b = new Bootstrap();
			b.group(group).channel(NioSocketChannel.class)
				.option(ChannelOption.TCP_NODELAY, true)
				.handler(new ChannelInitializer<SocketChannel>() {
					protected void initChannel(SocketChannel ch) throws Exception {
						ChannelPipeline pipeline = ch.pipeline();
						pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
						pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
						pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
						pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
						pipeline.addLast("handler", new TcpClientHandler());				
					};
				});
			
			//发起异步连接操作
			ChannelFuture f = b.connect(host,port).sync();
			//等待客户端链路关闭
			f.channel().closeFuture().sync();
		}finally{
			//退出，释放资源
			group.shutdownGracefully();
		}
		
	}
	
	public static void main(String[] args)throws Exception {
		int port = 8083;
		new TcpClient().connect(port, "127.0.0.1");		
	}
}

 

TcpClientHandler.java

package emulator.netty5;

import java.io.File;
import java.io.IOException;
import java.util.logging.Logger;

import org.apache.commons.io.FileUtils;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.CharsetUtil;

/**
 * 
 * @author lh
 *
 */
public class TcpClientHandler extends ChannelHandlerAdapter {

	private static final Logger logger = Logger.getLogger(TcpClientHandler.class.getName());
	
	
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		String retUrl = "D:\\workspaces\\eclipse-huifu\\emulator\\xml\\account\\manage\\req.xml";
		String ret = null;
		try {
			ret = FileUtils.readFileToString(new File(retUrl));
		} catch (IOException e) {
			e.printStackTrace();
		}
		ctx.writeAndFlush(ret);
	}
	
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		String body = (String)msg;
		System.out.println("server response :\n"+body);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		logger.warning("unexpected exception from downstream:"+ cause.getMessage());
		ctx.close();
	}
	
}

 

OVER!