使用网状实现多客户端连接并且互相通信

叁歲伎倆 2022-03-25 08:38 172阅读 0赞

[https://www.jianshu.com/p/adc2de3691c7][https_www.jianshu.com_p_adc2de3691c7]

使用网状实现多客户端连接并且互相通信的需求：

> 1.服务器启动，正多个客户端与服务器进行连接，一个客户端上线之后，服务器端控制台会打印XX上线了，其他的客户端控制台打印XX上线了。如果一个客户端下线了，服务器端的控制台上打印，XX下线了，其他的客户端控制台打印XX下线了。
> 
> 2.多个客户端都上线之后，一个客户端（比如说A）给服务端发送消息，那么客户端（比如说A，B，C，包括自己本身）都会收到消息，对于甲来说，会标志此消息是自己发送自己的，其他的客户端则会收到具体的消息。

服务器代码：

import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; public class MyChatServer { public static void main(String[] args) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup wokerGroup = new NioEventLoopGroup(); try{ ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,wokerGroup).channel(NioServerSocketChannel.class) .childHandler(new MyChatServerInializer()); ChannelFuture channelFuture = serverBootstrap.bind(8899).sync(); channelFuture.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); wokerGroup.shutdownGracefully(); } } }

服务器端初始化初始化器

import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.Delimiters; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.util.CharsetUtil; public class MyChatServerInializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //分割接收到的Bytebu，根据指定的分割符 pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter())); pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8)); pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8)); pipeline.addLast(new MyChatServerHandler()); } }

自定义处理程序：

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; public class MyChatServerHandler extends SimpleChannelInboundHandler<String>{ //保留所有与服务器建立连接的channel对象，这边的GlobalEventExecutor在写博客的时候解释一下，看其doc private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); /** * 服务器端收到任何一个客户端的消息都会触发这个方法 * 连接的客户端向服务器端发送消息，那么其他客户端都收到此消息，自己收到【自己】+消息 */ @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { Channel channel = ctx.channel(); channelGroup.forEach(ch -> { if(channel !=ch){ ch.writeAndFlush(channel.remoteAddress() +" 发送的消息:" +msg+" \n"); }else{ ch.writeAndFlush(" 【自己】"+msg +" \n"); } }); } //表示服务端与客户端连接建立 @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //其实相当于一个connection /** * 调用channelGroup的writeAndFlush其实就相当于channelGroup中的每个channel都writeAndFlush * * 先去广播，再将自己加入到channelGroup中 */ 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"); //验证一下每次客户端断开连接，连接自动地从channelGroup中删除调。 System.out.println(channelGroup.size()); //当客户端和服务端断开连接的时候，下面的那段代码netty会自动调用，所以不需要人为的去调用它 //channelGroup.remove(channel); } //连接处于活动状态 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); System.out.println(channel.remoteAddress() +" 上线了"); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); System.out.println(channel.remoteAddress() +" 下线了"); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }

客户端代码：

import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import java.io.BufferedReader; import java.io.InputStreamReader; public class MyChatClient { public static void main(String[] args) throws Exception{ EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); try{ Bootstrap bootstrap = new Bootstrap(); bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class) .handler(new MyChatClientInitializer()); Channel channel = bootstrap.connect("localhost",8899).sync().channel(); //标准输入 BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in)); //利用死循环，不断读取客户端在控制台上的输入内容 for (;;){ channel.writeAndFlush(bufferedReader.readLine() +"\r\n"); } }finally { eventLoopGroup.shutdownGracefully(); } } }

客户端初始化初始化器

import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.Delimiters; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.util.CharsetUtil; public class MyChatClientInitializer extends ChannelInitializer<SocketChannel>{ @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter())); pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8)); pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8)); pipeline.addLast(new MyChatClientHandler()); } }

客户端处理程序：

import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; public class MyChatClientHandler extends SimpleChannelInboundHandler<String> { @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }

启动服务器，启动第一个客户端的时候，服务器端控制台打印

/127.0.0.1:54771 上线了

第二个客户端启动，则服务器控制台打印，

/127.0.0.1:54771 上线了
    /127.0.0.1:54792 上线了

此时第一个客户端打印，  
【服务器】 - / 127.0.0.1:54792加入

启动第三个客户端依次可以自己验证，当我在客户端1控制台输入信息的时候，客户端1控制台打印了`【自己】hi，server`

hi，server
     【自己】hi，server

其他的二个客户端控制台打印

/127.0.0.1:54771 发送的消息:hi，server

如果其他的客户端下线之后，比如客户端1下线，服务器端控制台打印

/127.0.0.1:54771 下线了
    2

其他客户端控制台打印：