Netty-核心模块组件-4

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

Netty 核心模块组件

一、Bootstrap、ServerBootstrap

1、Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序，串联各个组件Netty中

Bootstrap 类是客户端程序的启动引导类

ServerBootstrap 是服务端启动引导类

2、常见的方法

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)，该方法用于服务器端，用来设置两个 EventLoop
public B group(EventLoopGroup group) ，该方法用于客户端，用来设置一个 EventLoop
public B channel(Class<? extends C> channelClass)，该方法用来设置一个服务器端的通道实现
public < T > B option(ChannelOption option, T value)，用来给 ServerChannel 添加配置
public < T > ServerBootstrap childOption(ChannelOption childOption, T value)，用来给接收到的通道添加配置
public ServerBootstrap childHandler(ChannelHandler childHandler)，该方法用来设置业务处理类（自定义的handler）针对WorkerGroup（干活的类）
public ServerBootstrap handler(ChannelHandler handler)针对BoosGroup（专门负责接收客户端请求的类）
public ChannelFuture bind(int inetPort) ，该方法用于服务器端，用来设置占用的端口号
public ChannelFuture connect(String inetHost, int inetPort) ，该方法用于客户端，用来连接服务器端

二、Future、ChannelFuture

Netty 中所有的 IO 操作都是异步的，不能立刻得知消息是否被正确处理。

但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过 Future 和 ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发注册的监听事件。

常见的方法：

Channel channel()，返回当前正在进行 IO 操作的通道

ChannelFuture sync()，等待异步操作执行完毕

三、Channel

1、Netty 网络通信的组件，能够用于执行网络 I/O 操作

2、通过 Channel 可获得当前网络连接的通道的状态

3、通过 Channel 可获得 网络连接的配置参数 （例如接收缓冲区大小）

4、Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成

5、调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以 I/O 操作成功、失败或取消时回调通知调用方

6、支持关联 I/O 操作与对应的处理程序

7、不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，常用的 Channel 类型

NioSocketChannel，异步的客户端 TCP Socket 连接。

NioServerSocketChannel，异步的服务器端 TCP Socket 连接。

NioDatagramChannel，异步的 UDP 连接。

NioSctpChannel，异步的客户端 Sctp 连接。

NioSctpServerChannel，异步的 Sctp 服务器端连接，这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

四、Selector

1、Netty 基于 Selector 对象实现 I/O 多路复用，通过 Selector 一个线程可以监听多个连接的 Channel 事件。

2、当向一个 Selector 中注册 Channel 后，Selector 内部的机制就可以自动不断地查询(Select) 这些注册的Channel 是否有已就绪的 I/O 事件（例如可读，可写，网络连接完成等），这样程序就可以很简单地使用一个线程高效地管理多个 Channel

五、ChannelHandler 及其实现类

1、ChannelHandler 是一个接口，处理 I/O 事件或拦截 I/O 操作，并将其转发到其 ChannelPipeline(业务处理链)中的下一个处理程序。

2、ChannelHandler 本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类

3、ChannelHandler 及其实现类一览图

4、我们经常需要自定义一个 Handler 类去继承 ChannelInboundHandlerAdapter

5、然后通过重写相应方法实现业务逻辑，我们接下来看看一般都需要重写哪些方法

六、Pipeline 和 ChannelPipeline

1、ChannelPipeline 是一个重点

1、ChannelPipeline 是一个 Handler 的集合

2、它负责处理和拦截 inbound 或者 outbound 的事件和操作，相当于一个贯穿 Netty 的链。(也可以这样理解：ChannelPipeline 是保存 ChannelHandler 的 List，用于处理或拦截Channel 的入站事件和出站操作)

3、ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及 Channel中各个的 ChannelHandler 如何相互交互

4、在 Netty 中每个 Channel 都有且仅有一个 ChannelPipeline 与之对应，它们的组成关系如下

channel能拿到他对应的channelPipeline

channelPipeline也可以获取到对应的channel

channelPipeline中包含一个个的ChannelHandlerContext的双向链表

每个ChannelHandlerContext（保存 Channel 相关的所有上下文信息）里面包含对应具体的channelHandler

2、常用方法

1、ChannelPipeline addFirst(ChannelHandler… handlers) 把一个业务处理类（handler）添加到链中的第一个位置

2、ChannelPipeline addLast(ChannelHandler… handlers) 把一个业务处理类（handler）添加到链中的最后一个位置

七、ChannelHandlerContext

1、保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象

2、即 ChannelHandlerContext 中包含一个具体的事件处理器 ChannelHandler ，同时ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息，方便对 ChannelHandler 进行调用.

3、常用方法

八、ChannelOption

1、Netty 在创建 Channel 实例后,一般都需要设置 ChannelOption 参数。

2、ChannelOption 参数如下:

九、EventLoopGroup 和其实现类 NioEventLoopGroup

1、EventLoopGroup 是一组 EventLoop（就是对应线程） 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop同时工作，每个 EventLoop 维护着一个 Selector 实例。

2、EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop 来处理任务。

3、在 Netty服务器端编程中，我们一般都需要提供两个 EventLoopGroup ，例如：

BossEventLoopGroup

WorkerEventLoopGroup

4、通常一个服务端口即一个 ServerSocketChannel 对应一个 Selector 和一个 EventLoop 线程。

5、服务端中，BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理，如下图所示

6、常用方法：

public NioEventLoopGroup()，构造方法

public Future<?> shutdownGracefully()，断开连接，关闭线程

十、 Unpooled 类

1、Netty 提供一个专门用来操作缓冲区(即 Netty 的数据容器)的工具类

2、内部维护了对应的readerIndex和writerIndex

3、相比NIO的ByteBuffer，Netty 提供的ByteBuf不用考虑flip反转去操作读写

4、常用方法如下所示

5、举例说明 Unpooled 获取 Netty 的数据容器 ByteBuf 的基本使用

6、案例一

package com.sun.netty.components;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

/**
 * @Author: sunguoqiang
 * @Description: TODO
 * @DateTime: 2023/1/4 17:14
 **/
public class NettyByteBuf01 {
    public static void main(String[] args) {
        /**
         *  创建一个 ByteBuf
         *  说明:
         *  1. 创建 对象，该对象包含一个数组 arr , 是一个 byte[10]
         *  2. 在 netty 的 buffer 中，不需要使用 flip 进行反转 , 底层维护了 readerindex 和 writerIndex
         *  3. 通过 readerindex 和 writerIndex 和 capacity， 将 buffer 分成三个区域
         *      1、0---readerindex 已经读取的区域、
         *      2、readerindex---writerIndex，可读的区域、
         *      3、writerIndex---capacity, 可写的区域
         */
        ByteBuf buffer = Unpooled.buffer(10);
        for (int i = 0; i < 10; i++) {
            buffer.writeByte(i);
        }
        System.out.println("capacity=" + buffer.capacity());//10
        // 输出
        for (int i = 0; i < buffer.capacity(); i++) { // 此方式readerindex不会改变
            System.out.println(buffer.getByte(i));
        }
        for (int i = 0; i < buffer.capacity(); i++) { // 此方式readerindex会改变
            System.out.println(buffer.readByte());
        }
        System.out.println("执行完毕");
    }
}

7、案例二

package com.sun.netty.components;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.nio.charset.Charset;

/**
 * @Author: sunguoqiang
 * @Description: TODO
 * @DateTime: 2023/1/4 17:14
 **/
public class NettyByteBuf02 {
    public static void main(String[] args) {
        // 创建 ByteBuf
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello,world!", Charset.forName("utf-8"));
        // 使用相关的方法
        if (byteBuf.hasArray()) { // true
            byte[] content = byteBuf.array();
            // 将 content 转成字符串
            System.out.println(new String(content, Charset.forName("utf-8")));
            // 输出ByteBuf对象
            System.out.println("byteBuf=" + byteBuf);
            // 输出byteBuf偏移量
            System.out.println(byteBuf.arrayOffset()); // 0
            // byteBuf读取索引位置
            System.out.println(byteBuf.readerIndex()); // 0
            // byteBuf写入索引位置
            System.out.println(byteBuf.writerIndex()); // 12
            // byteBuf容量
            System.out.println(byteBuf.capacity());   // 36
            // byteBuf可读取的第一个字节
            System.out.println(byteBuf.readByte()); // 104
            // byteBuf指定索引位置的字节
            System.out.println(byteBuf.getByte(0)); // 104
            // byteBuf可读取字节数
            int len = byteBuf.readableBytes(); // 12
            System.out.println("len=" + len);
            //使用 for 取出各个字节
            for (int i = 0; i < len; i++) {
                System.out.println((char) byteBuf.getByte(i));
            }
            // 按照某个范围读取
            System.out.println(byteBuf.getCharSequence(0, 4, Charset.forName("utf-8")));
            System.out.println(byteBuf.getCharSequence(4, 6, Charset.forName("utf-8")));
        }
    }
}

十一、Netty 应用实例-群聊系统

1、实例要求

1、编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2、实现多人群聊
3、服务器端：可以监测用户上线，离线，并实现消息转发功能
4、客户端：通过 channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到)
5、目的：进一步理解 Netty 非阻塞网络编程机制

2、服务端

GroupChatServer 服务端

package com.sun.netty.groupChat;

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.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * @Author: sunguoqiang
 * @Description: 群聊系统服务端
 * @DateTime: 2023/1/9 14:25
 **/
public class GroupChatServer {
    // 监听端口
    private final Integer port;

    public GroupChatServer(Integer port) {
        this.port = port;
    }

    public void run() throws InterruptedException {
        // 创建两个线程组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 向pipeline加入处理器
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            // 如果不加这个编码解码器的  无法直接传输字符串
                            pipeline.addLast("encoder", new StringEncoder());
                            pipeline.addLast("myHandler", new GroupChatServerHandler());
                        }
                    });
            System.out.println("服务器启动完成,绑定端口:" + port);
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            // 监听关闭时间
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        GroupChatServer groupChatServer = new GroupChatServer(8989);
        groupChatServer.run();
    }

}

View Code

GroupChatServerHandler 服务端处理器

package com.sun.netty.groupChat;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: sunguoqiang
 * @Description: 自定义服务器处理器
 * @DateTime: 2023/1/9 14:26
 **/
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    //定义管理每个客户端的channel组
    //GlobalEventExecutor.INSTANCE 全局的事件执行器，单例的
    private static final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

    // 当连接建立会第一个执行该方法，【客户端连接事件】
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 当客户端连接，第一时间将每个客户端的channel加入到channelGroup统一管理
        Channel channel = ctx.channel();
        // 给当前channelGroup管理的所有channel的客户端都发送消息
        channelGroup.writeAndFlush(sdf.format(new Date())+"[客户端]"+channel.remoteAddress()+"进入聊天室...");
        channelGroup.add(channel);
    }

    // channel活跃时触发,监测上线
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(sdf.format(new Date())+"[客户端]"+channel.remoteAddress()+"上线...");
//        channelGroup.writeAndFlush(sdf.format(new Date())+"[客户端]"+channel.remoteAddress()+"上线...");
    }

    // 当某个channel处于活动状态，就会触发，用于发送某某上线【客户端上线活动状态事件】
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush(sdf.format(new Date())+"[客户端]"+channel.remoteAddress()+"离线...");
    }

    // 当某个channel离开状态，就会触发，用于发送某某下线【客户端离开非活动状态事件】
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush(sdf.format(new Date())+"[客户端]"+channel.remoteAddress()+"离开聊天室...");
    }

    // 客户端发送消息会触发【读取客户端数据事件】
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        // 获取当前发送消息的用户
        Channel channel = channelHandlerContext.channel();
        // 排除发送者自己，不给他转发消息
        for (Channel channel1 : channelGroup) {
            if(channel.equals(channel1)){
                channel1.writeAndFlush(sdf.format(new Date()) + "[自己发送]："+msg);
            }else{
                channel1.writeAndFlush(sdf.format(new Date()) + "[客户端-"+channel.remoteAddress()+"]："+msg);
            }
        }
    }

    // 当发生异常会触发【异常触发事件】
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        channel.close();
    }
}

View Code

3、客户端

GroupChatClient 客户端

package com.sun.netty.groupChat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;

/**
 * @Author: sunguoqiang
 * @Description: 群聊系统客户端
 * @DateTime: 2023/1/9 14:59
 **/
public class GroupChatClient {

    private final String ip;
    private final Integer port;

    public GroupChatClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }

    public void run() throws InterruptedException {
        NioEventLoopGroup clientGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(clientGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            pipeline.addLast("myHandler", new GroupChatClientHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect(ip, port);
            // 监听连接是否成功
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        System.out.println("连接成功...");
                    } else {
                        System.out.println("连接失败...");
                    }
                }
            });
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String nextLine = scanner.nextLine();
                channelFuture.channel().writeAndFlush(nextLine);
            }
            channelFuture.channel().close().sync();
        } finally {

        }
    }

    public static void main(String[] args) throws InterruptedException {
        GroupChatClient groupChatClient = new GroupChatClient("127.0.0.1", 8989);
        groupChatClient.run();
    }
}

View Code

GroupChatClientHandler 客户端处理器

package com.sun.netty.groupChat;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * @Author: sunguoqiang
 * @Description: 聊天室客户端处理器
 * @DateTime: 2023/1/9 14:59
 **/
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
    // 读取服务端发来的消息【读写消息事件】
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        Channel channel = channelHandlerContext.channel();
        System.out.println(msg);
    }
}

View Code

十二、Netty 心跳检测机制案例

1、实例要求

编写一个 Netty 心跳检测机制案例, 当服务器超过 3 秒没有读时，就提示读空闲

当服务器超过 5 秒没有写操作时，就提示写空闲

实现当服务器超过 7 秒没有读或者写操作时，就提示读写空闲

2、服务端

package com.sun.netty.heartbeat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @Author: sunguoqiang
 * @Description: TODO
 * @DateTime: 2023/1/10 11:29
 **/
public class MyHeartBeatServer {

    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            /**
                             * 加入netty提供的IdleStateHandler
                             * IdleStateHandler；空闲状态处理器
                             * 构造参数↓
                             *  long readerIdleTime : 表示多长时间没读操作了，服务端没有接受到客户端的读操作；(触发后，服务端就会发送心跳检测包给客户端，检测是否还是连接状态)
                             *     long writerIdleTime : 表示有多长时间没有写操作了，服务端没有接收到客户端的写操作
                             *     long allIdleTime : 表示多长时间既没有读操作也没有写操作。
                             *     TimeUnit unit : 时间单位
                             */
                            pipeline.addLast(new IdleStateHandler(3, 5, 8, TimeUnit.SECONDS));
                            /**
                             * 加入一个对空闲状态检测进一步处理的handler
                             * 时间到了就会触发对应的事件(IdleStateEvent)，我们再写一个自定义的handler来监听这个时间就可以做出对应的操作
                             * 当事件触发之后，就会传递给对应的channel管道的下一个handler去处理
                             * 通过调用(触发)下一个handler的 userEventTiggerd()方法，该方法去处理
                             */
                            pipeline.addLast(new MyHeartBeatServerHandler());
                        }
                    });
            System.out.println("server start ok ...");
            ChannelFuture channelFuture = serverBootstrap.bind(7878).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端

package com.sun.netty.heartbeat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * @Author: sunguoqiang
 * @Description: 心跳检测服务端处理器，因为不需要接受数据，所以就直接继承 ChannelInboundHandlerAdapter
 * @DateTime: 2023/1/10 11:35
 **/
public class MyHeartBeatServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 当触发心跳事件后，会触发该方法
     * @param ctx 上下文
     * @param evt 事件
     * @throws Exception 异常
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            switch (event.state()) {
                case ALL_IDLE:
                    System.out.println("【读写空闲】事件");
                    break;
                case READER_IDLE:
                    System.out.println("【读空闲】事件");
                    break;
                case WRITER_IDLE:
                    System.out.println("【写空闲】事件");
            }
            System.out.println(ctx.channel().remoteAddress() + "发生事件:" + event.state());
//            ctx.channel().close();
        }
    }
}

心跳检测服务端处理器

3、客户端

package com.sun.netty.heartbeat;

import com.sun.netty.groupChat.GroupChatClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;

/**
 * @Author: sunguoqiang
 * @Description: 群聊系统客户端
 * @DateTime: 2023/1/9 14:59
 **/
public class MyHeartBeatClient {

    private final String ip;
    private final Integer port;

    public MyHeartBeatClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }

    public void run() throws InterruptedException {
        NioEventLoopGroup clientGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(clientGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            pipeline.addLast("myHandler", new GroupChatClientHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect(ip, port);
            // 监听连接是否成功
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        System.out.println("连接成功...");
                    } else {
                        System.out.println("连接失败...");
                    }
                }
            });
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String nextLine = scanner.nextLine();
                channelFuture.channel().writeAndFlush(nextLine);
            }
            channelFuture.channel().close().sync();
        } finally {

        }
    }

    public static void main(String[] args) throws InterruptedException {
        MyHeartBeatClient myHeartBeatClient = new MyHeartBeatClient("127.0.0.1", 7878);
        myHeartBeatClient.run();
    }
}

客户端

十三、Netty 通过 WebSocket 编程实现服务器和客户端长连接

1、实例要求

1、Http 协议是无状态的, 浏览器和服务器间的请求响应一次，下一次会重新创建连接.

2、要求：实现基于 webSocket 的长连接的全双工的交互

3、改变 Http 协议多次请求的约束，实现长连接了，服务器可以发送消息给浏览器

4、客户端浏览器和服务器端会相互感知，比如服务器关闭了，浏览器会感知，同样浏览器关闭了，服务器会感知

5、运行界面

2、代码

package com.sun.netty.websocket;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import java.time.LocalDateTime;

/**
 * @Title: WebSocketServerFrameHandler
 * @Author sunguoqiang
 * @Package com.sun.netty.websocket
 * @Date 2023/1/14 11:22
 * @description: * 自定义处理业务逻辑的处理器
 * * 对于websocket是通过【帧frame】的形式传递的（数据链路层的传输单元是帧）
 * * TextWebSocketFrame：表示一个文本帧，就是传输过程中的数据
 */
public class WebSocketServerFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("服务端收到消息：" + msg.text());
        //回复浏览器
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间：" + LocalDateTime.now() + ":" + msg.text()));
    }

    //客户端断开连接时会触发该事件
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        //id.asLongText表示获取这channel唯一的值
        System.out.println("handlerRemoved被调用：" + ctx.channel().id().asLongText());
    }

    //客户端连接时会触发该事件
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        //id.asLongText表示获取这channel唯一的值
        System.out.println("handlerAdded被调用：" + ctx.channel().id().asLongText());
        //id.asShortText表示获取这channel的值，这个不是唯一的。有可能重复
        System.out.println("handlerAdded被调用：" + ctx.channel().id().asShortText());
    }

    //当客户端发生异常时会触发该事件
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常发生：" + cause.getMessage());
        ctx.channel().close();
    }
}

WebSocketServerFrameHandler自定义处理业务逻辑的处理器

package com.sun.netty.websocket;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

/**
 * @Author: sunguoqiang
 * @Description: Websocket双工TCP长连接
 * @DateTime: 2023/1/12 9:59
 **/
public class WebsocketServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup boosGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boosGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))//netty自带的日志处理器
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //基于http协议，使用http的编码和解码器
                            pipeline.addLast(new HttpServerCodec());
                            //是以 块方式 写，添加ChunkedWriteHandler处理器
                            pipeline.addLast(new ChunkedWriteHandler());
                            /**
                             * 说明：
                             *    1、因为http的数据在传输过程中是【分段】的，HttpObjectAggregator可以将多个分段聚合起来
                             *    2、这就是当浏览器发送大量数据时，就会发送多次http请求
                             */
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            /**
                             * 说明：
                             *    1、对于websocket是通过【帧frame】的形式传递的（数据链路层的传输单元是帧）
                             *    2、可以看到WebSocketFrame 下面有6个子类
                             *    3、浏览器发送文件、请求时，ws://localhost:7000/websocket 表示请求的uri
                             *    4、WebSocketServerProtocolHandler核心功能：将http协议升级为ws协议（websocket长连接协议）
                             *    5、为什么http能够升级为ws协议呢？是通过状态码101
                             */
                            pipeline.addLast(new WebSocketServerProtocolHandler("/websocket"));
                            //自定义处理业务逻辑的处理器
                            pipeline.addLast(new WebSocketServerFrameHandler());
                        }
                    });
            System.out.println("服务端绑定端口7000.....启动成功");

            //启动服务端
            ChannelFuture cf = serverBootstrap.bind(7000).sync();
            cf.channel().closeFuture().sync();
        } finally {
            boosGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

Websocket双工TCP长连接—服务端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title><!DOCTYPE html>
        <html lang="en">
        <head>
        <meta charset="UTF-8">
        <title>Title</title>
</head>
<script>
    var socket;
    //判断当前浏览器是否支持websocket编程
    if (window.WebSocket) {
        socket = new WebSocket("ws://localhost:7000/websocket");
        //相当于服务器中的channelRead0事件【客户端读取服务端数据事件】，msg为服务端发来的数据
        socket.onmessage = function (resp) {
            var rt = document.getElementById('responseText');
            rt.value = rt.value + "\n" + resp.data
        }
        //相当于客户端与服务端连接开启的事件
        socket.onopen = function (resp) {
            var rt = document.getElementById('responseText');
            rt.value = "连接已开启";
        }
        //相当于客户端与服务端连接关闭的事件
        socket.onclose = function (resp) {
            var rt = document.getElementById('responseText');
            rt.value = rt.value + "\n" + "连接已关闭";
        }
    } else {
        alert("当前浏览器不支持websocket编程");
    }

    //发送消息给服务端的方法
    function send(msg) {
        //判断websocket是否创建好了
        if (!window.socket) {
            return
        }
        //判断当前状态是否已经连接开启
        if (socket.readyState == WebSocket.OPEN) {
            //通过socket发送消息
            socket.send(msg);
        } else {
            alert("连接未开启");
        }
    }
</script>
<body>
<form onsubmit="return false">
    <textarea name="message" style="height: 300px;width: 300px"></textarea>
    <input type="button" value="发送" onclick="send(this.form.message.value)">
    <textarea id="responseText" style="height: 300px;width: 300px"></textarea>
    <input type="button" value="清空" onclick="document.getElementById('responseText').value=''">
</form>

</body>
</html>