Netty群聊系统（Netty十一）

服务器端：

public class GroupChatServer {private int port;public GroupChatServer(int port) {this.port = port;}public void run() throws Exception {//创建两个线程组NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);NioEventLoopGroup workGroup = new NioEventLoopGroup();ServerBootstrap bootstrap = new ServerBootstrap();try {bootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();// 向pipline加入解码器pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast(new GroupChatServerHandler());}});System.out.println("netty服务器启动");ChannelFuture channelFuture = bootstrap.bind(port).sync();// 监听关闭channelFuture.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new GroupChatServer(7000).run();}
}

服务器端handler：

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {// 定义一个channel组，管理所有的channelprivate static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 当连接建立，一旦连接第一个被执行// 将当前channel加入到channelGroup@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();// 将客户加入聊天的信息推送给其他在线的客户端/*** 该方法会将channelgroup中所有channel遍历，并发送该消息,因此不需要自己遍历*/channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天\n");channelGroup.add(channel);}// 断开连接会被触发,将xx客户离开信息推送给其他当前在线的客户@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了\n");}// 表示channel处于活动状态，提示xx上线@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(ctx.channel().remoteAddress() + " 上线了 ~ ");}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println(ctx.channel().remoteAddress() + " 离线了 ");}@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {Channel channel = channelHandlerContext.channel();// 遍历channelGroup,根据不同的情况回送不同的消息channelGroup.forEach(ch -> {if (channel != ch) {// 不是当前channel,直接转发消息ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送消息" + s + "\n");} else {ch.writeAndFlush("[自己]发送了消息 " + s + "\n");}});}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

客户端：

public class GroupChatClient {private final String host;private final int port;public GroupChatClient(String host, int port) {this.host = host;this.port = port;}public void run() throws Exception {NioEventLoopGroup eventExecutors = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();try {bootstrap.group(eventExecutors).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast(new GroupChatClientHandler());}});ChannelFuture channelFuture = bootstrap.connect(host, port).sync();//得到channelChannel channel = channelFuture.channel();System.out.println("-----" + channel.localAddress() + "----");//客户端需要输入信息Scanner scanner = new Scanner(System.in);while (scanner.hasNextLine()) {String msg = scanner.nextLine();channel.writeAndFlush(msg + "\r\n");}} finally {eventExecutors.shutdownGracefully();}}public static void main(String[] args) throws Exception {new GroupChatClient("127.0.0.1", 7000).run();}
}

客户端handler：

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {System.out.println(s.toString());}
}