关于tars:TarsJava网络编程源码分析

作者：vivo 互联网服务器团队- Jin Kai

本文从Java NIO网络编程的基础知识讲到了Tars框架应用NIO进行网络编程的源码剖析。

一、Tars框架根本介绍

Tars是腾讯开源的反对多语言的高性能RPC框架，起源于腾讯外部2008年至今始终应用的对立利用框架TAF（Total Application Framework），目前反对C++、Java、PHP、Nodejs、Go语言。

该框架为用户提供了波及到开发、运维、以及测试的一整套解决方案，帮忙一个产品或者服务疾速开发、部署、测试、上线。它集可扩大协定编解码、高性能RPC通信框架、名字路由与发现、公布监控、日志统计、配置管理等于一体，通过它能够疾速用微服务的形式构建本人的稳固牢靠的分布式应用，并实现残缺无效的服务治理。

官网仓库地址：

https://github.com/TarsCloud/Tars

vivo推送平台也深度应用了该框架，部署服务节点超过一千个，通过线上每日一百多亿音讯推送量的考验。

此前已在vivo互联网技术公众号公布过《Tars Java 客户端源码剖析》，此篇文章为续集。

Tars-java 最新稳定版1.7.2以及之前的版本都应用Java NIO进行网络编程；本文将别离具体介绍java NIO的原理和Tars 应用NIO进行网络编程的细节。

二、Java NIO原理介绍

从1.4版本开始，Java提供了一种新的IO解决形式：NIO (New IO 或 Non-blocking IO) 是一个能够代替规范Java IO 的API，它是面向缓冲区而不是字节流，它是非阻塞的，反对IO多路复用。

2.1 Channels (通道) and Buffers (缓冲区)

规范的IO基于字节流进行操作的，而NIO是基于通道（Channel）和缓冲区（Buffer）进行操作。数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中，下图是一个残缺流程。

Channel类型：

反对文件读写数据的FileChannel
能通过UDP读写网络中的数据的DatagramChannel
能通过TCP读写网络数据的SocketChannel
能够监听新进来的TCP连贯，对每一个新进来的连贯都会创立一个SocketChannel的ServerSocketChannel 。

SocketChannel：

关上 SocketChannel：SocketChannel socketChannel = SocketChannel.open();
敞开 SocketChannel：socketChannel.close();
从Channel中读取的数据放到Buffer： int bytesRead = inChannel.read(buf);
将Buffer中的数据写到Channel： int bytesWritten = inChannel.write(buf);

ServerSocketChannel：

通过 ServerSocketChannel.accept() 办法监听新进来的连贯，当accept()办法返回的时候，它返回一个蕴含新进来的连贯的SocketChannel，因而accept()办法会始终阻塞到有新连贯达到。

通常不会仅仅只监听一个连贯,在while循环中调用 accept()办法. 如上面的例子：

代码1：

while(true){
    SocketChannel socketChannel = serverSocketChannel.accept();
     //do something with socketChannel...
}

ServerSocketChannel能够设置成非阻塞模式。在非阻塞模式下，accept() 办法会立即返回，如果还没有新进来的连贯,返回的将是null。因而，须要查看返回的SocketChannel是否是null。

代码2：

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(8888));
serverSocketChannel.configureBlocking(false);
while(true){
    SocketChannel socketChannel = serverSocketChannel.accept();
    if(socketChannel != null){
        //do something with socketChannel...
    }
}

Buffer类型：

ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer

Buffer的调配：

ByteBuffer buf = ByteBuffer.allocate(2048);

Buffer的读写：

个别是以下四个步骤：

写入数据到Buffer，最大写入量是capacity，写模式下limit值即为capacity值，position即为写到的地位。

调用flip()办法将Buffer从写模式切换到读模式，此时position挪动到开始地位0，limit挪动到position的地位。

从Buffer中读取数据，在读模式下能够读取之前写入到buffer的所有数据，即为limit地位。

调用clear()办法或者compact()办法。clear()办法将position设为0，limit被设置成capacity的值。compact()办法将所有未读的数据拷贝到Buffer起始处，而后将position设到最初一个未读元素前面。

mark() 与 reset()办法 通过调用Buffer.mark()办法，能够标记Buffer中的一个特定position，之后能够通过调用Buffer.reset()办法复原到这个position。

duplicate() 此办法返回承载先前字节缓冲区内容的新字节缓冲区。

remaining() limit 减去 position的值

2.2 Selector（选择器）

Java NIO引入了选择器的概念，选择器用于监听多个通道的事件。单个的线程能够监听多个数据通道。要应用Selector，得向Selector注册Channel，而后调用它的select()办法。这个办法会始终阻塞到某个注册的通道有事件就绪。一旦这个办法返回，线程就能够解决这些事件。

代码3：

channel.configureBlocking(false);
SelectionKey key = channel.register(selector,Selectionkey.OP_READ);

留神register()办法的第二个参数，这是一个监听的汇合，即在通过Selector监听Channel时关注什么事件汇合。

SelectionKey蕴含：

1) interest汇合：
selectionKey.interestOps() 能够监听四种不同类型的事件：OP\_ACCEPT、OP\_CONNECT、OP\_WRITE、OP\_READ

2) ready汇合：
selectionKey.readyOps(); ready 汇合是通道曾经准备就绪的操作的汇合，提供4个不便的办法：

selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

3) Channel： selectionKey.channel();

4) Selector： selectionKey.selector();

5) 可选的附加对象：

selectionKey.attachment(); 能够将一个对象或者更多信息附着到SelectionKey上，这样就能不便的辨认特定的通道。

提醒：

OP\_ACCEPT和OP\_CONNECT的区别：简略来说，客户端建设连贯是connect，服务器筹备接管连贯是accept。一个典型的客户端服务器网络交互流程如下图

selectedKeys() 一旦调用了select()办法，并且返回值表明有一个或更多个通道就绪了，而后能够通过调用selector的selectedKeys()办法，拜访已选择键集（selected key set）中的就绪通道。

wakeUp() 某个线程调用select()办法后阻塞了，即便没有通道曾经就绪，也有方法让其从select()办法返回。只有让其它线程在阻塞线程调用select()办法的对象上调用Selector.wakeup()办法即可。阻塞在select()办法上的线程会立马返回。如果有其它线程调用了wakeup()办法，但以后没有线程阻塞在select()办法上，下个调用select()办法的线程会立刻wake up。

close() 用完Selector后调用其close()办法会敞开该Selector，且使注册到该Selector上的所有SelectionKey实例有效。通道自身并不会敞开。

通过Selector抉择通道：

int select() 阻塞直到至多有一个通道在你注册的事件上就绪了
int select(long timeout) 减少最长阻塞毫秒数
int selectNow() 不会阻塞，不论什么通道就绪都立即返回

三、 Tars NIO网络编程

理解完 Java NIO的原理，咱们来看看Tars是如何应用NIO进行网络编程的。

Tars的网络模型是多reactor多线程模型。有一点非凡的是tars的reactor线程组里随机选一个线程解决网络事件，并且该线程同时也能解决读写。

外围类之间的关系如下：

3.1 一个典型的Java NIO服务端开发流程

创立ServerSocketChannel,设置为非阻塞，并绑定端口

创立Selector对象

给ServerSocketChannel注册SelectionKey.OP_ACCEPT事件

启动一个线程循环，调用Selector的select办法来查看IO就绪事件，一旦有IO就绪事件，就告诉用户线程去解决IO事件

如果有Accept事件，就创立一个SocketChannel，并注册SelectionKey.OP_READ

如果有读事件，判断一下是否全包，如果全包，就交给后端线程解决

写事件比拟非凡。isWriteable示意的是本机的写缓冲区是否可写。这个在绝大多少状况下都是为真的。在Netty中只有写半包的时候才须要注册写事件，如果一次写就齐全把数据写入了缓冲区就不须要注册写事件。

3.2 Tars客户端发动申请到服务器的流程

Communicator.stringToProxy()  依据servantName等配置信息创立通信器。

ServantProxyFactory.getServantProxy() 调用工厂办法创立servant代理。

ObjectProxyFactory.getObjectProxy()  调用工厂办法创立obj代理。

TarsProtocolInvoker.create() 创立协定调用者。

ServantProtocolInvoker.initClient(Url url)  依据servantProxyConfig中的配置信息找到servant的ip端口等进行初始化ServantClient。

ClientPoolManager.getSelectorManager() 如果第一次调用selectorManager是空的就会去初始化selectorManager。

reactorSet = new Reactor[selectorPoolSize]; SelectorManager初始化结构类中的会依据selectorPoolSize（默认是2）的配置创立Reactor线程数组。线程名称的前缀是servant-proxy-加上CommunicatorId，CommunicatorId生成规定是由locator的地址生成的UUID。

启动reactor线程。

3.3 Tars服务端启动步骤

tars反对TCP和UDP两种协定，RPC场景下是应用TCP协定。

new SelectorManager() 依据配置信息初始化selectorManager，线程池大小 processors > 8 ? 4 + (processors * 5 / 8) : processors + 1；线程名称前缀是server-tcp-reactor，而后启动reactor线程数组中的所有线程。

开启服务端监听的ServerSocketChannel，绑定服务端本地ip和监听的端口号，设置TCP连贯申请队列的最大容量为1024；设置非阻塞模式。

选取reactor线程数组中第0个线程作为服务端监听连贯OP_ACCEPT就绪事件的线程。

代码4：

public void bind(AppService appService) throws IOException {
 
    // 此处略去非关键代码
 
    if (endpoint.type().equals("tcp")) {  // 1
        this.selectorManager = new SelectorManager(Utils.getSelectorPoolSize(), new ServantProtocolFactory(codec), threadPool, processor, keepAlive, "server-tcp-reactor", false);     // 2
        this.selectorManager.setTcpNoDelay(serverCfg.isTcpNoDelay());
        this.selectorManager.start();
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.socket().bind(new InetSocketAddress(endpoint.host(), endpoint.port()), 1024);   // 3
        serverChannel.configureBlocking(false);
              selectorManager.getReactor(0).registerChannel(serverChannel, SelectionKey.OP_ACCEPT);  // 4
    } else if (endpoint.type().equals("udp")) {
        this.selectorManager = new SelectorManager(1, new ServantProtocolFactory(codec), threadPool, processor, false, "server-udp-reactor", true);
        this.selectorManager.start();
        // UDP开启的是DatagramChannel
        DatagramChannel serverChannel = DatagramChannel.open();
        DatagramSocket socket = serverChannel.socket();
        socket.bind(new InetSocketAddress(endpoint.host(), endpoint.port()));
        serverChannel.configureBlocking(false);
        // UDP协定不须要建连，监听的是OP_READ就绪事件
        this.selectorManager.getReactor(0).registerChannel(serverChannel, SelectionKey.OP_READ);
    }
}

3.4 Reactor线程启动流程

多路复用器开始轮询查看是否有就绪的事件。

解决register队列中残余的channel注册到以后reactor线程的多路复用器selector中。

获取已选键集中所有就绪的channel。

更新Session中最近操作工夫，Tars服务端启动时会调用 startSessionManager() , 单线程每30s扫描一次session会话列表，会查看每个session的 lastUpdateOperationTime 与以后工夫的时间差，如果超过60秒会将过期session对应的channel踢除。

散发IO事件进行解决。

解决unregister队列中残余的channel,从以后reactor线程的多路复用器selector中解除注册。

代码5：

public void run() {
        while (!Thread.interrupted()) {
            selector.select();  // 1
            processRegister();  // 2
            Iterator<SelectionKey> iter = selector.selectedKeys().iterator();   //  3
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                if (!key.isValid()) continue;
                try {
                    if (key.attachment() != null && key.attachment() instanceof Session) {
                      ((Session) key.attachment()).updateLastOperationTime(); //4
                    }
                 dispatchEvent(key);    // 5
                } catch (Throwable ex) {
                 disConnectWithException(key, ex);
                }
            }
            processUnRegister();  // 6
        }
}

3.5 IO事件散发解决

每个reactor线程都有一个专门的Accepter类去解决各种IO事件。TCPAccepter能够解决全副的四种事件（OP\_ACCEPT、OP\_CONNECT、OP\_WRITE、OP\_READ）、UDPAccepter因为不须要建设连贯所以只须要解决读和写两种事件。

1. 解决OP_ACCEPT

获取channel，解决TCP申请。

为这个TCP申请创立TCPSession，会话的状态是服务器已连贯

会话注册到sessionManager中，Tars服务可配置最大连接数maxconns，如果超过就会敞开以后会话。

寻找下一个reactor线程进行多路复用器与channel的绑定。

代码6：

public void handleAcceptEvent(SelectionKey key) throws IOException {
    ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 1
    SocketChannel channel = server.accept();
       channel.socket().setTcpNoDelay(selectorManager.isTcpNoDelay());
    channel.configureBlocking(false);
    Utils.setQosFlag(channel.socket());
    TCPSession session = new TCPSession(selectorManager);    // 2
    session.setChannel(channel);
    session.setStatus(SessionStatus.SERVER_CONNECTED);
    session.setKeepAlive(selectorManager.isKeepAlive());
    session.setTcpNoDelay(selectorManager.isTcpNoDelay());
    SessionManager.getSessionManager().registerSession(session);   // 3
      selectorManager.nextReactor().registerChannel(channel, SelectionKey.OP_READ, session); // 4
}

2. 解决OP_CONNECT

获取客户端连贯过去的channel通道

获取Session

与服务器建设连贯，将关注的趣味OPS设置为ready就绪事件，session中的状态批改为客户端已连贯

解决OP_CONNECT

代码7：

public void handleConnectEvent(SelectionKey key) throws IOException {
    SocketChannel client = (SocketChannel) key.channel();  // 1
    TCPSession session = (TCPSession) key.attachment();   //2
    if (session == null) throw new RuntimeException("The session is null when connecting to ...");
    try {  // 3
        client.finishConnect();
        key.interestOps(SelectionKey.OP_READ);
        session.setStatus(SessionStatus.CLIENT_CONNECTED);
    } finally {
        session.finishConnect();
    }
}

3.解决OP_WRITE、解决OP_READ

代码8：

public void handleReadEvent(SelectionKey key) throws IOException {
    TCPSession session = (TCPSession) key.attachment();
    if (session == null) throw new RuntimeException("The session is null when reading data...");
    session.read();
}
public void handleWriteEvent(SelectionKey key) throws IOException {
    TCPSession session = (TCPSession) key.attachment();
    if (session == null) throw new RuntimeException("The session is null when writing data...");
    session.doWrite();
}

3.6 seesion中网络读写的事件具体处理过程

1. 读事件处理

申请2k的ByteBuffer空间，读取channel中的数据到readBuffer中。依据sessionStatus判断是客户端读响应还是服务器读申请，别离进行解决。

代码9：

protected void read() throws IOException {
    int ret = readChannel();
    if (this.status == SessionStatus.CLIENT_CONNECTED) {
        readResponse();
    } else if (this.status == SessionStatus.SERVER_CONNECTED) {
        readRequest();
    } else {
        throw new IllegalStateException("The current session status is invalid. [status:" + this.status + "]");
    }
    if (ret < 0) {
        close();
        return;
    }
}
private int readChannel() throws IOException {
    int readBytes = 0, ret = 0;
    ByteBuffer data = ByteBuffer.allocate(1024 * 2);  // 1
    if (readBuffer == null) {
        readBuffer = IoBuffer.allocate(bufferSize);
    }
       // 2
    while ((ret = ((SocketChannel) channel).read(data)) > 0) {
        data.flip();  // 3
        readBytes += data.remaining();
        readBuffer.put(data.array(), data.position(), data.remaining());
        data.clear();
    }
    return ret < 0 ? ret : readBytes;
}

① 客户端读响应

从以后readBuffer中的内容复制到一个新的长期buffer中，并且切换到读模式，应用TarsCodec类解析出buffer内的协定字段到response，WorkThread线程告诉Ticket解决response。如果response为空，则重置tempBuffer到mark的地位，从新解析协定。

代码10：

public void readResponse() {
    Response response = null;
    IoBuffer tempBuffer = null;
        tempBuffer = readBuffer.duplicate().flip();
        while (true) {
            tempBuffer.mark();
            if (tempBuffer.remaining() > 0) {
                response = selectorManager.getProtocolFactory().getDecoder().decodeResponse(tempBuffer, this);
            } else {
                response = null;
            }
            if (response != null) {
                if (response.getTicketNumber() == Ticket.DEFAULT_TICKET_NUMBER) response.setTicketNumber(response.getSession().hashCode());
                selectorManager.getThreadPool().execute(new WorkThread(response, selectorManager));
            } else {
                tempBuffer.reset();
                readBuffer = resetIoBuffer(tempBuffer);
                break;
            }
        }
}

② 服务器读申请

工作放入线程池交给 WorkThread线程，最终交给Processor类出构建申请的响应体，包含分布式上下文，而后通过FilterChain的解决，最终通过jdk提供的反射办法invoke服务端本地的办法而后返回response。如果线程池抛出回绝异样，则返回SERVEROVERLOAD = -9，服务端过载爱护。如果request为空，则重置tempBuffer到mark的地位，从新解析协定。

代码11：

public void readRequest() {
    Request request = null;
    IoBuffer tempBuffer = readBuffer.duplicate().flip();
        while (true) {
            tempBuffer.mark();
            if (tempBuffer.remaining() > 0) {
                request = selectorManager.getProtocolFactory().getDecoder().decodeRequest(tempBuffer, this);
            } else {
                request = null;
            }
            if (request != null) {
                try {
                    request.resetBornTime();
                    selectorManager.getThreadPool().execute(new WorkThread(request, selectorManager));
                } catch (RejectedExecutionException e) {
                  selectorManager.getProcessor().overload(request, request.getIoSession());
                } catch (Exception ex) {
                  ex.printStackTrace();
                }
            } else {    
                tempBuffer.reset();
                readBuffer = resetIoBuffer(tempBuffer);
                break;
            }
        }
}

2. 写事件处理

同样也包含客户端写申请和服务端写响应两种，其实这两种都是往TCPSession中的LinkedBlockingQueue（有界队列最大8K）中插入ByteBuffer。LinkedBlockingQueue中的ByteBuffer最终会由TCPAcceptor中的handleWriteEvent监听写就绪事件并生产。

代码12：

protected void write(IoBuffer buffer) throws IOException {
    if (buffer == null) return;
    if (channel == null || key == null) throw new IOException("Connection is closed");
    if (!this.queue.offer(buffer.buf())) {
        throw new IOException("The session queue is full. [ queue size:" + queue.size() + " ]");
    }
    if (key != null) {
        key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
        key.selector().wakeup();
    }
}

四、总结

本文次要介绍了Java NIO编程的基础知识和 Tars-Java 1.7.2版本的网络编程模块的源码实现。

在最新的Tars-Java的master分支中咱们能够发现网络编程曾经由NIO改成了Netty，尽管Netty更加成熟稳固，然而作为学习者理解NIO的原理也是把握网络编程的必经之路。

更多对于Tars框架的介绍能够拜访：

https://tarscloud.org/

本文剖析源码地址（v1.7.x分支）：

https://github.com/TarsCloud/TarsJava

关于tars:TarsJava网络编程源码分析

一、Tars框架根本介绍

二、Java NIO原理介绍

2.1 Channels (通道) and Buffers (缓冲区)

2.2 Selector（选择器）

三、 Tars NIO网络编程

3.1 一个典型的Java NIO服务端开发流程

3.2 Tars客户端发动申请到服务器的流程

3.3 Tars服务端启动步骤

3.4 Reactor线程启动流程

3.5 IO事件散发解决

3.6 seesion中网络读写的事件具体处理过程

四、总结

评论

发表回复取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

关于tars:TarsJava网络编程源码分析

一、Tars框架根本介绍

二、Java NIO原理介绍

2.1 Channels (通道) and Buffers (缓冲区)

2.2 Selector（选择器）

三、 Tars NIO网络编程

3.1 一个典型的Java NIO服务端开发流程

3.2 Tars客户端发动申请到服务器的流程

3.3 Tars服务端启动步骤

3.4 Reactor线程启动流程

3.5 IO事件散发解决

3.6 seesion中网络读写的事件具体处理过程

四、总结

评论

发表回复 取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

发表回复取消回复