欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Netty 入门示例

程序员文章站 2022-04-23 11:57:53
...

服务端代码示例

TimeServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;


public class TimeServer {

    public void bind(int port) throws Exception {
        //配置服务端的NIO线程组
        //NioEventLoopGroup是个线程组,它包含了一组NIO线程,专门用于网络事件的处理,
        //实际上它们就是Reactor线程组。
        //这里创建两个的原因是一个用于服务端接受客户端的连接,另一个用于进行SocketChannel的网络读写。
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建ServerBootstrap对象,它是Netty用于启动NIO服务端的辅助启动类,目的是降低服务端的开发复杂度。
            ServerBootstrap b = new ServerBootstrap();
            //调用ServerBootstrap的group方法,将两个NIO线程组当作入参传递到ServerBootstrap中。
            //接着设置创建的Channel为NioServerSocketChannel,它的功能对应于JDK NIO类库中的ServerSocketChannel类。
            //然后配置NioServerSocketChannel的TCP参数,此处将它的backlog设置为1024,
            //最后绑定I/O事件的处理类ChildChannelHandler,它的作用类似于Reactor模式中的handler类,
            //主要用于处理网络I/O事件,例如记录日志、对消息进行编解码等。
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChildChannelHandler());
            //绑定端口,同步等待成功
            //服务端启动辅助类配置完成之后,调用它的bind方法绑定监听端口
            //随后,调用它的同步阻塞方法sync等待绑定操作完成。
            //完成之后Netty会返回一个ChannelFuture,它的功能类似于JDK的java.util.concurrent.Future,主要用于异步操作的通知回调。
            ChannelFuture f = b.bind(port).sync();

            //等待服务端监听端口关闭
            //使用f.channel().closeFuture().sync()方法进行阻塞,等待服务端链路关闭之后main函数才退出。
            f.channel().closeFuture().sync();
        } finally {
            //优雅退出,释放线程池资源
            //调用NIO线程组的shutdownGracefully进行优雅退出,它会释放跟shutdownGracefully相关联的资源。
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer {
        @Override
        protected void initChannel(Channel channel) throws Exception {
            channel.pipeline().addLast(new TimeServerHandler());
        }
    }

    public static void main(String[] args) throws Exception {
        new TimeServer().bind(8080);
    }
}

TimeServerHandler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

//TimeServerHandler继承自ChannelHandlerAdapter,它用于对网络事件进行读写操作
//通常我们只需要关注channelRead和exceptionCaught方法。
public class TimeServerHandler extends ChannelHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //做类型转换,将msg转换成Netty的ByteBuf对象。
        //ByteBuf类似于JDK中的java.nio.ByteBuffer 对象,不过它提供了更加强大和灵活的功能。
        ByteBuf buf = (ByteBuf) msg;
        //通过ByteBuf的readableBytes方法可以获取缓冲区可读的字节数,
        //根据可读的字节数创建byte数组
        byte[] req = new byte[buf.readableBytes()];
        //通过ByteBuf的readBytes方法将缓冲区中的字节数组复制到新建的byte数组中
        buf.readBytes(req);
        //通过new String构造函数获取请求消息。
        String body = new String(req, "UTF-8");
        System.out.println("The time server receive order : " + body);
        //如果是"QUERY TIME ORDER"则创建应答消息,
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
                System.currentTimeMillis()).toString() : "BAD ORDER";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        //通过ChannelHandlerContext的write方法异步发送应答消息给客户端。
        ctx.write(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //调用了ChannelHandlerContext的flush方法,它的作用是将消息发送队列中的消息写入到SocketChannel中发送给对方。
        //从性能角度考虑,为了防止频繁地唤醒Selector进行消息发送,
        //Netty的write方法并不直接将消息写入SocketChannel中,调用write方法只是把待发送的消息放到发送缓冲数组中,
        //再通过调用flush方法,将发送缓冲区中的消息全部写到SocketChannel中。
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        //当发生异常时,关闭ChannelHandlerContext,释放和ChannelHandlerContext相关联的句柄等资源。
        ctx.close();
    }
}

  

 

客户端代码示例:

TimeClient

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class TimeClient {

    public void connect(int port, String host) throws Exception {
        // 配置客户端NIO线程组
        //首先创建客户端处理I/O读写的NioEventLoop Group线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //继续创建客户端辅助启动类Bootstrap,随后需要对其进行配置。
            //与服务端不同的是,它的Channel需要设置为NioSocketChannel
            //然后为其添加handler,此处为了简单直接创建匿名内部类,实现initChannel方法,
            //其作用是当创建NioSocketChannel成功之后,
            //在初始化它的时候将它的ChannelHandler设置到ChannelPipeline中,用于处理网络I/O事件。
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            channel.pipeline().addLast(new TimeClientHandler());
                        }
                    });

            // 发起异步连接操作
            //客户端启动辅助类设置完成之后,调用connect方法发起异步连接,
            //然后调用同步方法等待连接成功。
            ChannelFuture f = b.connect(host, port).sync();

            // 等待客户端链路关闭
            //当客户端连接关闭之后,客户端主函数退出.
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出,释放NIO线程组
            //在退出之前,释放NIO线程组的资源。
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new TimeClient().connect(8080, "127.0.0.1");
    }
}

 

TimeClientHandler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

//重点关注三个方法:channelActive、channelRead和exceptionCaught。
public class TimeClientHandler extends ChannelHandlerAdapter {

    private final ByteBuf firstMessage;

    public TimeClientHandler() {
        byte[] req = "QUERY TIME ORDER".getBytes();
        firstMessage = Unpooled.buffer(req.length);
        firstMessage.writeBytes(req);
    }

    //当客户端和服务端TCP链路建立成功之后,Netty的NIO线程会调用channelActive方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        //发送查询时间的指令给服务端
        //调用ChannelHandlerContext的writeAndFlush方法将请求消息发送给服务端。
        ctx.writeAndFlush(firstMessage);
    }

    //当服务端返回应答消息时,channelRead方法被调用
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
        //从Netty的ByteBuf中读取并打印应答消息。
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");
        System.out.println("Now is : " + body);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 释放资源
        //当发生异常时,打印异常日志,释放客户端资源。
        ctx.close();
    }
}

  

maven依赖:

 <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>5.0.0.Alpha1</version>
        </dependency>

  

 

 

 

 

 

 

 

来源:https://www.cnblogs.com/wade-luffy/p/6165626.html

 

TimeClient