netty实现聊天室-蒲公英云

什么是IO多路复用：
I/O多路复用，I/O是指网络I/O, 多路指多个TCP连接(即socket或者channel），复用指复用一个或几个线程。
简单来说：就是使用一个或者几个线程处理多个TCP连接
最大优势是减少系统开销小，不必创建过多的进程/线程，也不必维护这些进程/线程

epoll：

1)没fd这个限制，所支持的FD上限是操作系统的最大文件句柄数，1G内存大概支持10万个句柄
2)效率提高，使用回调通知而不是轮询的方式，不会随着FD数目的增加效率下降

3)通过callback机制通知，内核和用户空间mmap同一块内存实现

依赖：

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

1、服务端

package com.example.demo.chat;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
public class ChatServer {
    private int port = 8080;
    public static void main(String[] args) throws Exception {
        new ChatServer().start();
    }
    private void start() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workGroup)//
                    .channel(NioServerSocketChannel.class)//
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()//
                                    .addLast("framer", new DelimiterBasedFrameDecoder(2048, Delimiters.lineDelimiter()))//
                                    .addLast("decoder", new StringDecoder(CharsetUtil.UTF_8))//
                                    .addLast("encoder", new StringEncoder(CharsetUtil.UTF_8))//
                                    .addLast("handler", new ChatServerHandler());
                        }
                    })//
                    .option(ChannelOption.SO_BACKLOG, 1024)//
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("=======================服务端启动了========================");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

2、服务端处理器

package com.example.demo.chat;
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;
public class ChatServerHandler extends SimpleChannelInboundHandler<String> {
    // 全局频道变量，存储所有客户端
    private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        for (Channel channel : channels) {
            channel.writeAndFlush(ctx.channel().remoteAddress() + "，进来啦\n");
        }
        channels.add(ctx.channel());
    }
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        channels.remove(ctx.channel());
        for (Channel channel : channels) {
            channel.writeAndFlush(ctx.channel().remoteAddress() + "，离开啦\n");
        }
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "，上线了\n");
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "，离线了\n");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        Channel ch = ctx.channel();
        for (Channel channel : channels) {
            if (ch == channel) {
                channel.writeAndFlush("我说：" + msg + "\n");
            } else {
                channel.writeAndFlush(channel.remoteAddress() + " 说：" + msg + "\n");
            }
        }
    }
}

3、客户端

package com.example.demo.chat;
import java.io.BufferedReader;
import java.io.InputStreamReader;
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;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
public class ChatClient {
    private int port = 8080;
    private String host = "127.0.0.1";
    public static void main(String[] args) throws Exception {
        new ChatClient().start();
    }
    private void start() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)//
                    .channel(NioSocketChannel.class)//
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()//
                                    .addLast("framer", new DelimiterBasedFrameDecoder(2048, Delimiters.lineDelimiter()))//
                                    .addLast("decoder", new StringDecoder(CharsetUtil.UTF_8))//
                                    .addLast("encoder", new StringEncoder(CharsetUtil.UTF_8))//
                                    .addLast("handler", new ChatClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect(host, port).sync();
            BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
            while (true) {
                future.channel().writeAndFlush(reader.readLine() + "\n");
            }
        } finally {
            group.shutdownGracefully();
        }
    }
}

4、客户端处理器

package com.example.demo.chat;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class ChatClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg);
    }
}