前面三个章节,我们使用了 Netty 实现了 DISCARD 丢弃服务和回复以及自定义编码解码,这篇博客,我们要用 Netty 实现简单的聊天室功能。
Ps: 突然想起来大学里面有个课程实训,给予 UDP 还是 TCP 实现的聊天程序,简单的分析一下,那个实现和基于 Netty 的实现是不一样的,基于 UDP 或者 TCP 做的聊天室中只能是客户端向服务发送消息(当然基于 UDP 的也可以建立两个 Channel 来实现服务器和客户端的双向通道),然后客户端接收到消息,这里的服务器仅仅作为一个接收消息处理之的作用,并不能主动向客户端推送消息。
基于前面几个章节的知识,这里我们做一个简单的聊天室功能,我们简单的说一下需求:
- 进入聊天室,服务器发送欢迎信息到该用户的客户端
- 有新人进入或者退出聊天室,那么聊天室的其他用户都能接收到通知信息
- 某位用户发送消息,其他用户的客户端显示格式为
[时间][发送用户的名称/地址]消息内容
,自己的客户端显示[时间][You]消息内容
这篇博文仅仅实现服务端的代码,使用 telent 测试,客户端的作为下一篇阐述。
工具类代码
工具类就一个时间格式化的工具,如下:
package com.zhoutao123.simpleChat.utils;
import java.text.SimpleDateFormat;
import java.util.Date;
public class DatetimeUtils {
private static final SimpleDateFormat smf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public static String getNowDatetime(){
return smf.format(new Date());
}
}
服务端代码
服务处理适配器
和之前的代码一样,这里我们继承 SimpleChannelInboundHandler,具体的解释在注释里面。
package com.zhoutao123.simpleChat.server;
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 static com.zhoutao123.simpleChat.utils.DatetimeUtils.getNowDatetime;
public class ServerHandle extends SimpleChannelInboundHandler<String> {
// 创建ChannelGroup 用于保存连接的Channel
public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
//当有新的Channel增加的时候
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
// 获取当前的Channel
Channel channel = ctx.channel();
//向其他Channel发送上线消息
channelGroup.writeAndFlush(String.format("[%s][服务器]\t用户:%s 加入聊天室!\n", getNowDatetime(), channel.remoteAddress()));
// 添加Channel到Group里面
channelGroup.add(channel);
// 向新用户发送欢迎信息
channel.writeAndFlush(String.format("你好,%s欢迎来到Netty聊天室\n", channel.remoteAddress()));
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
// 用户退出后向全部Channel发送下线消息
channelGroup.writeAndFlush(String.format("[%s][服务器]\t用户:%s 离开聊天室!\n", getNowDatetime(), ctx.channel().remoteAddress()));
// 移除
channelGroup.remove(ctx.channel());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
// 服务器接收到新的消息后之后执行
// 获取当前的Channel
Channel currentChannel = ctx.channel();
// 遍历
for (Channel channel : channelGroup) {
String sendMessage = "";
// 如果是当前的用户发送You的信息,不是则发送带有发送人的信息
if (channel == currentChannel) {
sendMessage = String.format("[%s][You]\t%s\n", getNowDatetime(), msg);
} else {
sendMessage = String.format("[%s][%s]\t %s\n", getNowDatetime(), currentChannel.remoteAddress(), msg);
}
channel.writeAndFlush(sendMessage);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 发送异常的时候通知移除
Channel channel = ctx.channel();
channelGroup.writeAndFlush(String.format("[%s][服务器]\t 用户 %s 出现异常掉线!\n", getNowDatetime(), channel.remoteAddress()));
ctx.close();
}
}
处理器初始化
这里主要是配置一些编码器以及解码器以及我们自己定义的 ServerHandleAdapter
package com.zhoutao123.simpleChat.server;
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;
public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast("handler", new ServerHandle());
System.out.println("SimpleChatClient:"+ch.remoteAddress() +"连接上");
}
}
启动服务器
启动的代码和以前一致,没有打的改动.
package com.zhoutao123.simpleChat.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class Server {
private final static int port = 8080;
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup work = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(boss, work)
.channel(NioServerSocketChannel.class)
.childHandler(new SimpleChatServerInitializer())
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
System.out.println("聊天服务已经启动.....");
ChannelFuture sync = serverBootstrap.bind(port).sync();
sync.channel().closeFuture().sync();
} finally {
work.shutdownGracefully();
boss.shutdownGracefully();
System.out.println("聊天服务已经被关闭");
}
}
}
telnet 测试
启动服务之后,我在 Linux 上使用 Telnet 命令来简单的测试了下,
这里创建了 4 个用户,发送了一些信息,可以观察一下:
欢迎来到这里!
我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。
注册 关于