Jedis を使って redis-rb の Redis::Distributed 互換の ConsistentHashing を使用するための ShardedJedis 継承クラス。

RubyCompatibleShardedJedis.java

package examples.redis;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.regex.Pattern;
import java.util.zip.CRC32;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.util.Hashing;
import redis.clients.util.SafeEncoder;
import redis.clients.util.ShardInfo;

public class RubyCompatibleShardedJedis extends ShardedJedis {
    
    // redis-rb に合わせて CRC32 を使う Hashing 実装クラス
    public static final Hashing CRC32 = new Hashing() {
        @Override
        public long hash(String key) {
            return hash(SafeEncoder.encode(key));
        }

        @Override
        public long hash(byte[] key) {
            CRC32 crc = new CRC32();
            crc.update(key);
            return crc.getValue();
        }
    };
    
    private TreeMap<Long, JedisShardInfo> nodes;
    private final Hashing algo;
    private final Map<ShardInfo<Jedis>, Jedis> resources = new LinkedHashMap<ShardInfo<Jedis>, Jedis>();
    private Pattern tagPattern = null;
    private int db = 0;
    
    public RubyCompatibleShardedJedis(List<JedisShardInfo> shards, Hashing algo,
            Pattern keyTagPattern) {
        this(shards, algo);     
        this.tagPattern = keyTagPattern;
    }

    public RubyCompatibleShardedJedis(List<JedisShardInfo> shards, Hashing algo, int db) {
        super(shards);
        this.algo = algo;
        this.db = db;
        this.initialize(shards);
    }
    
    public RubyCompatibleShardedJedis(List<JedisShardInfo> shards, Hashing algo) {
        super(shards);
        this.algo = algo;
        this.initialize(shards);
    }

    public RubyCompatibleShardedJedis(List<JedisShardInfo> shards, Pattern keyTagPattern) {
        this(shards);
        this.tagPattern = keyTagPattern;
    }

    public RubyCompatibleShardedJedis(List<JedisShardInfo> shards) {
        this(shards, CRC32);
    }

    private void initialize(List<JedisShardInfo> shards) {
        nodes = new TreeMap<Long, JedisShardInfo>();
        for (int i = 0; i != shards.size(); ++i) {
            final JedisShardInfo shardInfo = shards.get(i);
            if (shardInfo.getName() == null) {
              for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
            		nodes.put(this.algo.hash(String.format("redis://%s:%s/%s:%s", shardInfo.getHost(), shardInfo.getPort(), this.db, n)), shardInfo);
            	}
            } else {
            	for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
            		nodes.put(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n), shardInfo);
            	}
            }
            resources.put(shardInfo, shardInfo.createResource());
        }
    }
    
    public JedisShardInfo getShardInfo(byte[] key) {
        SortedMap<Long, JedisShardInfo> head = nodes.headMap(algo.hash(key));
        if (head.size() == 0) {
            return nodes.get(nodes.lastKey());
        }
        return head.get(head.lastKey());
    }
}