Android(安卓)okhttp3 利用socket进行read/write的底层实现跟踪

在okhttp3.internal.io.RealConnection#connectSocket中初始化了socket并进行了connect，此时tcp的三次握手已经搞定，接下来它通过okio库与远程socket建立I/O连接，如下代码所示：

  /** Does all the work necessary to build a full HTTP or HTTPS connection on a raw socket. */private void connectSocket(int connectTimeout, int readTimeout, int writeTimeout,      ConnectionSpecSelector connectionSpecSelector) throws IOException {    rawSocket.setSoTimeout(readTimeout);    try {      Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout);    } catch (ConnectException e) {      throw new ConnectException("Failed to connect to " + route.socketAddress());    }    source = Okio.buffer(Okio.source(rawSocket));    sink = Okio.buffer(Okio.sink(rawSocket));。。。}

Okio库是一个由square公司开发的，它补充了Java.io和java.nio的不足，以便能够更加方便，快速的访问、存储和处理你的数据。而OkHttp的底层也使用该库作为支持。

Okio中有两个关键的接口，Sink和Source，这两个接口都继承了Closeable接口；而Sink可以简单的看做OutputStream，Source可以简单的看做InputStream。而这两个接口都是支持读写超时设置的。它们各自有一个支持缓冲区的子类接口，BufferedSink和BufferedSource，而BufferedSink有一个实现类RealBufferedSink,BufferedSource有一个实现类RealBufferedSource；此外，Sink和Source它门还各自有一个支持gzip压缩的实现类GzipSink和GzipSource；一个具有委托功能的抽象类ForwardingSink和ForwardingSource；还有一个实现类便是InflaterSource和DeflaterSink，这两个类主要用于压缩，为GzipSink和GzipSource服务；整体的结构图如下

接下来以read为例，追踪底层实现（write的逻辑是类似的）。

1.okhttp3.internal.io.RealConnection#connectSocket

source = Okio.buffer(Okio.source(rawSocket));

2.okio#source

public static Source source(Socket socket) throws IOException {    if (socket == null) throw new IllegalArgumentException("socket == null");    AsyncTimeout timeout = timeout(socket);    Source source = source(socket.getInputStream(), timeout);    return timeout.source(source);}

在这里从socket拿InputStream

3./libcore/luni/src/main/java/java/net/Socket.java

    public InputStream getInputStream() throws IOException {        checkOpenAndCreate(false);        if (isInputShutdown()) {            throw new SocketException("Socket input is shutdown");        }        return impl.getInputStream();    }

4./libcore/luni/src/main/java/java/net/PlainSocketImpl.java

    @Override protected synchronized InputStream getInputStream() throws IOException {        checkNotClosed();        return new PlainSocketInputStream(this);    }

5./libcore/luni/src/main/java/java/net/PlainSocketImpl.java

    private static class PlainSocketInputStream extends InputStream {        private final PlainSocketImpl socketImpl;        public PlainSocketInputStream(PlainSocketImpl socketImpl) {            this.socketImpl = socketImpl;        }        @Override public int available() throws IOException {            return socketImpl.available();        }        @Override public void close() throws IOException {            socketImpl.close();        }        @Override public int read() throws IOException {            return Streams.readSingleByte(this);        }        @Override public int read(byte[] buffer, int offset, int byteCount) throws IOException {            return socketImpl.read(buffer, offset, byteCount);        }    }

接下来以read(byte[] buffer, int offset, int byteCount)为例。

6./libcore/luni/src/main/java/java/net/PlainSocketImpl.java

    private int read(byte[] buffer, int offset, int byteCount) throws IOException {        if (byteCount == 0) {            return 0;        }        Arrays.checkOffsetAndCount(buffer.length, offset, byteCount);        if (shutdownInput) {            return -1;        }        int readCount = IoBridge.recvfrom(true, fd, buffer, offset, byteCount, 0, null, false);        // Return of zero bytes for a blocking socket means a timeout occurred        if (readCount == 0) {            throw new SocketTimeoutException();        }        // Return of -1 indicates the peer was closed        if (readCount == -1) {            shutdownInput = true;        }        return readCount;    }

IoBridge.recvfrom(true, fd, buffer, offset, byteCount, 0, null, false)再次开始去调jni

7./libcore/luni/src/main/java/libcore/io/IoBridge.java

    public static int recvfrom(boolean isRead, FileDescriptor fd, ByteBuffer buffer, int flags, DatagramPacket packet, boolean isConnected) throws IOException {        int result;        try {            InetSocketAddress srcAddress = (packet != null && !isConnected) ? new InetSocketAddress() : null;            result = Libcore.os.recvfrom(fd, buffer, flags, srcAddress);            result = postRecvfrom(isRead, packet, isConnected, srcAddress, result);        } catch (ErrnoException errnoException) {            result = maybeThrowAfterRecvfrom(isRead, isConnected, errnoException);        }        return result;    }    private static int postRecvfrom(boolean isRead, DatagramPacket packet, boolean isConnected, InetSocketAddress srcAddress, int byteCount) {        if (isRead && byteCount == 0) {            return -1;        }        if (packet != null) {            packet.setReceivedLength(byteCount);            if (!isConnected) {                packet.setAddress(srcAddress.getAddress());                packet.setPort(srcAddress.getPort());            }        }        return byteCount;    }

==>Libcore.os.recvfrom

8.
/libcore/luni/src/main/java/libcore/io/BlockGuardOs.java

    @Override public int recvfrom(FileDescriptor fd, ByteBuffer buffer, int flags, InetSocketAddress srcAddress) throws ErrnoException, SocketException {        BlockGuard.getThreadPolicy().onNetwork();        return os.recvfrom(fd, buffer, flags, srcAddress);    }

/libcore/luni/src/main/java/libcore/io/ForwardingOs.java

public int recvfrom(FileDescriptor fd, ByteBuffer buffer, int flags, InetSocketAddress srcAddress) throws ErrnoException, SocketException {     return os.recvfrom(fd, buffer, flags, srcAddress); }

9./libcore/luni/src/main/java/libcore/io/Posix.java

    public int recvfrom(FileDescriptor fd, ByteBuffer buffer, int flags, InetSocketAddress srcAddress) throws ErrnoException, SocketException {        if (buffer.isDirect()) {            return recvfromBytes(fd, buffer, buffer.position(), buffer.remaining(), flags, srcAddress);        } else {            return recvfromBytes(fd, NioUtils.unsafeArray(buffer), NioUtils.unsafeArrayOffset(buffer) + buffer.position(), buffer.remaining(), flags, srcAddress);        }    }private native int recvfromBytes(FileDescriptor fd, Object buffer, int byteOffset, int byteCount, int flags, InetSocketAddress srcAddress) throws ErrnoException, SocketException;

终于看到了jni的声明

10./libcore/luni/src/main/native/libcore_io_Posix.cpp

static jint Posix_recvfromBytes(JNIEnv* env, jobject, jobject javaFd, jobject javaBytes, jint byteOffset, jint byteCount, jint flags, jobject javaInetSocketAddress) {    ScopedBytesRW bytes(env, javaBytes);    if (bytes.get() == NULL) {        return -1;    }    sockaddr_storage ss;    socklen_t sl = sizeof(ss);    memset(&ss, 0, sizeof(ss));    sockaddr* from = (javaInetSocketAddress != NULL) ? reinterpret_cast(&ss) : NULL;    socklen_t* fromLength = (javaInetSocketAddress != NULL) ? &sl : 0;    jint recvCount = NET_FAILURE_RETRY(env, ssize_t, recvfrom, javaFd, bytes.get() + byteOffset, byteCount, flags, from, fromLength);    fillInetSocketAddress(env, recvCount, javaInetSocketAddress, ss);    return recvCount;}#define NET_FAILURE_RETRY(jni_env, return_type, syscall_name, java_fd, ...) ({ \    return_type _rc = -1; \    do { \        { \            int _fd = jniGetFDFromFileDescriptor(jni_env, java_fd); \            AsynchronousSocketCloseMonitor _monitor(_fd); \            _rc = syscall_name(_fd, __VA_ARGS__); \        } \        if (_rc == -1) { \            if (jniGetFDFromFileDescriptor(jni_env, java_fd) == -1) { \                jniThrowException(jni_env, "java/net/SocketException", "Socket closed"); \                break; \            } else if (errno != EINTR) { \                /* TODO: with a format string we could show the arguments too, like strace(1). */ \                throwErrnoException(jni_env, # syscall_name); \                break; \            } \        } \    } while (_rc == -1); \    _rc; })

这边是jni的实现

11./bionic/libc/arch-arm/syscalls/recvfrom.S

ENTRY(recvfrom)    mov     ip, sp    .save   {r4, r5, r6, r7}    stmfd   sp!, {r4, r5, r6, r7}    ldmfd   ip, {r4, r5, r6}    ldr     r7, =__NR_recvfrom    swi     #0    ldmfd   sp!, {r4, r5, r6, r7}    cmn     r0, #(MAX_ERRNO + 1)    bxls    lr    neg     r0, r0    b       __set_errnoEND(recvfrom)

最终recvfrom是用汇编实现的，使用swi进行了系统调用

更多相关文章

随机推荐