RedisCluster 的 Java 和 Python 客户端 API 使用示例

4etfeww

这里的环境仍然采用 [Redis Cluster 的安装与配置] 中最终的集群配置环境，具体如下：

1 2 3 4 5 6 7 8 9 10

127.0.0.1:7000> cluster nodes 8868592d98d84b7cf5752cc0b97af4ac807d1a12 127.0.0.1:7007 slave bfc910f924d772fe03d9fe6a19aabd73d5730d26 0 1410882113063 8 connected f5bdda1518cd3826100a30f5953ed82a5861ed48 127.0.0.1:7002 slave bfc910f924d772fe03d9fe6a19aabd73d5730d26 0 1410882111659 8 connected 82578e8ec9747e46cbb4b8cc2484c71b9b2c91f4 127.0.0.1:7001 master - 0 1410882111158 2 connected 6461-10922 61dfb1055760d5dcf6519e35435d60dc5b207940 127.0.0.1:7004 slave 82578e8ec9747e46cbb4b8cc2484c71b9b2c91f4 0 1410882112662 5 connected 6d1ebedad33bb31ffbaa99bad095eef4a5920857 127.0.0.1:7006 master - 0 1410882111158 0 connected bfc910f924d772fe03d9fe6a19aabd73d5730d26 127.0.0.1:7005 master - 0 1410882111659 8 connected 11923-16383 35e0f6fdadbf81a00a1d6d1843698613e653867b 127.0.0.1:7003 slave 123ed65d59ff22370f2f09546f410d31207789f6 0 1410882110657 7 connected 123ed65d59ff22370f2f09546f410d31207789f6 127.0.0.1:7000 myself,master - 0 0 7 connected 0-6460 10923-11922 127.0.0.1:7000>

下面分别使用 Jedis 和 redis-py-cluster 来访问 cluster，操作主要有：set/get, pub/sub, 以及获取集群信息。

1，Jedis

    从目前来看对于 redis 以及 RedisCluster 支持最好的应该只有 Jedis 这个开源项目了（也是Redis官网推荐的 Java 库），参考: https://github.com/xetorthio/jedis

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83

import redis.clients.jedis.*; import java.util.HashSet; import java.util.Iterator; import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * jedis 测试 cluster * * @author steven */ public class App {     public static void main( String[] args )     {         Set<HostAndPort> clusterNodes = new HashSet<HostAndPort>();         // 这里只需要列出集群中的一个节点         // JedisCluster 会自己去 discovery 其他的集群节点         clusterNodes.add(new HostAndPort("127.0.0.1", 7000));         JedisCluster cluster = new JedisCluster(clusterNodes);         // set/get         cluster.set("foo", "jedis test");         String value = cluster.get("foo");         System.out.println("foo = " + value);         // get cluster nodes         System.out.println("------- cluster nodes --------");         Map<String, JedisPool> nodes = cluster.getClusterNodes();         Iterator<Map.Entry<String, JedisPool>> iterNodes = nodes.entrySet().iterator();         while (iterNodes.hasNext()) {             Map.Entry<String, JedisPool> entry = iterNodes.next();             Jedis jedis = entry.getValue().getResource();             System.out.println("============");             System.out.println(entry.getKey() + "\n" + jedis.info());         }         // pub/sub         System.out.println("------- pub/sub --------");         // 这里随机取了两个分别用于 publish 和 subscribe 的 jedis 连接         // Redis 不支持在同一个连接上既作为 publisher 又作为 subscriber         final JedisPool jedisPool = nodes.entrySet().iterator().next().getValue();         // 使用一个独立的线程来 publish         ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);         newFixedThreadPool.submit(new Runnable() {             @Override             public void run() {                 try {                     Thread.sleep(2000);                 } catch (InterruptedException e) {                     e.printStackTrace();                     return ;                 }                 jedisPool.getResource().publish("channel-test", "hello, redis cluster!");             }         });         // subscribe - 此处会一直阻塞来接收 publish 消息         jedisPool.getResource().subscribe(new JedisPubSub() {             @Override             public void onMessage(String channel, String message) {                 System.out.println("onMessage - " + channel + ":" + message);             }             @Override             public void onPMessage(String pattern, String channel, String message) {                 System.out.println("onPMessage - " + pattern + "|" + channel + ":" + message);             }             @Override             public void onSubscribe(String channel, int subscribedChannels) {                 System.out.println("onSubscribe - " + channel + ":" + subscribedChannels);             }             @Override             public void onUnsubscribe(String channel, int subscribedChannels) {                 System.out.println("onUnsubscribe - " + channel + ":" + subscribedChannels);             }             @Override             public void onPUnsubscribe(String pattern, int subscribedChannels) {                 System.out.println("onPUnsubscribe - " + pattern + ":" + subscribedChannels);             }             @Override             public void onPSubscribe(String pattern, int subscribedChannels) {                 System.out.println("onPSubscribe - " + pattern + ":" + subscribedChannels);             }         }, "channel-test");     } }

下面为 key 为 foo 的记录写入到了 7005 节点:

1 2 3 4 5 6

127.0.0.1:7000> get foo -> Redirected to slot [12182] located at 127.0.0.1:7005 "jedis test" 127.0.0.1:7005> keys foo 1) "foo" 127.0.0.1:7005>

2，redis-py-cluster

    python 的 redis cluster 库采用的是 redis-py-cluster(https://github.com/Grokzen/redis-py-cluster)

    下面代码只是演示使用 redis-py-cluster 的常见操作示例

    如果需要查看 redis-py-cluster 的完整测试可以直接参考: https://github.com/Grokzen/redis-py-cluster/tree/master/tests

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

import redis from rediscluster import RedisCluster       def main():   startup_nodes = [{"host": "127.0.0.1", "port": "7000"}]   rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)       print '-------- get/set ---------'   if not rc.set("foo", "hello, redis-py-cluster"):     print 'set foo to cluster failed!!!'     return -1   print 'foo = %s' % (rc.get("foo"),)        if __name__ == '__main__':   main()