我们使用的框架几乎都有网络通信的模块,比如常见的Dubbo、RocketMQ、ElasticSearch等。它们的网络通信模块使用Netty实现,之所以选择Netty,有两个主要原因:
- Netty封装了复杂的JDK 的 NIO操作,还封装了各种复杂的异常场景,丰富的API使得在使用上也非常方便,几行代码就可以实现高性能的网络通信功能。
- Netty已经经历各种大型中间件的生产环境的验证,高可用性和健壮性都得到了全方位验证,用起来更放心。
本文以入门实践为主,通过原理+代码的方式,实现一个简易IM聊天功能。分为两个部分:Netty的核心概念、IM聊天简易实现。
一、Netty核心概念
1、通信流程
既然是网络通信,那肯定有服务端和客户端。在客户端-A和客户端-B通信的过程中,实际上是利用服务端作为消息中转站,来实现A-B通信的。
不管是点-点通信,还是群通信,都可以认为是客户端-服务端之间的通信,有了这一点,许多设计方案都可以轻松理解。
2、服务端核心概念
(1) Boss线程:Boss线程负责监听端口,接受新的连接,监听连接的数据读写变化。
(2) Worker线程:Worker线程负责处理具体的业务逻辑,Boss线程接收到连接的读写变化后,然后交给Worker处理具体业务逻辑。
(3) 服务端的IO模型:Netty支持使用NIO和BIO进行通信,可以自行设置。一般使用NioServerSocketChannel来指定NIO模型。
(4) 服务端引导类:服务端通过引导类 ServerBootstrap来启动一系列的工作。
3、客户端核心概念
(1) Worker线程:客户端只有工作线程的概念,负责连接到服务端,监听数据读写变化。
(2) 客户端的IO模型:一般使用NioSocketChannel指定客户端的NIO模型
(3) 客户端引导类:客户端通过引导类Bootstrap来启动一些列工作。
4、通用核心概念
(1) Handler:负责处理接受到的消息,大部分的业务逻辑都是放在Handler里处理。自定义的Handler一般继承于SimpleChannelInboundHandler或者ChannelInboundHandlerAdapter。
(2) ByteBuf和编码、解码:数据的载体,Java对象编码成字节码,存放于ByteBuf,然后发送出去。服务端接收到消息后,从ByteBuf中取出数据,解码成Java对象。
(3) 通讯协议:许多框架都会自定义一套自己的协议,这样比较符合业务。比如dubbo协议、hessian协议。
一般的协议包括如下部分:魔数、版本号、序列化算法、指令、数据长度、数据内容,其余的都是为了适配自身业务而定的。
- 魔数:一般是固定数字,用来快速判断是否符合本协议,如果不符合本协议,则快速失败。
- 版本号:一般无需改动,如果早期设置的协议到了后续不适用了,在升级版本号。
- 序列化算法:Java对象转序列化的方式,比如JSON。
- 指令:操作大类。比如说登录指令、单点发送消息指令、建群指令等。这样服务端接收到对应指令就用对应的Handler去处理业务逻辑。指令占用的字节数可以根据自身业务适当调大。
- 数据长度:用来记录本次数据的长度。
- 数据内容:具体消息内容,比如聊天时的消息、登录时的用户名密码等。
(4) 粘包拆包
Netty属于上层应用,在发送消息时,还是通过底层操作系统将数据发送出去,操作系统在发送数据时,不会按照我们设想的消息长度去发送内容。这就需要我们在接收到内容时,自行做好内容的分割和等待。
比如有一条消息1024字节,如果接受的内容没这么长就需要继续等待,等这条消息的内容完整后,在处理。如果接受的内容包含了1条完整消息和1条不完整的消息,那么就需要拆分内容,将完整的消息先传递到后面处理,剩下不完整的消息则继续等待下一个内容。
Netty自带了几种拆包器:固定长度的拆包器 FixedLengthFrameDecoder、行拆包器 LineBasedFrameDecoder、分隔符拆包器 DelimiterBasedFrameDecoder、长度域拆包器LengthFieldBasedFrameDecoder。
一般在使用自定义协议时,会使用:长度域拆包器 LengthFieldBasedFrameDecoder。
(5) 空闲检测和定时心跳
在服务端和客户端的通信过程中,有时候会出现假死连接,或者长时间没有消息传递需要释放连接。对于这些连接,我们需要及时释放,毕竟每条连接都占用着CPU和内存资源。大量这种连接如果不及时释放,服务器资源迟早会耗尽,最终崩溃。
应对这种问题的解决方式是:Netty提供了IdleStateHandler做空闲检测,用来检测连接是否活跃,如果再指定的时间内,没有活跃,那么就关闭连接。然后就是客户端定时发送心跳请求,服务器响应心跳请求。
二、IM聊天简易实现
介绍完Netty的核心概念,接下来以一个简易的点对点IM聊天,将核心概念融入到案例中。IM聊天的核心模块大致是如下几个:
1、通信主体流程
通信主体流程就是搭建好:服务端、客户端、两端正常建立连接进行通信。
服务端代码:
public static void main(String[] args) {
ServerBootstrap serverBootstrap = new ServerBootstrap();
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
serverBootstrap
.group(boss, worker)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer() {
protected void initChannel(NioSocketChannel ch) {
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new SimpleChannelInboundHandler() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println("server accept: " + msg);
}
});
}
});
serverBootstrap.bind(9000)
.addListener(future -> {
if (future.isSuccess()) {
System.out.println("端口9000绑定成功");
} else {
System.err.println("端口9000绑定失败");
}
});
}
客户端代码:
public static void main(String[] args) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
NioEventLoopGroup group = new NioEventLoopGroup();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer() {
@Override
protected void initChannel(Channel ch) {
ch.pipeline().addLast(new StringEncoder());
}
});
bootstrap.connect("127.0.0.1", 9000)
.addListener(future -> {
if (future.isSuccess()) {
System.out.println("链接服务端成功");
Channel channel = ((ChannelFuture) future).channel();
channel.writeAndFlush("我是客户端A");
} else {
System.err.println("连接服务端失败");
}
});
}
2、数据包—包含通讯协议
定义数据包的抽象类,后续的各种类型的数据包都继承此类。数据包中定义通讯协议的各种字段。
@Data
public abstract class Packet {
/**
* 协议版本
*/
private Byte version = 1;
/**
* 指令,此处有多种实现:比如登录、登出、单聊、建群等等
*
* @return
*/
public abstract Byte getCommand();
/**
* 获取算法,默认使用JSON,如果使用其余算法,子类重写此方法
*
* @return
*/
public Byte getSerializeAlgorithm() {
return SerializerAlgorithm.JSON;
}
}
public class LoginRequestPacket extends Packet {
private String userName;
private String password;
@Override
public Byte getCommand() {
return Command.LOGIN_REQUEST;
}
}
3、序列化器
定义序列化器,功能包括:序列化、反序列化。可以定义多种序列化算法,文中以JSON为例。
public interface Serializer {
/**
* 序列化算法
*
* @return
*/
byte getSerializerAlgorithm();
/**
* java 对象转换成二进制
*/
byte[] serialize(Object object);
/**
* 二进制转换成 java 对象
*/
T deserialize(Class clazz, byte[] bytes);
}
public class JSONSerializer implements Serializer {
@Override
public byte getSerializerAlgorithm() {
return SerializerAlgorithm.JSON;
}
@Override
public byte[] serialize(Object object) {
return JSON.toJSONBytes(object);
}
@Override
public T deserialize(Class clazz, byte[] bytes) {
return JSON.parseObject(bytes, clazz);
}
}
4、编解码器
有了通讯协议、有了序列化协议,接下来就是对数据的编码和解码了。
public void encode(ByteBuf byteBuf, Packet packet) {
Serializer serializer = getSerializer(packet.getSerializeAlgorithm());
// 1. 序列化 java 对象
byte[] bytes = serializer.serialize(packet);
// 2. 实际编码过程
byteBuf.writeInt(MAGIC_NUMBER);
byteBuf.writeByte(packet.getVersion());
byteBuf.writeByte(packet.getSerializeAlgorithm());
byteBuf.writeByte(packet.getCommand());
byteBuf.writeInt(bytes.length);
byteBuf.writeBytes(bytes);
}
public Packet decode(ByteBuf byteBuf) {
// 跳过 magic number
byteBuf.skipBytes(4);
// 跳过版本号
byteBuf.skipBytes(1);
// 读取序列化算法
byte serializeAlgorithm = byteBuf.readByte();
// 读取指令
byte command = byteBuf.readByte();
// 读取数据包长度
int length = byteBuf.readInt();
// 读取数据
byte[] bytes = new byte[length];
byteBuf.readBytes(bytes);
Class