Netty 笔记

Netty 核心组件

• Channel
• 回调
• Future
• 事件和ChannelHandler

Channel

Channel 是 Java NIO 的一个基本构造，可以把它看作是传入或者传出数据的载体（通道），因此可以被打开或者关闭。

回调

回调函数就是一个通过函数指针调用的函数。如果你把函数的指针（地址）作为参数传递给另一个函数，当这个指针被用来调用其所指向的函数时，我们就说这是回调函数。

Netty 在内部是通过回调来处理事件；当事件发生时，Netty会回调用实现了 interface-ChannelHandler 的函数处理该事件。

Future

Future用于表示异步操作的结果，每个Netty 的出战 I/O 操作都会返回一个 ChannelFuture。

事件和 ChannelHandler

Netty 使用不同的事件来通知我们状态的改变或者操作的状态，我们可以通过发生的事件来触发相对于的动作。

发生的每个事件都会被分发给 ChannelHander 类中的某个使用户实现的方法。大部分情况我们都需要自己实现 ChannelHandler 然后通过通过重写方法实现自己的业务逻辑。

基于 Netty 的服务器和客户端

要做的很简单，就是：客户端发送消息给服务器，然后服务器接收客户端发动的消息并回复一条消息给客户端。

Netty 依赖

    <dependencies>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.32.Final</version>
        </dependency>
    </dependencies>

服务器

基于 Neety 的服务器需要两部分：

• ChannelHandler 实现类，我们需要通过实现它来接收客服端的数据（业务逻辑）
• 引导——服务器配置及启动代码（配置）

ChannelHandler 实现

package com.project.demo;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;


@Sharable // 标识 Channel-Handler 可以被多个 Channel 安全共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
    // 读取客服端消息
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 读取消息
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));
        
        // 将消息写给发送者, 并不发送，需要调用 flush 方法才会发送
        ctx.write(Unpooled.copiedBuffer("你好！", CharsetUtil.UTF_8));
    }
    
    // 一条消息读取完成(一条消息可能需要多次读取)时调用
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将消息发送给客服端，并关闭该 Channel
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
            .addListener(ChannelFutureListener.CLOSE);
    }
    
    // 在读取消息期间，发生异常时调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
        cause.printStackTrace();
        ctx.close();
    }
}

上面代码中，我们继承一个ChannelHandler的实现类 ChannelInboundHandlerAdapter

引导

ChannelHander 写好了，我还还需要做以下工作才能使服务器工作：

• 设置监听端口
• 配置 Channel，使入站消息发送给我们刚刚编写的 EchoServerHandler 处理。

package com.project.demo;

import java.net.InetSocketAddress;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;


/**
 * 1. 创建一个ServerBootstrap的实例以引导和绑定服务器
 * 2. 创建并分配一个NioEventLoopGroup实例以进行事件的处理
 * 3. 指定服务器绑定的本地的InetSocketAddress
 */
public class EchoServer {

    private int port;
    
    public EchoServer(int port) {
        this.port = port;
    }

    public void start() throws Exception {
        // 消息处理器实例
        final EchoServerHandler serverHandler = new EchoServerHandler();
        // Nio
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            // 配置
            b.group(group)
                .channel(NioServerSocketChannel.class)  // 指定使用的 NIO 
                .localAddress(new InetSocketAddress(port))  // 指定端口
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        // 注册自定义的 ChannelHandle 使之生效
                        ch.pipeline().addLast(serverHandler);
                    }
                });
            // 绑定服务器
            ChannelFuture f = b.bind().sync();
            // 获取 channel 的 CloseFuture
            f.channel().closeFuture().sync();
        } finally {
            // 关闭EventLoopGroup，释放所有的资源
            group.shutdownGracefully().sync();
        }
    }

    public static void main(String[] args) throws Exception {
        new EchoServer(8081).start();  // 启动
    }
}

客户端

客服端同服务器一样也需要编写两个部分的代码，ChannelHandler实现（业务逻辑）和引导（配置和启动）

ChannelHandler实现

package com.project.demo;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

// 客户端消息处理器
@Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
    // 接收数据时被调用
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("Chient received: " + msg.toString(CharsetUtil.UTF_8));
    }
    
    // 建立时被调用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
       // 发送消息到服务器
       ctx.writeAndFlush(Unpooled.copiedBuffer("Hello！", CharsetUtil.UTF_8));
    }
    
    // 处理过程中发生异常时被调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
       cause.printStackTrace();
       // 关闭 Channel
       ctx.close();
    }
}

引导

客户端的引导需要指定主机及端口

package com.project.demo;

import java.net.InetSocketAddress;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
 * 1. 创建了一个Bootstrap实例
 * 2. 为进行事件处理分配了一个NioEventLoopGroup实例
 * 3. 为服务器连接创建了一个InetSocketAddress实例
 * 4. 当连接建立时，EchoClientHandle 实例会被安装到 ChannelPipeline 中
 * 5. 配置完成后，通过调用Bootstrap.connect()方法连接到远程节点（服务器）
 */
public class EchoClient {

    private final String host;
    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host, port))  // 指定 host，port
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });
            // 连接到远程节点，阻塞等待直到连接完成
            ChannelFuture f = b.connect().sync();
            // 阻塞，直到 Channel 关闭
            f.channel().closeFuture().sync();
        } finally {
            // 关闭连接次并释放资源
            group.shutdownGracefully().sync();
        }
    }
    
    public static void main(String[] args) throws Exception {
        new EchoClient("127.0.0.1", 8081).start();
    }
}

到这里这个基于 Netty 的简单程序就编写完了，想测试是否成功可以先启动 EchoServer 然后再启动 EchoClient，客户端启动会发送一条 Hello 给服务器，服务器接收到消息后会回复 你好 给客户端。

Netty 常用组件

Netty网络抽象的代表：

• Channel——Socket;
• EventLoop——控制流、多线程处理、并发；
• ChannelFuture——异步通知。