03_netty实现聊天室功能

/** * 聊天室服务端 */public class ChatRoomServer { ???private final int port ; ???public ChatRoomServer(int port) { ???????this.port = port; ???} ???public void start(){ ???????EventLoopGroup boss = new NioEventLoopGroup(); ???????EventLoopGroup worker = new NioEventLoopGroup(); ???????try{ ???????????ServerBootstrap bootstrap = new ServerBootstrap(); ???????????bootstrap.group(boss,worker) ???????????????????.channel(NioServerSocketChannel.class) ???????????????????.childHandler(new ChatServerInitialize()) ???????????????????.option(ChannelOption.SO_BACKLOG, 128) ???????????????????.option(ChannelOption.SO_KEEPALIVE, true); ???????????ChannelFuture future = bootstrap.bind(port).sync(); ???????????future.channel().closeFuture().sync(); ???????}catch (Exception e){ ???????????e.printStackTrace(); ???????}finally { ???????????boss.shutdownGracefully(); ???????????worker.shutdownGracefully(); ???????} ???} ???public static void main(String[] args) { ???????new ChatRoomServer(9999).start(); ?//服务端监听本地的9999端口 ???}}

public class ChatServerInitialize extends ChannelInitializer<SocketChannel>{ ???@Override ???protected void initChannel(SocketChannel channel) throws Exception { ???????System.out.println("用户【"+channel.remoteAddress()+"】接入聊天室......"); ???????ChannelPipeline pipeline = channel.pipeline(); ???????pipeline.addLast("framer",new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter())); ???????pipeline.addLast("decoder",new StringDecoder()); ???????pipeline.addLast("encoder",new StringEncoder()); ???????pipeline.addLast("handler",new ChatServerHandler()); ???}}

/** * 聊天服务器对各种情况的处理 */public class ChatServerHandler extends SimpleChannelInboundHandler<String> { ???public static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); ???/** ????* 当从服务端收到新的客户端连接时 ????* 客户端的 Channel 存入 channels 列表中，并通知列表中的其他客户端 Channel ????*/ ???@Override ???public void handlerAdded(ChannelHandlerContext ctx) throws Exception { ???????Channel clientChannel = ctx.channel(); ???????channels.add(clientChannel); ???????for (Channel ch : channels) { ???????????if (ch != clientChannel) { ?//通知除了自己以外的其他用户 ???????????????ch.writeAndFlush("【提示】：用户【" + clientChannel.remoteAddress() + "】进入聊天室...\n"); ???????????} ???????} ???} ???/** ????* 每当从服务端收到客户端断开时 ????* 客户端的 Channel 自动从 channels 列表中移除了，并通知列表中的其他客户端 Channel ????*/ ???@Override ???public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { ???????Channel clientChannel = ctx.channel(); ???????channels.remove(clientChannel); ???????for (Channel ch : channels) { ???????????if (ch != clientChannel) { ?//通知除了自己以外的其他用户 ???????????????ch.writeAndFlush("【提示】：用户【" + clientChannel.remoteAddress() + "】退出聊天室...\n"); ???????????} ???????} ???} ???/** ????* 接受到客户端发出的消息 ????* 判断channel是否是 ????*/ ???@Override ???protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { ???????Channel clientChannel = ctx.channel(); ???????for (Channel ch : channels) { ???????????if (ch != clientChannel) { ???????????????ch.writeAndFlush("用户【" + clientChannel.remoteAddress() + "】说：" + msg + "\n"); ???????????} else { ???????????????ch.writeAndFlush("【我】说：" + msg + "\n"); ???????????} ???????} ???} ???/** ????* 服务端监听到客户端活动 ????*/ ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????Channel clientChannel = ctx.channel(); ???????System.out.println("用户【"+clientChannel.remoteAddress()+"】在线中..."); ???} ???/** ????* 服务端监听到客户端 不活动 ????*/ ???@Override ???public void channelInactive(ChannelHandlerContext ctx) throws Exception { ???????Channel clientChannel = ctx.channel(); ???????System.out.println("用户【 " +clientChannel.remoteAddress()+"】:离线了"); ???}}

/** * 聊天客户端 */public class ChatClient { ???private final String host; ???private final int port; ???public ChatClient(String host, int port) { ???????this.host = host; ???????this.port = port; ???} ???public void start() { ???????EventLoopGroup worker = new NioEventLoopGroup(); ???????Bootstrap bootstrap = new Bootstrap(); ???????try{ ???????????bootstrap.group(worker) ???????????????????.channel(NioSocketChannel.class) ???????????????????.handler(new ClientInitializer()); ???????????Channel channel ?= bootstrap.connect(host,port).sync().channel(); ???????????//客户端从键盘输入数据 ???????????BufferedReader input = new BufferedReader(new InputStreamReader(System.in)); ???????????while(true){ ???????????????channel.writeAndFlush(input.readLine()+"\n"); ???????????} ???????}catch (Exception e){ ???????????e.printStackTrace(); ???????}finally { ???????????worker.shutdownGracefully(); ???????} ???} ???public static void main(String[] args) { ???????new ChatClient("127.0.0.1",9999).start(); //连接服务器端 ???}}

public class ChatClientInitializer extends ChannelInitializer<SocketChannel>{ ???@Override ???protected void initChannel(SocketChannel socketChannel) throws Exception { ???????//当有客户端连接服务器时，netty会调用这个初始化器的 initChannel方法 ???????System.out.println("客户端开始初始化......"); ???????ChannelPipeline pipeline = socketChannel.pipeline(); ???????pipeline.addLast("framer",new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter())); ???????pipeline.addLast("decoder",new StringDecoder()); ???????pipeline.addLast("encoder",new StringEncoder()); ???????pipeline.addLast("handler",new ChatClientHandler()); ???}}

public class ChatClientHandler extends SimpleChannelInboundHandler<String> { ???/** ????* 打印服务端发送过来的数据 ????*/ ???@Override ???protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception { ???????System.out.println(s); ???}}