Netty实现群聊Demo

服务端：

public class GroupChatServer {

    private int port;

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

    public void run() throws Exception{
        EventLoopGroup boosGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(8);
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(workerGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    //.childOption(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast("decode", new StringDecoder());
                            ch.pipeline().addLast("encode", new StringEncoder());
                            ch.pipeline().addLast("handler", new GroupChatServerHandler());
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            channelFuture.channel().closeFuture().sync();
        }finally {
            boosGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new GroupChatServer(8888).run();
    }

}

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    // 连接建立调用
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        SocketAddress remoteAddr = ctx.channel().remoteAddress();
        channels.writeAndFlush("新增连接" + remoteAddr);
        channels.add(ctx.channel());
    }

    //channel处于活动状态
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

    }

    //channel处于不活动状态
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

    }

    //断开连接
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        SocketAddress remoteAddr = ctx.channel().remoteAddress();
        //channels.remove(ctx.channel()); 自动从ChannelGroup删除
        channels.writeAndFlush("断开连接" + remoteAddr);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        channels.forEach(channel -> {
            if (channel != ctx.channel()) {
                channel.writeAndFlush(ctx.channel().remoteAddress() + "发送了:" + msg);
            }
        });
    }

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

客户端：

public class GroupChatClient {

    private String host;
    private int port;

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

    public void run() throws Exception {
        EventLoopGroup executors = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(executors)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast("decode", new StringDecoder());
                            ch.pipeline().addLast("encode", new StringEncoder());
                            ch.pipeline().addLast("handler", new GroupChatClientHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            Scanner sc = new Scanner(System.in);
            while (sc.hasNextLine()) {
                String message = sc.nextLine();
                channelFuture.channel().writeAndFlush(message + "\r\n");
            }
        } finally {
            executors.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new GroupChatClient("127.0.0.1", 8888).run();;
    }

}

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {

    }
}