目录

背景

源码分析

参考

---

背景

为了方便观察redis的调用情况，接入了redis通用访问组件。项目redis设置socketTimeout时间为200ms，配置最大重试次数为1次。发现项目在调用redis时，有一些单个set语句会达到1s以上的耗时，然后也能拿到执行命令的结果。而根据参数配置，即便再重试一次，也最多400ms就超时了。

以下是11月25日api中的两个耗时日志：

RedisClusterClientProxy.printRunCommandTimeoutLog:87 redis cluster cost time: 1608, method name: lpush, key name: asyncnotification:queue

RedisClusterClientProxy.printRunCommandTimeoutLog:87 redis cluster cost time: 1609, method name: get, key name: orderInfo:order-10000074

源码分析

观察redis通用访问组件是用的Jedis的依赖包，通过查看源码，可以找到Connection类的socket、inputStream变量相关初始化

this.socket = new Socket();

this.socket.setReuseAddress(true);

this.socket.setKeepAlive(true);

this.socket.setTcpNoDelay(true);

this.socket.setSoLinger(true, 0);

this.socket.connect(new InetSocketAddress(this.host, this.port), this.connectionTimeout);

this.socket.setSoTimeout(this.soTimeout);

this.outputStream = new RedisOutputStream(this.socket.getOutputStream());

this.inputStream = new RedisInputStream(this.socket.getInputStream());

读取网络数据的时候，其底层其实是利用的SocketInputStream类来读取数据

AbstractPlainSocketImpl类

protected synchronized InputStream getInputStream() throws IOException {
    synchronized (fdLock) {
        if (isClosedOrPending())
            throw new IOException("Socket Closed");
        if (shut_rd)
            throw new IOException("Socket input is shutdown");
        if (socketInputStream == null)
            socketInputStream = new SocketInputStream(this);
    }
    return socketInputStream;
}

SocketInputStream类相关的read方法

int read(byte b[], int off, int length, int timeout) throws IOException {
        int n = 0;

        // EOF already encountered
        if (eof) {
            return -1;
        }

        // connection reset
        if (impl.isConnectionReset()) {
            throw new SocketException("Connection reset");
        }

        // bounds check
        if (length <= 0 || off < 0 || off + length > b.length) {
            if (length == 0) {
                return 0;
            }
            throw new ArrayIndexOutOfBoundsException();
        }

        boolean gotReset = false;

        Object traceContext = IoTrace.socketReadBegin();
        // acquire file descriptor and do the read
        FileDescriptor fd = impl.acquireFD();
        try {
            n = socketRead0(fd, b, off, length, timeout);
            if (n > 0) {
                return n;
            }
        } catch (ConnectionResetException rstExc) {
            gotReset = true;
        } finally {
            impl.releaseFD();
            IoTrace.socketReadEnd(traceContext, impl.address, impl.port,
                                  timeout, n > 0 ? n : 0);
        }

        /*
         * We receive a "connection reset" but there may be bytes still
         * buffered on the socket
         */
        if (gotReset) {
            traceContext = IoTrace.socketReadBegin();
            impl.setConnectionResetPending();
            impl.acquireFD();
            try {
                n = socketRead0(fd, b, off, length, timeout);
                if (n > 0) {
                    return n;
                }
            } catch (ConnectionResetException rstExc) {
            } finally {
                impl.releaseFD();
                IoTrace.socketReadEnd(traceContext, impl.address, impl.port,
                                      timeout, n > 0 ? n : 0);
            }
        }

        /*
         * If we get here we are at EOF, the socket has been closed,
         * or the connection has been reset.
         */
        if (impl.isClosedOrPending()) {
            throw new SocketException("Socket closed");
        }
        if (impl.isConnectionResetPending()) {
            impl.setConnectionReset();
        }
        if (impl.isConnectionReset()) {
            throw new SocketException("Connection reset");
        }
        eof = true;
        return -1;
    }

这个方法的核心其实就是 socketRead0(fd, b, off, length, timeout)的调用

private native int socketRead0(FileDescriptor fd,
                                   byte b[], int off, int len,
                                   int timeout)
        throws IOException;

项目使用的是OpenJDKjdk1.8.0_202，socketRead0是native方法，通过下载openjdk1.8源码，我们在openjdk\jdk\src\solaris\native\java\net的目录下找到了相关实现，在SocketInputStream.c文件里，代码如下：

Java_java_net_SocketInputStream_socketRead0(JNIEnv *env, jobject this,
                                            jobject fdObj, jbyteArray data,
                                            jint off, jint len, jint timeout)
{
    char BUF[MAX_BUFFER_LEN];
    char *bufP;
    jint fd, nread;

    if (IS_NULL(fdObj)) {
        /* shouldn't this be a NullPointerException? -br */
        JNU_ThrowByName(env, JNU_JAVANETPKG "SocketException",
                        "Socket closed");
        return -1;
    } else {
        fd = (*env)->GetIntField(env, fdObj, IO_fd_fdID);
        /* Bug 4086704 - If the Socket associated with this file descriptor
         * was closed (sysCloseFD), then the file descriptor is set to -1.
         */
        if (fd == -1) {
            JNU_ThrowByName(env, "java/net/SocketException", "Socket closed");
            return -1;
        }
    }

    /*
     * If the read is greater than our stack allocated buffer then
     * we allocate from the heap (up to a limit)
     */
    if (len > MAX_BUFFER_LEN) {
        if (len > MAX_HEAP_BUFFER_LEN) {
            len = MAX_HEAP_BUFFER_LEN;
        }
        bufP = (char *)malloc((size_t)len);
        if (bufP == NULL) {
            bufP = BUF;
            len = MAX_BUFFER_LEN;
        }
    } else {
        bufP = BUF;
    }

    if (timeout) {
        nread = NET_Timeout(fd, timeout);
        if (nread <= 0) {
            if (nread == 0) {
                JNU_ThrowByName(env, JNU_JAVANETPKG "SocketTimeoutException",
                            "Read timed out");
            } else if (nread == JVM_IO_ERR) {
                if (errno == EBADF) {
                     JNU_ThrowByName(env, JNU_JAVANETPKG "SocketException", "Socket closed");
                 } else if (errno == ENOMEM) {
                     JNU_ThrowOutOfMemoryError(env, "NET_Timeout native heap allocation failed");
                 } else {
                     NET_ThrowByNameWithLastError(env, JNU_JAVANETPKG "SocketException",
                                                  "select/poll failed");
                 }
            } else if (nread == JVM_IO_INTR) {
                JNU_ThrowByName(env, JNU_JAVAIOPKG "InterruptedIOException",
                            "Operation interrupted");
            }
            if (bufP != BUF) {
                free(bufP);
            }
            return -1;
        }
    }

    nread = NET_Read(fd, bufP, len);

    if (nread <= 0) {
        if (nread < 0) {

            switch (errno) {
                case ECONNRESET:
                case EPIPE:
                    JNU_ThrowByName(env, "sun/net/ConnectionResetException",
                        "Connection reset");
                    break;

                case EBADF:
                    JNU_ThrowByName(env, JNU_JAVANETPKG "SocketException",
                        "Socket closed");
                    break;

                case EINTR:
                     JNU_ThrowByName(env, JNU_JAVAIOPKG "InterruptedIOException",
                           "Operation interrupted");
                     break;

                default:
                    NET_ThrowByNameWithLastError(env,
                        JNU_JAVANETPKG "SocketException", "Read failed");
            }
        }
    } else {
        (*env)->SetByteArrayRegion(env, data, off, nread, (jbyte *)bufP);
    }

    if (bufP != BUF) {
        free(bufP);
    }
    return nread;
}

数据的读取是通过NET_Timeout (fd, timeout)来实现的，在linux_close.c文件中，发现了NET_Timeout的实现：

int NET_Timeout(int s, long timeout) {
    long prevtime = 0, newtime;
    struct timeval t;
    fdEntry_t *fdEntry = getFdEntry(s);

    /*
     * Check that fd hasn't been closed.
     */
    if (fdEntry == NULL) {
        errno = EBADF;
        return -1;
    }

    /*
     * Pick up current time as may need to adjust timeout
     */
    if (timeout > 0) {
        gettimeofday(&t, NULL);
        prevtime = t.tv_sec * 1000  +  t.tv_usec / 1000;
    }

    for(;;) {
        struct pollfd pfd;
        int rv;
        threadEntry_t self;

        /*
         * Poll the fd. If interrupted by our wakeup signal
         * errno will be set to EBADF.
         */
        pfd.fd = s;
        pfd.events = POLLIN | POLLERR;

        startOp(fdEntry, &self);
        rv = poll(&pfd, 1, timeout);
        endOp(fdEntry, &self);

        /*
         * 如果中断，则调整超时。如果超时
         * 已过期返回0(表示超时已过期)。
         */
        if (rv < 0 && errno == EINTR) {
            if (timeout > 0) {
                gettimeofday(&t, NULL);
                newtime = t.tv_sec * 1000  +  t.tv_usec / 1000;
                timeout -= newtime - prevtime;
                if (timeout <= 0) {
                    return 0;
                }
                prevtime = newtime;
            }
        } else {
            return rv;
        }

    }
}

代码中的关键点在 poll(&pfd, 1, timeout)。poll是linux中的字符设备驱动中的一个函数，作用是把当前的文件指针挂到设备内部定义的等待。

这样就很好理解了，其实这个时间是我两次读取数据之间的最长阻塞时间，如果我在网络抖动的情况下，我每次200ms之内返回一部分数据，这样我就一直不会超时了。

根据原理复现现象

因为不好用redis控制间隔发送一部分数据的现象，所以我们可以另辟蹊径，http读取网络数据，用的也是java的io类，其底层其实也是利用的SocketInputStream来读取数据。

private static Logger LOGGER = LoggerFactory.getLogger(HomeController.class);

@RequestMapping("/testSocketTimeout")
public void testSocketTimeout(HttpServletRequest request , HttpServletResponse response) throws Exception {
    PrintWriter writer = response.getWriter();
    int i = 2;
    while (--i >= 0){
        writer.print("hello world! I'm niwanjia!");
        writer.flush();
        Thread.sleep(2000);
        LOGGER.info("testSocketTimeout");
    }
}

这个代码就是每隔2s发送一条数据，循环发送，模拟网络不好的时候，收到的数据断断续续，再来一个test用来发送请求，直接进行debug测试：

@Test
public void tetsHttpClientHttp() throws IOException {
    RequestConfig requestConfig = RequestConfig.custom()
            .setConnectTimeout(1000)
            .setSocketTimeout(3000)
            .setConnectionRequestTimeout(1000)
            .build();

    CloseableHttpClient httpClient = HttpClientBuilder.create().setDefaultRequestConfig(requestConfig).build();
    HttpGet httpGet = new HttpGet("http:/127.0.0.1/testSocketTimeout");
    CloseableHttpResponse response =httpClient.execute(httpGet);
    HttpEntity responseEntity = response.getEntity();
    if (responseEntity != null) {
        System.out.println("响应内容为:" + EntityUtils.toString(responseEntity));
    }
}

客户端返回结果：

响应内容为:hello world! I'm niwanjia!hello world! I'm niwanjia!

当我们修改socketTimeout为1000时，经验证可以抛出java.net.SocketTimeoutException: Read timed out 异常

java.net.SocketTimeoutException: Read timed out

at java.net.SocketInputStream.socketRead0(Native Method)

at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)

at java.net.SocketInputStream.read(SocketInputStream.java:170)

at java.net.SocketInputStream.read(SocketInputStream.java:141)

总结

1、假设设置JedisClient客户端的socketTimeout为200ms, 如果在网络抖动的情况下，每次200ms之内返回一部分数据，这样就没有违反两次读取数据之间的最长阻塞时间，因此也就一直不会超时了。

2、上述复现，如果把ip地址换成域名地址来请求，这种现象却没有复现，目前暂时不知道原因，后续有空再研究看看；

3、假设A服务通过thrift方式调用B服务，而B服务有通过redis做处理，那么这时会出现B虽然执行成功，但A却认为执行thrift超时了，影响事件的结果。如果这种现象经常发生，其实是挺恶劣的，应用程序可能会一直等待服务器返回结果。