手写个即时通讯

976次阅读

共计 4291 个字符，预计需要花费 11 分钟才能阅读完成。

1.前言

要实现即时通讯，基本的网络知识储备必不可少，最基本的OSI七层网络模型要清楚,OSI定义

The Open Systems Interconnection (OSI) model describes seven layers that computer systems use to communicate over a network. It was the first standard model for network communications, adopted by all major computer and telecommunication companies in the early 1980s

20世纪80年代提出，也就是我们现在网络通信的基石

自下而上，分别为物理层，数据连接层，网络层，传输层，会话层，表示层，应用层。绝大多程序员处在应用层开发，我们要写个即时通讯，通常从第四层开发，基于TCP/UDP协议，通常还会包裹一层私有协议。

2.Java如何开发

虽然Java不像C++之流，但是即时通讯还是可以开发的，毕竟生态完善，有很多开源组件可用，不用重复造论子。比如Netty，一款优秀的IO库，笔者就Netty的Nio展开。

新建基于Maven构建的Springboot项目，添加如下依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
</dependency>

由于Boot已经引入了依赖定义，所以此处可以不指定版本号。数据传输使用google的protobuf数据格式，可能会有读者想问，为啥用这个？我们要知道数据网络传输是需要序列化的，这样数据接受方再反序列化得到发送的数据，这中间必然会有数据体积问题，安全问题等，而protobuf序列化体积小，而且本身是无法自描述的，就和摩斯码似的，也就有了一定的安全性，同时支持跨语言通信。添加依赖

<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>${protobuf.version}</version>
</dependency>
<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java-util</artifactId>
    <version>${protobuf.version}</version>
</dependency>
<dependency>
    <groupId>com.google.protobuf.nano</groupId>
    <artifactId>protobuf-javanano</artifactId>
    <version>${protobuf-javanano.version}</version>
</dependency>

到这里，就可以写代码了。

3.编写服务端

3.1通信的Protobuf文件定义，需根据使用场景自行编写，笔者提供单聊样例如下

message SingleChatRequest{
  uint64 msgSeq = 1; 
  string fromAccount = 2;
  string toAccount = 3;
  string msgContent = 4;
  Timestamp sendTime = 5;
}

message SingleChatResponse{
  uint64 msgSeq = 1; 
  SingleTalkStatus status = 2;
 
  enum SingleChatStatus{
    SUCCESS = 0;
    FAILED = -1;
  }
}

这里包含了几个最基本的定义，消息Seq，用于消息排序，发送人，接受人,消息内容，发送时间，以及发送结果的响应，读者需要根据自己的使用场景调整，接下来使用protobuf生成需要的的Java类就可以了

3.2定义基类，添加线程组生命钩子

abstract class AbstractSocketServer {
    protected final Logger logger = LoggerFactory.getLogger(getClass());
    private static final List<EventLoopGroup> EVENT_LOOP_GROUPS = new ArrayList<>();

    private void shutdown() {
        EVENT_LOOP_GROUPS.forEach(EventExecutorGroup::shutdownGracefully);
    }

    protected void registerShutdownHook(String hookName, EventLoopGroup... eventLoopGroup) {
        EVENT_LOOP_GROUPS.addAll(Arrays.asList(eventLoopGroup));
        Runtime.getRuntime().addShutdownHook(new Thread(hookName) {
            @Override
            public void run() {
                logger.info("Shutting down {}", hookName);
                shutdown();
            }
        });
    }
}

3.3添加我们自己的通道初始化基类

public abstract class AbstractNioChannelInitializer extends ChannelInitializer<NioSocketChannel> {

    protected void initPipeline(ChannelPipeline pipeline, ChannelHandler... handlers){
        initDecoderAndEncoder(pipeline);
        initSelfPipeline(pipeline);
        initCommonPipeline(pipeline, handlers);
    }

    final void initDecoderAndEncoder(ChannelPipeline pipeline){
        pipeline.addLast(new DefaultProtobufDecoder());
        pipeline.addLast(new DefaultProtobufEncoder());
    }

    final void initCommonPipeline(ChannelPipeline pipeline, ChannelHandler... handlers){
        Assert.notNull(handlers,"处理器不能为空");
        for (ChannelHandler channelHandler : handlers) {
            pipeline.addLast(channelHandler);
        }
    }
    protected void initSelfPipeline(ChannelPipeline pipeline){
    }
}

其中DefaultProtobufEncoder，DefaultProtobufDecoder是一层私有协议的实现，读者可以参考Netty提供的protobuf encode以及decode自行编写。作用就是包装发送的protobuf数据和拆解接受到的protobuf数据，拆解后的数据就是类似前面我们定义的SingleChatRequest生成的Java类，接下来我们只需添加一个简单的SimpleChannelInboundHandler处理器实现就可以处理发送过来的单聊数据了。

3.4Netty通道实现

public class NativeSocketInitializer extends AbstractNioChannelInitializer {
    
    @Override
    protected void initChannel(NioSocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        initPipeline(pipeline,some handlers);
    }
}

3.5Netty服务端

public class NativeSocketServer extends AbstractSocketServer implements ApplicationRunner {
    private static final String NATIVE_SOCKET_SERVER_SHUTDOWN_HOOK = "nativeSocketServer";
    private final NativeSocketInitializer nativeSocketInitializer;

    public NativeSocketServer(NativeSocketInitializer nativeSocketInitializer) {
        this.nativeSocketInitializer = nativeSocketInitializer;
    }

    @Override
    public void run(ApplicationArguments args) {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boss, worker).channel(NioServerSocketChannel.class).childHandler(nativeSocketInitializer);
            serverBootstrap.bind(9527).sync();
            if (logger.isInfoEnabled()) {
                logger.info("native socket server started on port : {}", 9527);
            }
            registerShutdownHook(NATIVE_SOCKET_SERVER_SHUTDOWN_HOOK, boss, worker);
        } catch (InterruptedException e) {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}