Netty入门学习

Z时代
2024-01-10
分类：综合

Netty入门

Netty介绍

Netty 是一个广泛使用的 Java 网络编程框架。

Netty 技术和方法的特点

设计
- 针对多种传输类型的统一接口 - 阻塞和非阻塞
- 简单但更强大的线程模型
- 真正的无连接的数据报套接字支持
- 链接逻辑支持复用

易用性
- 大量的 Javadoc 和代码实例
- 除了在 JDK 1.6 + 额外的限制。（一些特征是只支持在Java 1.7 +。可选的功能可能有额外的限制。）

性能
- 比核心 Java API 更好的吞吐量，较低的延时
- 资源消耗更少，这个得益于共享池和重用
- 减少内存拷贝

健壮性
- 消除由于慢，快，或重载连接产生的 OutOfMemoryError
- 消除经常发现在 NIO 在高速网络中的应用中的不公平的读/写比

安全
- 完整的 SSL / TLS 和 StartTLS 的支持
- 运行在受限的环境例如 Applet 或 OSGI

社区
- 发布的更早和更频繁
- 社区驱动

Netty基本概念

BOOTSTRAP

Netty 应用程序通过设置 bootstrap（引导）类的开始，该类提供了一个用于应用程序网络层配置的容器。

CHANNEL

底层网络传输 API 必须提供给应用 I/O操作的接口，如读，写，连接，绑定等等。对于我们来说，这是结构几乎总是会成为一个“socket”。 Netty 中的接口 Channel 定义了与 socket 丰富交互的操作集：bind, close, config, connect, isActive, isOpen, isWritable, read, write 等等。 Netty 提供大量的 Channel 实现来专门使用。这些包括 AbstractChannel，AbstractNioByteChannel，AbstractNioChannel，EmbeddedChannel， LocalServerChannel，NioSocketChannel 等等。

CHANNELHANDLER

ChannelHandler 支持很多协议，并且提供用于数据处理的容器。我们已经知道 ChannelHandler 由特定事件触发。 ChannelHandler 可专用于几乎所有的动作，包括将一个对象转为字节（或相反），执行过程中抛出的异常处理。

常用的一个接口是 ChannelInboundHandler，这个类型接收到入站事件（包括接收到的数据）可以处理应用程序逻辑。当你需要提供响应时，你也可以从 ChannelInboundHandler 冲刷数据。一句话，业务逻辑经常存活于一个或者多个 ChannelInboundHandler。

CHANNELPIPELINE

ChannelPipeline 提供了一个容器给 ChannelHandler 链并提供了一个API 用于管理沿着链入站和出站事件的流动。每个 Channel 都有自己的ChannelPipeline，当 Channel 创建时自动创建的。 ChannelHandler 是如何安装在 ChannelPipeline？主要是实现了ChannelHandler 的抽象 ChannelInitializer。ChannelInitializer子类通过 ServerBootstrap 进行注册。当它的方法 initChannel() 被调用时，这个对象将安装自定义的 ChannelHandler 集到 pipeline。当这个操作完成时，ChannelInitializer 子类则从 ChannelPipeline 自动删除自身。

EVENTLOOP

EventLoop 用于处理 Channel 的 I/O 操作。一个单一的 EventLoop通常会处理多个 Channel 事件。一个 EventLoopGroup 可以含有多于一个的 EventLoop 和提供了一种迭代用于检索清单中的下一个。

CHANNELFUTURE

Netty 所有的 I/O 操作都是异步。因为一个操作可能无法立即返回，我们需要有一种方法在以后确定它的结果。出于这个目的，Netty 提供了接口 ChannelFuture,它的 addListener 方法注册了一个 ChannelFutureListener ，当操作完成时，可以被通知（不管成功与否）。

TCP示例

用几个简单的示例，展示一些常用的几种连接方式

Netty-服务端实现（TCP server）

public class TcpServer {
    public static void main(String[] args) throws Exception {
        //负责建立连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //负责干活的，处理数据
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .localAddress(
                        //绑定本地8888端口
                        new InetSocketAddress(8888))
                //增加一个消息处理器
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new ServerHandler());
                    }
                });
        //异步方式建立连接
        ChannelFuture f = b.bind().sync();
        f.channel().closeFuture().sync();
    }
    /**
     * 实现一个简单的消息处理器，当收到客户端发送来的消息后，回复一个收到！
     */
    static class ServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            ByteBuf copy = msg.copy();
            System.out.println("服务端收到消息: " + copy.toString(CharsetUtil.UTF_8));
            // 回复收到！
            ctx.writeAndFlush(Unpooled.copiedBuffer("收到！！", CharsetUtil.UTF_8));
        }
    }
}

Netty-客户端实现（TCP client）

public class TcpClient {
    public static void main(String[] args) throws Exception {
        //建立一个客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(bossGroup)
                .channel(NioSocketChannel.class)
                .remoteAddress(
                        //连接到刚刚我们创建的服务端上
                        new InetSocketAddress("127.0.0.1", 8888))
                //增加一个消息处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new ClientHandler());
                    }
                });
        //异步方式建立连接
        ChannelFuture f = b.connect().sync();
        f.channel().closeFuture().sync();
    }
    /**
     * 简单的消息处理器
     */
    static class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
        /**
         * 当建立连接时，会回调这个函数。
         * 与服务端建立连接后，上报一条消息
         *
         * @param ctx 建立的通道
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            ctx.writeAndFlush(Unpooled.copiedBuffer("我是客户端，发送上线消息", CharsetUtil.UTF_8));
        }
        /**
         * 接收来自服务器返回的消息
         *
         * @param ctx 建立的通道
         * @param in  服务器返回的消息
         */
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
                                 ByteBuf in) {
            System.out.println("客户端收到服务端的消息:" + in.toString(CharsetUtil.UTF_8));
        }
    }
}

UDP示例