基于netty的时间服务器demo

ゝ一世哀愁。 2022-07-11 13:40 226阅读 0赞

最近在在《netty权威指南》，这本书里面的netty的入门程序就是一个基于netty的时间服务器，下面我先给出这个时间服务器的实例，然后对逻辑做一定的说明，这个例子比较简单，只是后面的很多内容是基于这个案例来说明和加深的。

# 服务端 #

TimeServer 启动服务器类

package netty.quanwei.p3;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /** * Created by louyuting on 17/1/31. * netty 时间服务器 服务端 */
    public class TimeServer { 
    
        public void bind(int port) throws Exception{
            //配置服务端NIO 线程组
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup worker = new NioEventLoopGroup();
    
            ServerBootstrap server = new ServerBootstrap();
    
            try {
                server.group(boss, worker)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChildChannelInitializer());
    
                //绑定端口, 同步等待成功
                ChannelFuture future = server.bind(port).sync();
    
                //等待服务端监听端口关闭
                future.channel().closeFuture().sync();
            } finally {
                //优雅关闭 线程组
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }
    
        /** * main 函数 * @param args */
        public static void main(String[] args) {
            TimeServer server = new TimeServer();
            try {
                server.bind(18888);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

handler容器初始化

package netty.quanwei.p3;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.SocketChannel;
    
    /** * Created by louyuting on 17/1/31. */
    public class ChildChannelInitializer extends ChannelInitializer<SocketChannel> { 
    
    
        @Override
        protected void initChannel(SocketChannel channel) throws Exception {
            channel.pipeline().addLast("timeServerHandler", new TimeServerHandler());
        }
    }

timeServerHandler 核心handler

package netty.quanwei.p3;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import utils.DateUtil;
    import utils.LogUtil;
    
    import java.util.Date;
    
    /** * Created by louyuting on 17/1/31. */
    public class TimeServerHandler extends ChannelInboundHandlerAdapter{ 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    
            LogUtil.log_debug("Server -> read");
    
            ByteBuf buf = (ByteBuf)msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
    
            String body = new String(req, "UTF-8");
    
            LogUtil.log_debug("timeServer received order: " + body);
    
            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)? DateUtil.fmtDateToMillisecond(new Date()) : "BAD ORDER";
    
            //response
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            //异步发送应答消息给客户端: 这里并没有把消息直接写入SocketChannel,而是放入发送缓冲数组中
            ctx.writeAndFlush(resp);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            LogUtil.log_debug("Server -> read complete");
    
            //将发送缓冲区中数据全部写入SocketChannel
            //ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            //释放资源
            ctx.close();
        }
    }

# 客户端 #

TimeClient类：

package netty.quanwei.p3;
    
    import io.netty.bootstrap.Bootstrap;
    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.NioSocketChannel;
    
    /** * Created by louyuting on 17/1/31. * netty 时间服务器 客户端 */
    public class TimeClient { 
    
        public void connect(int port, String host) throws Exception{
            //配置客户端NIO 线程组
            EventLoopGroup group = new NioEventLoopGroup();
    
            Bootstrap client = new Bootstrap();
    
            try {
                client.group(group)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new TimeCLientHandler());
                            }
                        });
    
                //绑定端口, 异步连接操作
                ChannelFuture future = client.connect(host, port).sync();
    
                //等待客户端连接端口关闭
                future.channel().closeFuture().sync();
            } finally {
                //优雅关闭 线程组
                group.shutdownGracefully();
            }
        }
    
        /** * main 函数 * @param args */
        public static void main(String[] args) {
            TimeClient client = new TimeClient();
            try {
                client.connect(18888, "127.0.0.1");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

timeClientHandler类

package netty.quanwei.p3;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import utils.LogUtil;
    
    /** * Created by louyuting on 17/1/31. */
    public class TimeCLientHandler extends ChannelInboundHandlerAdapter{ 
        private final ByteBuf firstMSG;
    
        public TimeCLientHandler() {
            byte[] req = "QUERY TIME ORDER".getBytes();
            firstMSG = Unpooled.buffer(req.length);
            firstMSG.writeBytes(req);
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            LogUtil.log_debug("client -> active");
    
            ctx.writeAndFlush(firstMSG);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LogUtil.log_debug("client -> read");
    
            ByteBuf buf = (ByteBuf)msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
    
            String body = new String(req, "UTF-8");
    
            LogUtil.log_debug("NOW is: " + body);
    
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

# 运行结果： #

先启动server，然后再运行client：  
Server运行结果：

2017-01-31 12:40:19：Server -> read
    2017-01-31 12:40:19：timeServer received order: QUERY TIME ORDER
    2017-01-31 12:40:19：Server -> read complete

client运行结果：

2017-01-31 12:40:19：client -> active
    2017-01-31 12:40:19：client -> read
    2017-01-31 12:40:19：NOW is: 2017-01-31 12:40:19-861

# 逻辑分析： #

先启动服务器，然后启动客户端；

客户端启动后首先与服务器建立连接，并发送`QUERY TIME ORDER`  
指令给服务器，服务器接收到指令之后，返回当前时间给客户端。

客户端首先服务器响应之后，回显服务器时间。