netty实现简单聊天室功能

首先看一下项目目录结构

从目录结构看，整个项目比较简单，总体由一个服务端启动器、客户端启动器、以及两个Handler组成，下面来逐个看一下代码。

ServerStarter 聊天室服务端启动代码

package com.lhb.chatRoom;

import com.lhb.chatRoom.handler.NettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
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;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

/**
 * @Class ServerStarter
 * @Description 服务启动类
 * @Author lihuibin
 * @Date 2021/10/10 17:00
 * @Version 1.0
 * 更新记录:
 * 更新时间        更新人        更新位置        更新内容
 * 2021/10/10       lihuibin       新建           新建
 */
public class ServerStarter {
    public static void main(String[] args) throws InterruptedException {
        // 创建两个线程组bossGroup和workerGroup，含有的子线程NioEventLoop的个数默认为cpu核数的两倍
        // bossGroup只是处理连接请求，真正的和客户端业务处理，会交给workerGroup完成
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 创建服务器端的启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 使用链式编程来配置参数
            // 设置两个线程组
            bootstrap.group(bossGroup, workerGroup)
                    // 使用NioServerSocketChannel作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    // 初始化服务器连接队列大小,服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接
                    // 多个客户端同时来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //让客户端保持长期活动状态
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    // 创建通道初始化对象，设置初始化参数
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("encoder",new StringEncoder());
                            pipeline.addLast("decoder",new StringDecoder());
                            // 对workerGroup的SocketChannel设置处理器
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("netty server start....");
            // 绑定一个端口并且同步，生成了一个ChannelFuture异步对象，通过isDone()等方法可以判断异步事件的执行情况
            // 启动服务其（并绑定端口），bind是异步操作，sync方法是等待异步操作执行完毕
            ChannelFuture cf = bootstrap.bind(9000).sync();
            // 给cf注册监听器，监听器我们关心的事件
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        System.out.println("监听端口9000成功");
                    } else {
                        System.out.println("监听端口9000失败");
                    }
                }
            });
            // 对通道关闭进行监听，closeFuture是异步操作，监听通道关闭
            // 通过sync方法同步等待通道关闭处理完毕，这里会阻塞等待通道关闭完成
            cf.channel().closeFuture().sync();
            cf.addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future)
                        throws Exception {
                    if(future.isCancelled()){
                        System.out.println("服务器正在关闭..");
                    }
                    if(future.isCancellable()){
                        System.out.println("服务器已经关闭..OK");
                    }
                }
            });
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

NettyServerHandler 服务端处理器

package com.lhb.chatRoom.handler;

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;

/**
 * @Class NettyServerHandler
 * @Description 自定义Handler需要继承netty规定好的某个HandlerAdapter（规范）
 * @Author lihuibin
 * @Date 2021/10/10 17:55
 * @Version 1.0
 * 更新记录:
 * 更新时间        更新人        更新位置        更新内容
 * 2021/10/10       lihuibin       新建           新建
 */
public class NettyServerHandler extends SimpleChannelInboundHandler<String> {

    // GlobalEventExcutor.INSTANCE是全局的时间执行器，是一个单例
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    /**
     * 读取客户端发送的数据
     * @param ctx 上下文对象，含有通道channel，管道pipline
     * @param msg 就是客户端发送的数据
     * @throws Exception
     */
//    @Override
//    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        if (!channelGroup.isEmpty()) {
//            for (Channel channel : channelGroup) {
//                if (channel != ctx.channel()) {
//                    channel.writeAndFlush("【"+ ctx.channel().remoteAddress() +"】: " + msg + "\n");
//                } else {
//                    channel.writeAndFlush("【发送消息】: " + msg + "\n");
//                }
//            }
//        }
//    }

    /**
     * 读取客户端发送的数据
     * @param channelHandlerContext
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        if (!channelGroup.isEmpty()) {
            for (Channel channel : channelGroup) {
                if (channel != channelHandlerContext.channel()) {
                    channel.writeAndFlush("【"+ channelHandlerContext.channel().remoteAddress() +"】: " + msg + "\n\r");
                } else {
                    channel.writeAndFlush("【发送消息】: " + msg + "\n\r");
                }
            }
        }


    }

    /**
     * 数据读取完毕处理方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    }

    /**
     * 处理异常，一般是需要关闭通道
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " 上线了");
        if (!channelGroup.isEmpty()) {
            channelGroup.writeAndFlush("【"+ ctx.channel().remoteAddress() +"】 " + sdf.format(new Date(System.currentTimeMillis())) + " 上线了");
        }
        channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (!channelGroup.isEmpty()) {
            channelGroup.remove(ctx.channel());
            channelGroup.writeAndFlush(ctx.channel().remoteAddress() + " 下线了");
        }
    }
}

ClientStarter客户端启动器代码

package com.lhb.chatRoom;

import com.lhb.chatRoom.handler.NettyClientHandler;
import com.lhb.chatRoom.test.ClientMessageHandler;
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 io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.Scanner;

/**
 * @Class ClientStarter
 * @Description 客户端启动类
 * @Author lihuibin
 * @Date 2021/10/10 18:07
 * @Version 1.0
 * 更新记录:
 * 更新时间        更新人        更新位置        更新内容
 * 2021/10/10       lihuibin       新建           新建
 */
public class ClientStarter {
    public static void main(String[] args) throws InterruptedException {
        // 客户端需要一个事件循环组
        NioEventLoopGroup group = new NioEventLoopGroup();
        try{
            // 创建客户端启动对象
            // 注意客户端使用的不是ServerBootstrap 而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            // 设置相关参数
            // 设置线程组
            bootstrap.group(group)
                    // 使用NioSocketChannel作为客户端的通道实现
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //加入处理器
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast("encoder",new StringEncoder());
                            pipeline.addLast("decoder",new StringDecoder());
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("netty client start...");
            // 启动客户端去链接服务器端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",9000).sync();

            channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {

                @Override
                public void operationComplete(Future<? super Void> future)
                        throws Exception {
                    if(future.isSuccess()){
                        System.out.println("客户端启动中...");
                    }
                    if(future.isDone()){
                        System.out.println("客户端启动成功...OK！");
                    }
                }
            });
            System.out.println("*******************************");
            Channel channel = channelFuture.channel();
            //获取channel
            Scanner scanner = new Scanner(System.in);
            while(scanner.hasNextLine()){
                String str = scanner.nextLine();
                channel.writeAndFlush(str);
            }
            channelFuture.channel().closeFuture().sync();
            scanner.close();
        } finally {
            group.shutdownGracefully();
        }
    }
}

NettyClientHandler客户端处理器

package com.lhb.chatRoom.handler;

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

/**
 * @Class NettyClientHandler
 * @Description 客户端处理
 * @Author lihuibin
 * @Date 2021/10/10 18:14
 * @Version 1.0
 * 更新记录:
 * 更新时间        更新人        更新位置        更新内容
 * 2021/10/10       lihuibin       新建           新建
 */
public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
    /**
     * 当客户端连接服务器完成就会触发该方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
//        System.out.println("channelActive");
    }

    /**
     * 当通道有读取事件时就会触发，即服务端发送数据给客户端
     * @param ctx
     * @param msg
     * @throws Exception
     */
//    @Override
//    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        System.out.println(msg);
//    }

    /**
     * 当通道有读取事件时就会触发，即服务端发送数据给客户端
     * @param channelHandlerContext
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println(s);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

启动程序验证

启动一个ServerStarter服务端和三个ClientStarter客户端