`
Donald_Draper
  • 浏览: 949546 次
社区版块
存档分类
最新评论

netty 网络通信示例二

netty 
阅读更多
netty 网络通信示例一 ：http://donald-draper.iteye.com/blog/2383326
上篇文章我们看了一个简单的网络通信实例，在通信的过程成由于网络等原因，可能存在粘包的问题，对于粘包问题，处理呢。我们来看一个获取服务器时间的实例，这个实例也许不够恰当，我们只是示范处理粘包问题：
服务端： 
package netty.main.time;

import java.net.InetSocketAddress;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import netty.handler.time.TimeServerHandler;

/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:48:17
 */
public class TimeServer {
	private static final Logger log = LoggerFactory.getLogger(TimeServer.class);
	 static final boolean SSL = System.getProperty("ssl") != null;
	private static final  String ip = "192.168.31.153";
	private static final  int port = 10010;
    public static void main(String[] args) throws Exception {
      run();
    }
    public static void run() throws Exception {
    	 // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

    	/*
    	 * EventLoopGroup（多线程事件loop），处理IO操作，这里我们用了两个事件loop
    	 * 第一个boss用于处理器监听连接请求，第二个worker用于数据的传输；
    	 * 具体线程是多少依赖于事件loop的具体实现
    	 * */
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
        	//ServerBootstrap，用于配置服务端，一般为ServerSocket通道
            ServerBootstrap serverBoot = new ServerBootstrap(); 
            serverBoot.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) 
             .childHandler(new ChannelInitializer<SocketChannel>() { 
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                	//添加通道处理器到通道关联的管道，准确的中文翻译为管道线， 此管道线与Mina中过滤链十分相似,
                	//ChannelInitializer用于配置通道的管道线，ChannelPipeline
                	 ChannelPipeline pipeline = ch.pipeline();
                     if (sslCtx != null) {
                    	 pipeline.addLast(sslCtx.newHandler(ch.alloc()));
                     }
//                     pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                     pipeline.addLast(new TimeServerHandler());
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)//socket监听器连接队列大小、
             .childOption(ChannelOption.SO_KEEPALIVE, true); //保活，此配置针对ServerSocket通道接收连接产生的Socket通道
            InetSocketAddress inetSocketAddress = new InetSocketAddress(ip,port);
            // 绑定地址，开始监听
            ChannelFuture f = serverBoot.bind(inetSocketAddress).sync();
            log.info("=========Server is start=========");
            //等待，直到ServerSocket关闭
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

服务端处理器： 
package netty.handler.time;

import java.nio.charset.Charset;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:48:01
 */
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
   private static final Logger log = LoggerFactory.getLogger(TimeServerHandler.class);
   private static final String TIME_PROTOCL = "?time";
   private static final Charset charsetDecoder= Charset.forName("UTF-8");
   /**
    * 读client通道数据，通道处理器上下文ChannelHandlerContext与Mina的会话很像
    */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
    	ByteBuf in = (ByteBuf)msg;
    	String message = (String) in.readCharSequence(in.writerIndex(), charsetDecoder);
        log.info("===Server reciever message:" +message);
        if(message.equals(TIME_PROTOCL)){
        	//通过通道处理器上下文的ByteBufAllocator创建容量至少为8个字节的ByteBuf
        	ByteBuf time = ctx.alloc().buffer(8);
        	time.writeLong(System.currentTimeMillis());
        	/*
        	在发送数据时，我们并没有调用nio的ByteBuffer#flip类似的方法，这是由于
        	为了避免nio忘记flip操作的问题，Netty通过readIndex和writeIndex两个index
        	表示ByteBuf的相对开始和结束位置；当向ByteBuffer中写数据时，writeIndex将会增长，
        	而readIndex不变。
        	*/
        	/*
        	ctx#write,writeAndFlush方法返回一个写结果ChannelFuture,
        	ChannelFuture表示一个IO事件操作，如果想要在ctx写操作后，关闭连接，不可以用如下方式：
        	Channel ch = ...;
        	ch.writeAndFlush(message);
        	ch.close();
        	因为Netty的写操作时异步的，上面这种关闭连接方式，有可能在消息没发送完前，连接已经关闭，为了
        	能在消息发送完毕后再关闭会话，可以通过添加通道结果监听器，在消息发送完时，触发监听器operationComplete
        	事件。*/
        	
        	final ChannelFuture cfuture = ctx.writeAndFlush(time);
        	final ChannelHandlerContext ctx_refer = ctx;
        	cfuture.addListener(new ChannelFutureListener() {
        	        @Override
        	        public void operationComplete(ChannelFuture future) {
        	            assert cfuture == future;
        	            ctx_refer.close();
        	        }
        	 }); 
        	//上面添加监听器，可以直接使用通道结果监听器内部的CLOSE监听器
        	//cfuture.addListener(ChannelFutureListener.CLOSE);
        }   
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    	//异常发生时，关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}


客户端： 
package netty.main.time;

import java.net.InetSocketAddress;

import javax.net.ssl.SSLException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import netty.handler.time.TimeClientHandler;
/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:48:10
 */
public final class TimeClient {
	private static final Logger log = LoggerFactory.getLogger(TimeClient.class);
	private static final boolean SSL = System.getProperty("ssl") != null;
	private static final String ip = System.getProperty("host", "192.168.31.153");
	private static final int port = Integer.parseInt(System.getProperty("port", "10010"));
    public static void main(String[] args) throws Exception {
       run();
    }
    private static void run() throws SSLException, InterruptedException{
    	 //配置安全套接字上下文
        final SslContext sslCtx;
        if (SSL) {
            sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
        } else {
            sslCtx = null;
        }
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
        	//Bootstrap与 ServerBootstrap相似，不同的是Bootstrap用于配置客户端，
        	//一般为Socket通道，或无连接通道
            Bootstrap bootstrap = new Bootstrap();
            //EventLoopGroup有 boss和worker两组,对于客户端只需要用worker
            bootstrap.group(workerGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                	 //添加安全套接字处理器和通道处理器到
                     ChannelPipeline pipeline = ch.pipeline();
                     if (sslCtx != null) {
                    	 pipeline.addLast(sslCtx.newHandler(ch.alloc(), ip, port));
                     }
//                     pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                     pipeline.addLast(new TimeClientHandler());
                 }
             });
            InetSocketAddress inetSocketAddress = new InetSocketAddress(ip,port);
            //连接socket地址
            ChannelFuture f = bootstrap.connect(inetSocketAddress).sync();
            log.info("=========Client is start=========");
            //等待，直到连接关闭
            f.channel().closeFuture().sync();
        } finally {
        	workerGroup.shutdownGracefully();
        }
    }
}

客户端处理器： 
package netty.handler.time;


import java.nio.charset.Charset;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:47:53
 */
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
	private static final Logger log = LoggerFactory.getLogger(TimeClientHandler.class);
	private static final String TIME_PROTOCL = "?time";
	private static final Charset charsetEncoder= Charset.forName("UTF-8");
	/**
	 * 在通道连接建立时（准备传输数据）触发
	 */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
    	ByteBuf timeReq = ctx.alloc().buffer(5);
    	timeReq.writeCharSequence(TIME_PROTOCL, charsetEncoder);
    	ctx.writeAndFlush(timeReq);
    }  
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    	ByteBuf in = (ByteBuf)msg;
    	long nowTime = in.readLong();
    	Date nowDay = new Date(nowTime);
    	log.info("===Server Time:" +nowDay.toLocaleString());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

启动服务端与客户端，控制台输出：
服务端：
[INFO ] 2017-07-05 22:50:09 netty.main.time.TimeServer =========Server is start=========
[INFO ] 2017-07-05 22:50:14 netty.handler.time.TimeServerHandler ===Server reciever message:?time
客户端：
[INFO ] 2017-07-05 22:50:14 netty.main.time.TimeClient =========Client is start=========
[INFO ] 2017-07-05 22:50:14 netty.handler.time.TimeClientHandler ===Server Time:2017-7-5 22:50:14

针对粘包问题，我们对上面的实例进行改造：
服务端： 
package netty.main.time;

import java.net.InetSocketAddress;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import netty.handler.time.TimeServerHandler2;

/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:48:17
 */
public class TimeServerForDecoder {
	private static final Logger log = LoggerFactory.getLogger(TimeServerForDecoder.class);
	 static final boolean SSL = System.getProperty("ssl") != null;
	private static final  String ip = "192.168.31.153";
	private static final  int port = 10010;
    public static void main(String[] args) throws Exception {
      run();
    }
    public static void run() throws Exception {
    	 // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

    	/*
    	 * EventLoopGroup（多线程事件loop），处理IO操作，这里我们用了两个事件loop
    	 * 第一个boss用于处理器监听连接请求，第二个worker用于数据的传输；
    	 * 具体线程是多少依赖于事件loop的具体实现
    	 * */
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
        	//ServerBootstrap，用于配置服务端，一般为ServerSocket通道
            ServerBootstrap serverBoot = new ServerBootstrap(); 
            serverBoot.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) 
             .childHandler(new ChannelInitializer<SocketChannel>() { 
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                	//添加通道处理器到通道关联的管道，准确的中文翻译为管道线， 此管道线与Mina中过滤链十分相似,
                	//ChannelInitializer用于配置通道的管道线，ChannelPipeline
                	 ChannelPipeline pipeline = ch.pipeline();
                     if (sslCtx != null) {
                    	 pipeline.addLast(sslCtx.newHandler(ch.alloc()));
                     }
//                     pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                     pipeline.addLast(new TimeServerHandler2());
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)//socket监听器连接队列大小、
             .childOption(ChannelOption.SO_KEEPALIVE, true); //保活，此配置针对ServerSocket通道接收连接产生的Socket通道
            InetSocketAddress inetSocketAddress = new InetSocketAddress(ip,port);
            // 绑定地址，开始监听
            ChannelFuture f = serverBoot.bind(inetSocketAddress).sync();
            log.info("=========Server is start=========");
            //等待，直到ServerSocket关闭
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
  
}


服务端处理器： 
package netty.handler.time;

import java.nio.charset.Charset;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 
 * @author donald
 * 2017年6月21日
 * 下午12:48:01
 */
public class TimeServerHandler2 extends ChannelInboundHandlerAdapter {
   private static final Logger log = LoggerFactory.getLogger(TimeServerHandler2.class);
   private static final String TIME_PROTOCL = "?time";
   private static final Charset charsetDecoder= Charset.forName("UTF-8");
   /**
    * 读client通道数据，通道处理器上下文ChannelHandlerContext与Mina的会话很像
    */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
    	ByteBuf in = (ByteBuf)msg;
    	String message = (String) in.readCharSequence(in.writerIndex(), charsetDecoder);
        log.info("===Server reciever message:" +message);
        if(message.equals(TIME_PROTOCL)){
        	//通过通道处理器上下文的ByteBufAllocator创建容量至少为8个字节的ByteBuf
        	ByteBuf time = ctx.alloc().buffer(8);
        	time.writeLong(System.currentTimeMillis());
        	/*
        	在发送数据时，我们并没有调用nio的ByteBuffer#flip类似的方法，这是由于
        	为了避免nio忘记flip操作的问题，Netty通过readIndex和writeIndex两个index
        	表示ByteBuf的相对开始和结束位置；当向ByteBuffer中写数据时，writeIndex将会增长，
        	而readIndex不变。
        	*/
        	/*
        	ctx#write,writeAndFlush方法返回一个写结果ChannelFuture,
        	ChannelFuture表示一个IO事件操作，如果想要在ctx写操作后，关闭连接，不可以用如下方式：
        	Channel ch = ...;
        	ch.writeAndFlush(message);
        	ch.close();
        	因为Netty的写操作时异步的，上面这种关闭连接方式，有可能在消息没发送完前，连接已经关闭，为了
        	能在消息发送完毕后再关闭会话，可以通过添加通道结果监听器，在消息发送完时，触发监听器operationComplete
        	事件。*/
        	ctx.writeAndFlush(time);
        }   
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    	//异常发生时，关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

客户端： 
package netty.main.time;

import java.net.InetSocketAddress;

import javax.net.ssl.SSLException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import netty.codec.time.TimeDecoder;
import netty.handler.time.TimeClientHandler;
/**
 *  客户端要与TimeServerForDecoder服务端配合使用
 * @author donald
 * 2017年6月21日
 * 下午12:48:10
 */
public final class TimeClientWithDecoder {
	private static final Logger log = LoggerFactory.getLogger(TimeClientWithDecoder.class);
	private static final boolean SSL = System.getProperty("ssl") != null;
	private static final String ip = System.getProperty("host", "192.168.31.153");
	private static final int port = Integer.parseInt(System.getProperty("port", "10010"));
    public static void main(String[] args) throws Exception {
       run();
    }
    private static void run() throws SSLException, InterruptedException{
    	 //配置安全套接字上下文
        final SslContext sslCtx;
        if (SSL) {
            sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
        } else {
            sslCtx = null;
        }
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
        	//Bootstrap与 ServerBootstrap相似，不同的是Bootstrap用于配置客户端，
        	//一般为Socket通道，或无连接通道
            Bootstrap bootstrap = new Bootstrap();
            //EventLoopGroup有 boss和worker两组,对于客户端只需要用worker
            bootstrap.group(workerGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                	 //添加安全套接字处理器和通道处理器到
                     ChannelPipeline pipeline = ch.pipeline();
                     if (sslCtx != null) {
                    	 pipeline.addLast(sslCtx.newHandler(ch.alloc(), ip, port));
                     }
//                     pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                     pipeline.addLast(new TimeDecoder(),new TimeClientHandler());
                 }
             });
            InetSocketAddress inetSocketAddress = new InetSocketAddress(ip,port);
            //连接socket地址
            ChannelFuture f = bootstrap.connect(inetSocketAddress).sync();
            log.info("=========Client is start=========");
            //等待，直到连接关闭
            f.channel().closeFuture().sync();
        } finally {
        	workerGroup.shutdownGracefully();
        }
    }
}

客户端解码器： 
package netty.codec.time;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
/**
 * 字节流消息解码器ByteToMessageDecoder，是#ChannelInboundHandler的实现，可以解决粘包问题；
 * 字节消息解码的内部有一个可累计buffer，当有数据到达时，将会调用#decode方法，解码消息，如果累计buffer中
 * 没有足够的数据，则不会添加对象到out，如果有对象添加到out，表示解码器成功解码了一个消息；我们不需要一次解码多个消息，
 * 解码器将会不断地调用#decode方法，直到没有对象可以添加到out。
 * @author donald
 * 2017年6月22日
 * 上午8:55:20
 */
public class TimeDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < 8) {
            return;
        }
        out.add(in.readBytes(8));
    }
}

启动服务端与客户端：
服务端：
[INFO ] 2017-07-05 22:57:58 netty.main.time.TimeServerForDecoder =========Server is start=========
[INFO ] 2017-07-05 22:58:08 netty.handler.time.TimeServerHandler2 ===Server reciever message:?time
客户端：
[INFO ] 2017-07-05 22:58:08 netty.main.time.TimeClientWithDecoder =========Client is start=========
[INFO ] 2017-07-05 22:58:08 netty.handler.time.TimeClientHandler ===Server Time:2017-7-5 22:58:08

解码器还有另外一种形式： 
package netty.codec.time;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;
/**
 * 回复解码器ReplayingDecoder为字节流消息解码器ByteToMessageDecoder的实现
 * @author donald
 * 2017年6月22日
 * 上午8:55:20
 */
public class TimeDecoder2 extends ReplayingDecoder<Void> {
    @Override
    protected void decode(
            ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        out.add(in.readBytes(8));
    }
}

还有另外一种形式，在客户端处理器中处理，当然这种方式我们不建议使用： 
package netty.handler.time;


import java.nio.charset.Charset;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
 *TimeClientHandlerX处理粘包情况，
 *由于网络等原因，有时候我们不能够一次接收一个完成的数据包，我们必须等待完成的数据包，
 *我们可以等待数据包数据完成时，才解析数据。
 *在此示例中，我们接收的是一个8字节的long数据，在网络不佳的情况下，也有可能出现不能一次
 *接收的情况。
 * @author donald
 * 2017年6月21日
 * 下午12:47:53
 */
public class TimeClientHandler2 extends ChannelInboundHandlerAdapter {
	private static final Logger log = LoggerFactory.getLogger(TimeClientHandler2.class);
	private static final String TIME_PROTOCL = "?time";
	private static final Charset charsetEncoder= Charset.forName("UTF-8").newEncoder().charset();
	private ByteBuf buf;
	/**
	 * 在通道连接建立时（准备传输数据）触发
	 */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
    	ByteBuf timeReq = ctx.alloc().buffer(5);
    	timeReq.writeCharSequence(TIME_PROTOCL, charsetEncoder);
    	ctx.writeAndFlush(timeReq);
    }  
    /**
     * Gets called after the {@link ChannelHandler} was added to the 
     * actual context and it's ready to handle events.
     * 在通道处理器添加到实际上下文，准备处理事件时触发，可以用于初始化阻塞时间较短的任务
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        buf = ctx.alloc().buffer(8);
    }
    /**
     * Gets called after the {@link ChannelHandler} was removed from the actual
     *  context and it doesn't handle events anymore.
     * 在通道处理器从实际上下文移除，不再处理事件时触发，可以用于释放初始化任务申请的资源
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        buf.release();
        buf = null;
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
    	ByteBuf in = (ByteBuf)msg;
    	//将所有的字节序列数据累积到buf中
    	buf.writeBytes(in);
    	in.release();
    	/*
    	 待buffer中有足够的数据时，解析数据，否则当更多的数据到达时，netty将会再次调用#channelRead方法
    	 */
    	if(buf.readableBytes()>=8){
    		long nowTime = buf.readLong();
        	Date nowDay = new Date(nowTime);
        	log.info("===Server Time:" +nowDay.toLocaleString());
    	}
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
分享到：
| netty 网络通信示例一
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    Spring Boot 整合 Netty + WebSocket 实时消息推送

    1、基于Spring Boot 实现的WebSocket实时数据通信Demo 2、结合Netty实现多客户端之间进行网络通信 3、在Web端建立多客户端之间的通信机制

    基于Netty框架的demo项目

    项目中，你可以清晰地看到BIO与NIO模型在Netty中的灵活运用，如何通过Netty的高性能特性来优化网络通信。此外，demo还详细展示了如何通过自定义编解码器处理网络通信中的数据编解码问题，以及如何利用心跳机制确保...

    利用Java搭建个简单的Netty通信实例教程

    主要给大家介绍了关于如何利用Java搭建个简单的Netty通信，文中通过示例代码介绍的非常详细，对大家学习或者使用Java具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧

    Spring Boot与Netty：构建高性能的网络应用

    1. 异步和事件驱动：Netty 使用异步和事件驱动的方式来处理网络通信，这意味着它不会阻塞调用线程，从而提高了应用的响应性和吞吐量。 2. 高性能：Netty 的设计使得它成为了目前所有 NIO 框架中性能最好的框架之一。...

    精通并发与netty视频教程(2018)视频教程

    25_gRPC通信示例与JVM回调钩子 26_gRPC服务器流式调用实现 27_gRPC双向流式数据通信详解 28_gRPC与Gradle流畅整合及问题解决的完整过程与思考 29_Gradle插件问题解决方案与Nodejs环境搭建 30_通过gRPC实现Java与...

    精通并发与netty 无加密视频

    第25讲：gRPC通信示例与JVM回调钩子 第26讲：gRPC服务器流式调用实现 第27讲：gRPC双向流式数据通信详解 第28讲：gRPC与Gradle流畅整合及问题解决的完整过程与思考 第29讲：Gradle插件问题解决方案与Nodejs环境...

    精通并发与 netty 视频教程(2018)视频教程

    20_通过Apache Thrift实现Java与Python的RPC调用 21_gRPC深入详解 22_gRPC实践 23_Gradle Wrapper在Gradle项目构建中的最佳实践 24_gRPC整合Gradle与代码生成 25_gRPC通信示例与JVM回调钩子 26_gRPC服务器流式调用...

    java开源包3

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    DotNettySample.zip

    DotNetty是微软团队参考Java上的Netty开发的网络通信框架，并且保留了Netty原来绝大部分的编程接口。但目前最大的问题是没有官方说明文档，官方示例也仅仅是控制台应用程序，参考价值较低。本项目展示了在WPF中...

    基于 Java Web 项目的 SpringBoot 框架初始化模板，可基于此快速开发毕设等中小型项目

    基于 Java Web 项目的 SpringBoot 框架初始化模板，该模板整合了常用的框架，保证大家在此...19、基于 Netty 的 WebSocket 全双工通信设计示例 20、对象存储、消息队列、缓存、分布式锁、限流、国际化、网络等工具类

    Accordion:点对点我的世界服务器通信

    核心模块包含基于 netty 的网络代码，这些代码与 minecraft 没有特别关联。 我的世界 minecraft 模块包含一个特定于 minecraft 的网络设置，并且可以从 bukkit 和 bungee 插件轻松访问它们。 我的世界示例 该模块...

    JAVA上百实例源码以及开源项目

    　Java局域网通信——飞鸽传书源代码，大家都知道VB版、VC版还有Delphi版的飞鸽传书软件，但是Java版的确实不多，因此这个Java文件传输实例不可错过，Java网络编程技能的提升很有帮助。 Java聊天程序，包括服务端和...

    JAVA上百实例源码以及开源项目源代码

    Java局域网通信——飞鸽传书源代码 28个目标文件 内容索引:JAVA源码,媒体网络,飞鸽传书 Java局域网通信——飞鸽传书源代码，大家都知道VB版、VC版还有Delphi版的飞鸽传书软件，但是Java版的确实不多，因此这个Java...

    java开源包2

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    Linux多线程服务端编程：使用muduo C++网络库

    《Linux多线程服务端编程：使用muduo C++网络库》主要讲述采用现代C++在x86-64 Linux上编写多线程TCP网络服务程序的主流常规技术，重点讲解一种适应性较强...第2部分muduo 网络库 第6章muduo 网络库简介125 6.1由来. ....

    Java资源包01

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    java开源包1

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    java开源包11

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    java开源包6

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

    java开源包5

    6、支持多种通信框架（Mina/Netty/Grizzly），支持多种序列化/反序列化（Java/Hessian/PB）； 7、支持自定义通信协议，可完全替换NFS-RPC自带的协议。 淘宝开放平台JAVA版SDK top4java 设计原则 容易维护扩展（不...

Magicbox
Global site tag (gtag.js) - Google Analytics