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

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

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

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 节点:

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

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()