群聊案例
1.案例需求
- 编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
- 实现多人群聊
- 服务器端：可以监测用户上线，离线，并实现消息转发功能
- 客户端：通过channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到)
目的：进一步理解Netty非阻塞网络编程机制
2.服务端代码
2.1 服务端处理器
在服务端处理器中我们要处理客户端的上下线及消息的分发
package com.dpb.netty.goupchat; 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; import java.text.SimpleDateFormat; import java.util.Date; /** * @program: netty4demo * @description: 群聊服务处理器 * 1.客户端连接 提示其他客户端上线 * 2.客户端发送消息，需要将详细转发给其他客户端 * @author: 波波烤鸭 * @create: 2019-12-29 14:02 */ public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> { // 管理所有客户端的Channel 是一个单例对象 private static final ChannelGroup grouup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss"); /** * 客户端连接触发的方法 * @param ctx * @throws Exception */ @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { // 获取当前连接对象的 Channel对象 Channel channel = ctx.channel(); // 通知其他客户端 当前客户端连接上线了 grouup.writeAndFlush("[客户端上线了]" + channel.remoteAddress() + " 上线了 " + sdf.format(new Date()) + "\n"); // 将当前channel添加到 channelGroup中 grouup.add(channel); } /** * 客户端断开连接触发的方法 * @param ctx * @throws Exception */ @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { // 通知其他客户端 我离线了 ctx.writeAndFlush("[客户端下线了]" + ctx.channel().remoteAddress() + " : " + sdf.format(new Date())+ "\n"); } /** * 读取客户端请求消息的方法 * @param context * @param s * @throws Exception */ @Override protected void channelRead0(ChannelHandlerContext context, String s) throws Exception { // 获取当前的 handler Channel channel = context.channel(); // 遍历channelGroup grouup.forEach(ch ->{ if(ch == channel){ // 是自己 ch.writeAndFlush( "我说:" + s); }else{ // 不是自己 ch.writeAndFlush(channel.remoteAddress() + "说:" + s); } }); } }
2.2 服务端
服务端我们要启动我们的服务，绑定端口等。
package com.dpb.netty.goupchat; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; /** * @program: netty4demo * @description: 群聊系统服务端 * @author: 波波烤鸭 * @create: 2019-12-29 13:50 */ public class GroupChatServer { // 服务的端口号 private int port; public GroupChatServer(int port){ this.port = port; } /** * 服务运行的方法 */ public void run() throws Exception{ // 创建BossGroup和WorkGroup EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workGroup = new NioEventLoopGroup(); // 创建 ServerBootstrap 服务启动对象 ServerBootstrap bootstrap = new ServerBootstrap(); try{ // 给bootstrap设置相关的参数 bootstrap.group(bossGroup,workGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG,128) .childOption(ChannelOption.SO_KEEPALIVE,true) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { // 获取pipeline对象 ChannelPipeline pipeline = sc.pipeline(); // 设置对应的handler pipeline.addLast("docoder",new StringDecoder()); pipeline.addLast("encoder",new StringEncoder()); pipeline.addLast(new GroupChatServerHandler()); } }); System.out.println("服务端启动了......."); // 绑定端口 ChannelFuture future = bootstrap.bind(port).sync(); future.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); workGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception{ GroupChatServer server = new GroupChatServer(8888); server.run(); } }
3. 客户端代码
3.1 客户端处理器
获取服务器转发的消息
package com.dpb.netty.goupchat; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; /** * @program: netty4demo * @description: 客户端处理器 * @author: 波波烤鸭 * @create: 2019-12-29 14:19 */ public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> { /** * 读取消息 * @param channelHandlerContext * @param s * @throws Exception */ @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception { System.out.println(s.trim()); } }
3.2 客户端代码
连接服务器，发送消息
package com.dpb.netty.goupchat; import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; 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.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import java.util.Scanner; /** * @program: netty4demo * @description: * @author: 波波烤鸭 * @create: 2019-12-29 14:14 */ public class GroupChatClient { public String host; public int port ; public GroupChatClient(String host,int port){ this.host = host; this.port = port; } public void run() throws Exception{ EventLoopGroup clientGroup = new NioEventLoopGroup(1); Bootstrap bootstrap = new Bootstrap(); try{ bootstrap.group(clientGroup) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast("decoder",new StringDecoder()); sc.pipeline().addLast("encoder",new StringEncoder()); sc.pipeline().addLast(new GroupChatClientHandler()); } }); ChannelFuture future = bootstrap.connect(host, port).sync(); Channel channel = future.channel(); // 发送消息 Scanner scanner = new Scanner(System.in); while(scanner.hasNextLine()){ String msg = scanner.nextLine(); channel.writeAndFlush(msg + "\n"); } }finally { clientGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { GroupChatClient client = new GroupChatClient("localhost",8888); client.run(); } }
4.测试
启动服务器，然后启动多个客户端查看效果
查看效果，功能实现~
