压力测试下的Redis hooking(发布-订阅)——负载下的性能

Question

基于suggested solusion并关注 example ，我正在尝试在收到另一个 key 已过期的通知后立即删除一个 key 。

问题是，在设置 600K 新 key 并将其中一半的过期时间设置为 2 秒的重负载压力测试下，我得到以下异常:

Exception in thread "main" redis.clients.jedis.exceptions.JedisConnectionException: Unknown reply: t

问题是编写这样的监听器的最佳实践是什么？ (线程池？如果是在什么上下文中实现它？)

绝地武士版本:2.7.2

Redis版本:2.8.19

---

到目前为止我的代码:

订阅者类:

public class Subscriber {

    public static void main(String[] args) {
        JedisPool pool = new JedisPool(new JedisPoolConfig(), "localhost");

        Jedis jedis = pool.getResource();
        jedis.psubscribe(new KeyExpiredListener(), "__key*__:*");

    }

}

监听类:

public class KeyExpiredListener extends JedisPubSub {

    private String generalKeyTimeoutPrefix = "TO_";

    @Override
    public void onPMessage(String pattern, String channel, String message) {
        String originalKey = null;
        try {
            if(channel.endsWith("__keyevent@0__:expired") && message.startsWith(generalKeyTimeoutPrefix)) {
                originalKey = message.substring(generalKeyTimeoutPrefix.length());
                del(originalKey);
            }
        } catch(Exception e) {
            logger.error("..", e);
        }
    }

    private void del(String key) {
        Jedis jedis = new Jedis("localhost");
        jedis.connect();
        try {
            jedis.del(key);
        } catch (Exception e) {
            logger.error("...");
        } finally {
            jedis.disconnect();
        }
    }
}

key 生成器:

public class TestJedis {
    public static void main(String[] args) throws InterruptedException {
        JedisPool pool = new JedisPool(new JedisPoolConfig(), "localhost");
        Jedis jedis = pool.getResource();
        String prefixForlKeys = "token_";
        String prefixForTimeoutKeys = "TO_";
        for (int i = 0; i < 300000; i++) {
            String key = prefixForlKeys + i;
            String timeoutKey = prefixForTimeoutKeys + key;
            jedis.set(key, "some_data");
            jedis.set(timeoutKey, "");
            jedis.expire(timeoutKey, 2);
        }
      System.out.println("Finished to create the keys");  
    }
}

Answer 1

问题出在您对 del() 方法的实现上:它不使用连接池，不重用连接，因此它最终占用了所有可用的本地端口。尝试使用这样的东西:

private void del(String key) {
    Jedis jedis = pool.getResource();
    jedis.connect();
    try {
        jedis.del(key);
    } catch (Exception e) {
        e.printStackTrace(  );
    } finally {
        jedis.close();
    }
}

而不是为每个过期的 key 打开/关闭连接。