Netty4（七）：实现聊天室

2018-03-16 本文已影响94人聪明的奇瑞

案例代码下载
初学者的话推荐直接套用 all-in-one 的 jar 包，若熟悉 Netty 可以根据需求添加不同的 jar 包

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

服务端

首先编写 Handler 类，继承自 SimpleChannelInboundHandler，SimpleChannelInboundHandler 能处理特定类型的消息，覆盖其几个方法：
- handlerAdded()：每当从服务端收到新的客户端连接时，将客户端的 Channel 存入 ChannelGroup 列表中，并通知列表中的其他客户端 Channel
- handlerRemoved()：每当从服务端收到客户端断开时，客户端的 Channel 自动从 ChannelGroup 列表中移除了，并通知列表中的其他客户端 Channel
- channelRead0()：每当从服务端读到客户端写入信息时，将信息转发给其他客户端的 Channel
- channelActive()：服务端监听到客户端活动
- channelInactive()：服务端监听到客户端不活动
- exceptionCaught()：出现异常时回调该方法

public class SimpleChatServerHandler extends SimpleChannelInboundHandler<String> {

    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {  // (2)
        Channel incoming = ctx.channel();
        channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
        channels.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {  // (3)
        Channel incoming = ctx.channel();
        channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception { // (4)
        Channel incoming = ctx.channel();
        for (Channel channel : channels) {
            if (channel != incoming){
                channel.writeAndFlush("[" + incoming.remoteAddress() + "]" + s + "\n");
            } else {
                channel.writeAndFlush("[you]" + s + "\n");
            }
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
        Channel incoming = ctx.channel();
        System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"在线");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
        Channel incoming = ctx.channel();
        System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"掉线");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (7)
        Channel incoming = ctx.channel();
        System.out.println("SimpleChatClient:"+incoming.remoteAddress()+"异常");
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

编写 ChannelInitializer 用于配置客户端连接的 Channel，这里的几个 Handler 作用如下：
- DelimiterBasedFrameDecoder：允许你指定的一个或多个分隔符来分割接收的 ByteBuf，它的构造方法参数意义如下：
  - maxFrameLength：解码的帧的最大长度
  - stripDelimiter：解码时是否去掉分隔符
  - failFast：为 true 时当 frame 长度超过 maxFrameLength 立即报 TooLongFrameException 异常，为 false 读取完整个帧再报异常
  - delimiter：分隔符
- StringDecoder：将入站的 ByteBuf 解码为字符串
- StringEncoder：将出站的 ByteBuf 编码为字符串

public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()))
                .addLast("decoder", new StringDecoder())
                .addLast("encoder", new StringEncoder())
                .addLast("handler", new SimpleChatServerHandler());

        System.out.println("SimpleChatClient:"+ch.remoteAddress() +"连接上");
    }
}

编写服务器

public class SimpleChatServer {

    private int port;

    public SimpleChatServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup();             // 处理客户端连接事件的线程池
        EventLoopGroup workerGroup = new NioEventLoopGroup();           // 处理连接后所有事件的线程池
        try {
            ServerBootstrap b = new ServerBootstrap();                  // NIO 服务的辅助启动类
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)              // 指定连接该服务器的 Channel 类型为 NioServerSocketChannel
                    .childHandler(new SimpleChatServerInitializer())      // 指定需要执行的 Handler
                    .option(ChannelOption.SO_BACKLOG, 128)          // 设置 bossGroup 的相关参数
                    .childOption(ChannelOption.SO_KEEPALIVE, true);         // 设置 workerGroup 相关参数
            System.out.println("SimpleChatServer 启动了");
            ChannelFuture f = b.bind(port).sync();      // 绑定端口，调用 ChannelFuture 的 sync() 阻塞方法等待绑定完成
            // 调用 closeFuture() 方法返回此通道关闭时的 ChannelFuture
            // 调用 ChannelFuture 的 sync() 阻塞方法直到服务端关闭链路之后才退出 main() 函数
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出机制。。。退出线程池（该方法源码没读过，也不知怎么个优雅方式）
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();

            System.out.println("SimpleChatServer 关闭了");
        }
    }

    public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        new SimpleChatServer(port).run();

    }
}

客户端

编写 Handler 类，处理消息

public class SimpleChatClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg);
    }
}

编写 ChannelInitializer 用于配置客户端连接的 Channel

public class SimpleChatClientInitializer extends ChannelInitializer {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast("framer",new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()))
                .addLast("decoder", new StringDecoder())
                .addLast("encoder", new StringEncoder())
                .addLast("handler",new SimpleChatClientHandler());
    }
}

编写客户端

public class SimpleChatClient {

    private final String host;
    private final int port;

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

    public void run(){
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new SimpleChatClientInitializer());
            Channel channel = bootstrap.connect(host,port).sync().channel();        // 连接服务端
            // 接收控制台输入的内容并发送给服务端
            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
            while (true){
                channel.writeAndFlush(br.readLine()+"\r\n");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        new SimpleChatClient("localhost",8080).run();
    }
}

Netty4（七）：实现聊天室

服务端

客户端

猜你喜欢

热点阅读