Jedis使用总结【pipeline】【分布式的id生成器】【分布式锁【watch】【multi】】【redis分布式】

叁歲伎倆 2022-08-04 12:53 47阅读 0赞

一、 Pipeline

官方的说明是：starts a pipeline,which is a very efficient way to send lots of command and read all the responses when you finish sending them。简单点说pipeline适用于批处理。当有大量的操作需要一次性执行的时候，可以用管道。

示例：

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  Jedis jedis = **new** Jedis(String, **int**);  
2.  Pipeline p = jedis.pipelined();  
3.  p.set(key,value);//每个操作都发送请求给redis-server  
4.  p.get(key,value);  
5.    
6.  p.sync();//这段代码获取所有的response

这里我进行了20w次连续操作（10w读，10w写），不用pipeline耗时：187242ms，用pipeline耗时：1188ms，可见使用管道后的性能上了一个台阶。看了代码了解到，管道通过一次性写入请求，然后一次性读取响应。也就是说jedis是：request response，request response，...；pipeline则是：request request... response response的方式。这样无需每次请求都等待server端的响应。

二、 跨jvm的id生成器   
谈到这个话题，首先要知道redis-server端是单线程来处理client端的请求的。  
这样来实现一个id生成器就非常简单了，只要简单的调用jdeis.incr(key);就搞定了。  
你或许会问，incr是原子操作吗，能保证不会出现并发问题吗，前面说过，server端是单线程处理请求的。

三、 【跨jvm的锁实现【watch】【multi】】  
首先说下这个问题的使用场景，有些时候我们业务逻辑是在不同的jvm进程甚至是不同的物理机上的jvm处理的。这样如何来实现不同jvm上的同步问题呢，其实我们可以基于redis来实现一个锁。

暂时找到三种实现方式：  
1. 通过jedis.setnx(key,value)实现

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.   **import** java.util.Random;  
2.  **import** org.apache.commons.pool.impl.GenericObjectPool.Config;  
3.    
4.  **import** redis.clients.jedis.Jedis;  
5.  **import** redis.clients.jedis.JedisPool;  
6.  **import** redis.clients.jedis.Transaction;  
7.    
8.  /\*\* 
9.   \* @author Teaey 
10.  \*/  
11. **public** **class** RedisLock \{  
12.     //加锁标志  
13.     **public** **static** **final** String LOCKED = "TRUE";  
14.     **public** **static** **final** **long** ONE\_MILLI\_NANOS = 1000000L;  
15.     //默认超时时间（毫秒）  
16.     **public** **static** **final** **long** DEFAULT\_TIME\_OUT = 3000;  
17.     **public** **static** JedisPool pool;  
18.     **public** **static** **final** Random r = **new** Random();  
19.     //锁的超时时间（秒），过期删除  
20.     **public** **static** **final** **int** EXPIRE = 5 \* 60;  
21.     **static** \{  
22.         pool = **new** JedisPool(**new** Config(), "host", 6379);  
23.     \}  
24.     **private** Jedis jedis;  
25.     **private** String key;  
26.     //锁状态标志  
27.     **private** **boolean** locked = **false**;  
28.   
29.     **public** RedisLock(String key) \{  
30.         **this**.key = key;  
31.         **this**.jedis = pool.getResource();  
32.     \}  
33.   
34.     **public** **boolean** lock(**long** timeout) \{  
35.         **long** nano = System.nanoTime();  
36.         timeout \*= ONE\_MILLI\_NANOS;  
37.         **try** \{  
38.             **while** ((System.nanoTime() - nano) < timeout) \{  
39.                 **if** (jedis.setnx(key, LOCKED) == 1) \{  
40.                     jedis.expire(key, EXPIRE);  
41.                     locked = **true**;  
42.                     **return** locked;  
43.                 \}  
44.                 // 短暂休眠，nano避免出现活锁  
45.                 Thread.sleep(3, r.nextInt(500));  
46.             \}  
47.         \} **catch** (Exception e) \{  
48.         \}  
49.         **return** **false**;  
50.     \}  
51.     **public** **boolean** lock() \{  
52.         **return** lock(DEFAULT\_TIME\_OUT);  
53.     \}  
54.   
55.     // 无论是否加锁成功，必须调用  
56.     **public** **void** unlock() \{  
57.         **try** \{  
58.             **if** (locked)  
59.                 jedis.del(key);  
60.         \} **finally** \{  
61.             pool.returnResource(jedis);  
62.         \}  
63.     \}  
64. \}  
65. 2. 通过事务(multi)实现  
66. 由于采纳第一张方法，第二种跟第三种实现只贴了关键代码，望谅解。^\_^  
67.      **public** **boolean** lock\_2(**long** timeout) \{  
68.         **long** nano = System.nanoTime();  
69.         timeout \*= ONE\_MILLI\_NANOS;  
70.         **try** \{  
71.             **while** ((System.nanoTime() - nano) < timeout) \{  
72.                 Transaction t = jedis.multi();  
73.                 // 开启事务，当server端收到multi指令  
74.                 // 会将该client的命令放入一个队列，然后依次执行，知道收到exec指令  
75.                 t.getSet(key, LOCKED);  
76.                 t.expire(key, EXPIRE);  
77.                 String ret = (String) t.exec().get(0);  
78.                 **if** (ret == **null** || ret.equals("UNLOCK")) \{  
79.                     **return** **true**;  
80.                 \}  
81.                 // 短暂休眠，nano避免出现活锁  
82.                 Thread.sleep(3, r.nextInt(500));  
83.             \}  
84.         \} **catch** (Exception e) \{  
85.         \}  
86.         **return** **false**;  
87.     \}

3. 通过事务+监听实现

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **public** **boolean** lock\_3(**long** timeout) \{  
2.    **long** nano = System.nanoTime();  
3.          timeout \*= ONE\_MILLI\_NANOS;  
4.          **try** \{  
5.              **while** ((System.nanoTime() - nano) < timeout) \{  
6.                  jedis.watch(key);  
7.                  // 开启watch之后，如果key的值被修改，则事务失败，exec方法返回null  
8.                  String value = jedis.get(key);  
9.                  **if** (value == **null** || value.equals("UNLOCK")) \{  
10.                     Transaction t = jedis.multi();  
11.                     t.setex(key, EXPIRE, LOCKED);  
12.                     **if** (t.exec() != **null**) \{  
13.                         **return** **true**;  
14.                     \}  
15.                 \}  
16.                 jedis.unwatch();  
17.                 // 短暂休眠，nano避免出现活锁  
18.                 Thread.sleep(3, r.nextInt(500));  
19.             \}  
20.         \} **catch** (Exception e) \{  
21.         \}  
22.         **return** **false**;  
23.     \}

最终采用第一种实现，因为加锁只需发送一个请求，效率最高。

四、 【redis分布式】  
    最后一个话题，jedis的分布式。在jedis的源码里发现了两种hash算法（MD5，MURMUR Hash(默认）），也可以自己实现redis.clients.util.Hashing接口扩展。

List<JedisShardInfo> hosts = new ArrayList<JedisShardInfo>();

**\[java\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  //server1  
2.  JedisShardInfo host1 = **new** JedisShardInfo("", 6380, 2000);  
3.  //server2  
4.  JedisShardInfo host2 = **new** JedisShardInfo("", 6381, 2000);  
5.  hosts.add(host1);  
6.  hosts.add(host2);  
7.  ShardedJedis jedis = **new** ShardedJedis(hosts);  
8.  jedis.set("key", "");

[view plain]: http://blog.csdn.net/woshisap/article/details/45293359#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/653633
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/653633/fork