标签归档:sockettimeoutexception

Jedis不相信眼泪

最近愉快地上线了一个新的调度系统,主节点和从节点使用redis队列进行通信,然后奇怪的事情发生了,有一个海外的节点,和redis之间的网络一直比较差,经常出现异常(类似下面):

redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketTimeoutException: Read timed out
    at redis.clients.jedis.Protocol.process(Protocol.java:79)
    at redis.clients.jedis.Protocol.read(Protocol.java:131)
    at redis.clients.jedis.Connection.getAll(Connection.java:225)
    at redis.clients.jedis.Connection.getAll(Connection.java:217)
    at redis.clients.jedis.Pipeline.sync(Pipeline.java:68)
    at com.game.data.stat.share.redis.JedisUtil.batchPop(JedisUtil.java:151)
    at com.game.data.stat.share.db.JedisUtilTest.testBatchPop(JedisUtilTest.java:185)
    at com.game.data.stat.share.db.JedisUtilTest.main(JedisUtilTest.java:488)
Caused by: java.net.SocketTimeoutException: Read timed out
    at java.net.SocketInputStream.socketRead0(Native Method)
    at java.net.SocketInputStream.read(Unknown Source)
    at java.net.SocketInputStream.read(Unknown Source)
    at java.net.SocketInputStream.read(Unknown Source)
    at redis.clients.util.RedisInputStream.fill(RedisInputStream.java:109)
    at redis.clients.util.RedisInputStream.readByte(RedisInputStream.java:45)
    at redis.clients.jedis.Protocol.process(Protocol.java:64)
    ... 7 more

一开始以为是纯粹的网络延时太大,于是加大了的socket的超时时间,可是这个异常还是如大姨妈一般时常来访,而在重启服务之后,异常便会消失。
于是在本地做了下测试,发现当断开redis连接之后重连,便会不断出现如上异常。
可是奇怪的是异常只出现在使用redis管道的地方,其他redis操作一切正常,如下:

Pipeline pipe = jedis.pipelined();
List<Response<String>> responseList = new ArrayList<Response<String>>();
for (int i = 0; i < size; i++) {
     responseList.add(pipe.lpop(key));
}
pipe.sync();

在调用管道的sync方法时,抛出了异常,本着先简单快速解决问题的思路,试着把jedis客户端从2.1升级到2.7,再测试,异常消失了,于是更怀疑这是jedis客户端的bug。
异常是没出现了,可是原因依旧未知,于是继续测试,先贴一下部分测试代码:

 public static void testBatchPop() {

        String key = "test_key";
        // for (int i = 0; i < 10; i++) {
        // JedisUtil.push(key, i + "");
        // }
        List<String> list = null;
        String value;
        while (true) {
            try {
                value = System.currentTimeMillis() + "";
                System.out.println("push:" + value + ", return:"
                        + JedisUtil.push(key, value));
                System.out.println("push:" + value + ", return:"
                        + JedisUtil.push(key, value));

                // batch pop
                try {
                    // String str = JedisUtil.pop(key);
                    // System.out.println("pop:" + str);
                    // 使用redis管道批量lpop
                    list = JedisUtil.batchPop(key, 3);
                    if (list == null || list.isEmpty()) {
                        System.out.println("empty list");
                    } else {
                        for (String str : list) {
                            System.out.println("pop:" + str);
                        }
                    }
                } catch (Exception e) {
                    System.out.println(ExceptionUtil.parseString(e));
                }

                System.out.println("length of " + key + ":"
                        + JedisUtil.llen(key));
            } catch (Exception e) {
                System.out.println(ExceptionUtil.parseString(e));
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println(ExceptionUtil.parseString(e));
            }
        }
    }

代码很好理解,不断往redis队列里面push消息,再批量pop出来消费,断开重连redis之后,batchPop方法抛异常,push方法正常。再贴一下其中使用到redis管道的方法,目光如炬的小伙伴可能已经发现问题所在了:

public static List<String> batchPop(String key, int size) {
        Jedis jedis = pool.getResource();
        try {
            Pipeline pipe = jedis.pipelined();
            List<Response<String>> responseList = new ArrayList<Response<String>>();
            for (int i = 0; i < size; i++) {
                responseList.add(pipe.lpop(key));
            }
            pipe.sync();
            List<String> list = new ArrayList<String>();
            for (Response<String> response : responseList) {
                if (response != null && StringUtils.isNotBlank(response.get())
                        && !NULL_RES.equals(response.get())) {
                    list.add(response.get());
                }
            }
            return list;
        } catch (JedisConnectionException ex) {
            log.error("Jedis Connection Exception");
            returnBrokenResource(pool, jedis);
            throw ex;
        } finally {
                returnResource(pool, jedis);
        }
    }

该方法使用jedis连接池获取连接,每次批量从redis队列中pop消息。看到这里的代码,立刻把Jedis连接池pool列入可疑名单中,猜测会不会是returnBrokenResource的时候,没有把断开的连接给销毁掉。
为了排除是否连接池也有问题,所以将以上代码改为每次操作时,先new一个新的Jedis,而不是使用连接池,结果异常消失了。
然后自己又根据jedis的源代码,使用Apache的Commons Pool 1.5.5写了个简单的连接池(jedis 2.1也是使用了这个版本的Commons Pool),将创建对象,销毁对象每一步的日志都打印出来,并且对jedis进行一层封装,赋予一个ID,这样易于观察每个jedis链接的生命周期,部分代码如下:

    private static class JedisTest extends Jedis {
        private int jedisID;
        public int getJedisID() {
            return jedisID;
        }
        public void setJedisID(int jedisID) {
            this.jedisID = jedisID;
        }
        public JedisTest(String host, int port, int connectionTimeout,
                int soTimeout, int jedisID) {
            super(host, port, connectionTimeout, soTimeout);
            this.jedisID = jedisID;
        }
    }

使用Commons Pool时,需要创建一个工厂类,将jedis连接池源代码中的工厂类拷贝出来使用即可:

    private static int NUM;
    // jedis 2.1
    private static class JedisFactory extends BasePoolableObjectFactory {
        private final String host;
        private final int port;
        private final int timeout;
        private final String password;
        private final int database;

        public JedisFactory(String host, int port, int timeout,
                String password, int database) {
            this.host = host;
            this.port = port;
            this.timeout = timeout;
            this.password = password;
            this.database = database;
        }

        public Object makeObject() throws Exception {
            NUM++;
            JedisTest jedis = new JedisTest(this.host, this.port, this.timeout,NUM);
            System.out.println("makeObject jedisID:" + jedis.getJedisID());
            jedis.connect();
            if (null != this.password) {
                jedis.auth(this.password);
            }
            if (this.database != 0) {
                jedis.select(this.database);
            }

            return jedis;
        }

        public void destroyObject(Object obj) throws Exception {
            if (obj instanceof Jedis) {
                JedisTest jedis = (JedisTest) obj;
                System.out.println("destroyObject jedisID:"
                        + jedis.getJedisID());
                if (!(jedis.isConnected()))
                    return;
                try {
                    try {
                        jedis.quit();
                    } catch (Exception e) {
                    }
                    jedis.disconnect();
                } catch (Exception e) {
                }
            }
        }

        public boolean validateObject(Object obj) {
            boolean res = false;
            if (obj instanceof Jedis) {
                JedisTest jedis = (JedisTest) obj;
                System.out.println("validateObject jedisID:"
                        + jedis.getJedisID());
                try {
                    res = ((jedis.isConnected()) && (jedis.ping()
                            .equals("PONG")));
                } catch (Exception e) {
                    res = false;
                }
            }
            System.out.println("validateObject return:" + res);
            return res;
        }
    }

经测试,确实调用了destroyObject方法,但是每次从连接池中拿到的都是上次销毁的jedis连接。
回头看一下上面的batchPop方法,发现问题所在了,当抛出JedisConnectionException异常时,调用了returnBrokenResource,将连接放回池中销毁,但是最后的finally代码块里又调用了一次returnResource。没错,这里两次将连接放回了连接池里,所以这个broken的jedis根本没有被销毁掉,又被重复拿出来使用了。
但是,真相到这里还远远没有明了,为啥使用管道lpop的时候异常,使用非管道push的时候却是正常的?
深入看一下jedis管道lpop的源代码,如下:

public Response<String> lpop(String key) {
        this.client.lpop(key);
        return getResponse(BuilderFactory.STRING);
    }

对比下非管道时的lpop方法:

public String lpop(String key) {
        checkIsInMulti();
        this.client.lpop(key);
        return this.client.getBulkReply();
    }

发现它们其实都使用了this.client.lpop(key),继续进入这个方法,一直点啊点,直到这里:

protected Connection sendCommand(Protocol.Command cmd, byte[][] args) {
        connect();
        Protocol.sendCommand(this.outputStream, cmd, args);
        this.pipelinedCommands += 1;
        return this;
}

一个重要的方法出现了:connect(),点进去:

    public void connect() {
        if (isConnected())
            return;
        try {
            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.timeout);
            this.socket.setSoTimeout(this.timeout);
            this.outputStream = new RedisOutputStream(
                    this.socket.getOutputStream());
            this.inputStream = new RedisInputStream(
                    this.socket.getInputStream());
        } catch (IOException ex) {
            throw new JedisConnectionException(ex);
        }
    }

没错,所有的redis操作都会调用到sendCommand方法,而该方法会先判断jedis客户端是否连接,如果不是,则重新new一个socket,这就解释了,为什么断开redis之后重连,可以继续往redis队列中push消息,但是,更疑惑的是,既然重新new了socket,为何管道操作还是会报异常。
继续测试,发现虽然 JedisUtil.batchPop抛出了异常,但是redis队列的长度并没有在增加,所以队列里面的消息应该已经被成功pop出来了,而且,当batchPop的数量为1时,程序竟然偶尔会恢复正常。
继续研究pipline的sync代码,发下如下:

    public List<Object> getAll(int except) {
        List all = new ArrayList();
        flush();
        while (this.pipelinedCommands > except) {
            try {
                all.add(Protocol.read(this.inputStream));
            } catch (JedisDataException e) {
                all.add(e);
            }
            this.pipelinedCommands -= 1;
        }
        return all;
    }

其中pipelinedCommands记录了往redis发送的命令的条数(见上面的sendCommand方法),所以调用sync取返回结果时,要循环pipelinedCommands次,从输入流中读取返回结果,程序也是在这里抛出了SocketTimeoutException。对比这里的sync方法与其他非管道操作获取结果的方法,区别便在于其他操作只调用了一次Protocol.read(this.inputStream),而管道要调用pipelinedCommands次。
注意到这点,真相已经隐隐约约出现了,做个假设,如果pipelinedCommands的数量大于所要读取的返回结果,那么这里是否就读不到输入流的结果,然后就一直阻塞直到socket timeout?
重新回过头看一下前面的代码,梳理整个流程,测试过程中,最耗时的其实是在调用sync读取返回结果,所以连接中断最有可能在这个时候发生,抛出异常,然后这个已经broken的jedis又被放回到了连接池中,下一次又被重新从连接池中拿出来,此时调用pipline.sync之前,虽然socket又重新new了一遍,但是pipelinedCommands的值并没有清0啊,因此调用batchPop的时候,pipelinedCommands的数量要大于所要返回的结果数量,因此读取输入流时阻塞,直到抛出socket timeout异常。
测试中,如果批量获取1条记录,就减少了pipelinedCommands大于返回结果数量的概率,所以偶尔可以测试正常!
另外,看一下jedis的配置文件:

redis.pool.testOnBorrow=true
redis.pool.testOnReturn=true
redis.pool.testWhileIdle=true

发现从连接池获取jedis之前都会检查是否已连接,如果获取到的是broken的jedis,应该会重新创建一个才对。仔细打印出redis.pool.testOnBorrow的值,发现被坑了,这里后面多了个空格!空格!空格……WTF!
再接着看一下jedis 2.7,发现其用了commons pool2-2.3,对比Commons Pool 1.5.5的源代码(好累,不贴代码了),在新的版本中,如果一个对象如果已经被销毁,则不能重新放回对象池中,因此每次从连接池中拿到的是新的jedis,所以没有出现上面的异常。
到这里,总结一下,这个问题的发生其实是正好三方的代码都有问题:

  • 业务代码中,重复将已经broken的jedis放回了连接池中;
  • Commons Pool 1.5.5又没有对重复returnObject做限制;
  • jedis重新new socket的时候,也没有清空pipelinedCommands。

在新版本的jedis客户端中,其实已经简化了编程,如下使用即可:

    public static long push(String queueName, String value) {
        Jedis jedis = null;
        try {
            jedis = pool.getResource();
            // 以队列右边为尾,将数据从右边插入
            return jedis.rpush(queueName, value);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }

看一下jedis.close的源代码,如下:

    public void close() {
        if (this.dataSource != null) {
            if (this.client.isBroken())
                this.dataSource.returnBrokenResource(this);
            else
                this.dataSource.returnResource(this);
        } else
            this.client.close();
    }

这里已经根据客户端是否为broken,决定如何返还连接。当发生连接异常时,client.borken将被标志。

回头想一想,在查找这个问题时,也犯了一些错误,如果一开始就可以仔细梳理代码流程,不要盲目测试,也不用花这么多时间。
与君共勉吧!


简单,率性,不虚妄