Netty聊天室

项目结构

服务端

Server

public class GroupChatServer { private int port; //监听端口 public GroupChatServer(int port) { this.port = port; } //编写run方法，处理客户端请求 public void run() throws InterruptedException { //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG,128) .childOption(ChannelOption.SO_KEEPALIVE,true) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //获取pipeline ChannelPipeline pipeline = ch.pipeline(); //pipeline加入解码器 pipeline.addLast("decoder",new StringDecoder()); //pipeline加入编码器 pipeline.addLast("encoder",new StringEncoder()); //加入组件的业务处理的handler pipeline.addLast(new GroupChatServerHandler()); } }); System.out.println("netty 服务器启动"); ChannelFuture channelFuture = serverBootstrap.bind(port).sync(); //监听关闭 channelFuture.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws InterruptedException { new GroupChatServer(6666).run(); } }

ServerHandler

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> { //定义一个channle组，管理所有的channel //GlobalEventExecutor.INSTANCE 是全局事件执行器，是一个单例 private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); //一但建立连接，该方法就会被执行 @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //将该客户加入聊天的信息推送给其它在线的客户端 //该方法会把channelGroup的所有channel遍历，调用writeAndFlush方法 channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"加入聊天\n"); channelGroup.add(channel); } //一但断开连接，该方法就会被执行 @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { //提示其它所有用户当前用户已经离线 Channel channel = ctx.channel(); channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"离开了\n"); System.out.println("channelGroup size"+channelGroup.size()); } //channel 处于活动状态执行该方法 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress()+"上线了~"); } //channel 处于非活动状态执行该方法 @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress()+"离线了~"); } //读取数据时触发 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { Channel channel = ctx.channel(); //遍历channelGroup根据不同情况，回送不同消息 channelGroup.forEach(ch -> { if (channel != ch){ //不是读取的channel，转发 ch.writeAndFlush("[用户]"+channel.remoteAddress() +" 发送了消息: "+msg+"\n"); }else { ch.writeAndFlush("[自己]发送了消息: "+msg+"\n"); } }); } //发生异常触发 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭通道 ctx.close(); } }

客户端

Client

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 InterruptedException { NioEventLoopGroup 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 { //得到pipeline ChannelPipeline pipeline = ch.pipeline(); //pipeline加入解码器 pipeline.addLast("decoder",new StringDecoder()); //pipeline加入编码器 pipeline.addLast("encoder",new StringEncoder()); //加入自定义handler pipeline.addLast(new GroupChatClientHandler()); } }); ChannelFuture channelFuture = bootstrap.connect(host, port).sync(); Channel channel = channelFuture.channel(); System.out.println("----------"+channel.localAddress()+"----------"); //客户端需要输入消息 Scanner scanner = new Scanner(System.in); while (scanner.hasNextLine()){ String msg = scanner.nextLine(); //通过channel 发生到服务器端 channel.writeAndFlush(msg+"\r\n"); } }finally { group.shutdownGracefully(); } } public static void main(String[] args) throws InterruptedException { new GroupChatClient("localhost",6666).run(); } }

ClientHandler

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