使用Netty编写一个简单的群聊系统

87 阅读 0 评论 58 点赞

我是靠谱客的博主震动金鱼，最近开发中收集的这篇文章主要介绍使用Netty编写一个简单的群聊系统，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

概述
具体业务跟下面文章的一样，只是下面文章用的是NIO，这里使用Netty改造。
https://blog.csdn.net/qq_40837310/article/details/111248486

目的是为了熟悉Netty基本API使用。

服务端编码：

public class ServerGroupChatServer {

private int port;

public ServerGroupChatServer(int port){
this.port = port;
}

public void start() throws InterruptedException {

NioEventLoopGroup bossGroup = new NioEventLoopGroup();

NioEventLoopGroup workerGroup = new NioEventLoopGroup();

try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
//设置TCP等待队列为128
.option(ChannelOption.SO_BACKLOG,128)
//长连接
.childOption(ChannelOption.SO_KEEPALIVE,true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//添加handler
socketChannel.pipeline().addLast(new MyNettyChannelHandler());
}
});

ChannelFuture bindFuture = serverBootstrap.bind("127.0.0.1", port);
//添加一个监听器，监听绑定情况
bindFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
boolean success = future.isSuccess();
if (success)
System.out.println("服务绑定成功");
else{
System.out.println("服务绑定失败");
}
}
});

bindFuture.channel().closeFuture().sync();

}finally {
//关闭
bossGroup.shutdownGracefully();
//关闭
workerGroup.shutdownGracefully();
}

}

public static void main(String[] args) throws InterruptedException {
ServerGroupChatServer serverGroupChatServer = new ServerGroupChatServer(9898);
serverGroupChatServer.start();
}
}

服务端的Handler：

public class MyNettyChannelHandler extends ChannelInboundHandlerAdapter {

private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//进入这里表示channel已经处于非激活状态，关闭，在这里提示该客户端下线。
System.out.println("客户端：" + ctx.channel().remoteAddress() + "已离线");
channels.remove(ctx.channel());
super.channelInactive(ctx);
}

//READ事件触发时调用
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//按照原业务，读取到消息就进行转发
for (Channel channel:channels){
//排除自身
if (channel == ctx.channel())
continue;
ByteBuf byteBuf = (ByteBuf) msg;
byteBuf.retain();
//发送
channel.writeAndFlush(byteBuf);
}

ctx.fireChannelRead(msg);
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//进入这里就证明连接以建立，提醒其他用户此客户端已上线
//ChannelGroup的写是里面所有channel 的写
channels.writeAndFlush(Unpooled.copiedBuffer("客户端：" + ctx.channel().remoteAddress() + " 已上线。", CharsetUtil.UTF_8));
//把当前通道加到集合中
channels.add(ctx.channel());
super.channelActive(ctx);
}

//触发异常关闭客户端
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.fireChannelActive();
}
}

客户端代码：

public class NettyGroupChatClient {

private int serverPort;

private String clientName;

private SocketChannel socketChannel;

public NettyGroupChatClient(int serverPort,String clientName){
this.clientName = clientName;
this.serverPort = serverPort;
}

public void start() throws InterruptedException {
NioEventLoopGroup worker = new NioEventLoopGroup();

Bootstrap bootstrap = new Bootstrap();

try {
bootstrap.group(worker)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new MyNettyChannelHandlerClient());
//sc.pipeline().addLast(new OUT1());
//sc.pipeline().addLast(new OUT2());
socketChannel = sc;
}
});

ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", serverPort).sync();
//添加一个监听器，当连接成功时就给服务端每三秒发送一次数据
channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()){
System.out.println("连接成功");
new Thread(()->{
while (true) {
socketChannel.writeAndFlush(Unpooled.copiedBuffer("Hello, I am " + clientName, CharsetUtil.UTF_8));
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
else
System.out.println("连接失败");
}

});
channelFuture.channel().closeFuture().sync();
}finally {
worker.shutdownGracefully();
}

}

public static void main(String[] args) throws InterruptedException {
NettyGroupChatClient Mike = new NettyGroupChatClient(9898,"John");
Mike.start();

}

}

客户端的Handler

public class MyNettyChannelHandlerClient extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf) msg;
System.out.println(byteBuf.toString(CharsetUtil.UTF_8));
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
}
}

运行结果：
客户端1：

客户端2：

客户端3：

都能过接收其他两个客户端发的消息。

把三个客户端关闭后：
服务端控制台：