public class Cluster {

    public static final int DEFAULT_WEIGHT = 100;

    private static final Logger LOG = LoggerFactory.getLogger(Cluster.class);

    private static final SecureRandom RANDOM = new SecureRandom();

    private String id = String.valueOf(hashCode());

    private boolean invalid = false;

    private List<Server> servers;

    private List<Server> masters = new ArrayList<Server>();

    private List<Server> invalidServers = new ArrayList<Server>();

    private int weight = DEFAULT_WEIGHT;

    private Range<Integer> hashRange;   // 该 Cluster 原本的哈希段

    private Cluster takenOverBy;        // 当 Cluster 不可用时，其哈希段会被另一个 Cluster 接管

    public Cluster(List<Server> servers, int weight) {

        servers.removeAll(Collections.singleton((Server) null));
        if (servers.isEmpty()) {
            throw new SsdbClientException("servers is empty");
        }

        this.servers = new ArrayList<Server>(servers);
        this.weight = weight;
        this.id = servers.get(0).getHost() + ":" + servers.get(0).getPort();

        fillMasters();
    }

    public Cluster(Server server, int weight) {
        this(Collections.singletonList(server), weight);
    }

    public Cluster(List<Server> servers) {
        this(servers, DEFAULT_WEIGHT);
    }

    public Cluster(Server server) {
        this(server, DEFAULT_WEIGHT);
    }

    //////////////////////////////////////////////////////////////

    // 将一个 Server 列表转换为一个 Cluster 列表，其中每个 Cluster 包含一个 Server
    public static List<Cluster> toClusters(List<Server> servers) {
        ArrayList<Cluster> clusters = new ArrayList<Cluster>();
        for (Server server : servers) {
            clusters.add(new Cluster(server));
        }
        return clusters;
    }

    public static Cluster fromSingleServer(Server server) {
        return new Cluster(server);
    }

    public static Cluster fromSingleServer(String host, int port) {
        return fromSingleServer(new Server(host, port));
    }

    public static Cluster fromSingleServer(String host, int port, int timeoutSeconds) {
        return fromSingleServer(new Server(host, port, timeoutSeconds));
    }

    public static Cluster fromSingleServer(String host, int port, int timeoutSeconds, int bufferSize) {
        return fromSingleServer(new Server(host, port, timeoutSeconds, bufferSize));
    }

    public static Cluster fromSingleServer(String host, int port, String pass, int timeoutSeconds, int bufferSize) {
        return fromSingleServer(new Server(host, port, pass, true, timeoutSeconds, bufferSize));
    }

    public static Cluster fromSingleServer(String host, int port, String pass) {
        return fromSingleServer(new Server(host, port, pass));
    }

    public static Cluster fromServers(List<Server> servers) {
        return new Cluster(servers);
    }

    public static Cluster fromServers(Server... servers) {
        return new Cluster(Arrays.asList(servers));
    }

    //////////////////////////////////////////////////////////////

    // 将主服务器加入到 masters 列表
    private void fillMasters() {
        for (Server server : servers) {
            if (server.isMaster()) {
                this.masters.add(server);
            }
        }
    }

    public boolean isInvalid() {
        return invalid;
    }

    public void setInvalid(boolean invalid) {
        this.invalid = invalid;
    }

    public Range<Integer> getHashRange() {
        return hashRange;
    }

    public void setHashRange(Range<Integer> hashRange) {
        this.hashRange = hashRange;
    }

    public Cluster getTakenOverBy() {
        return takenOverBy;
    }

    public void setTakenOverBy(Cluster takenOverBy) {
        this.takenOverBy = takenOverBy;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public int getWeight() {
        return weight;
    }

    public void setWeight(int weight) {
        this.weight = weight;
    }

    public List<Server> getServers() {
        return Collections.unmodifiableList(servers);
    }

    public List<Server> getInvalidServers() {
        return Collections.unmodifiableList(invalidServers);
    }

    /**
     * 添加一台服务器
     *
     * @param server 要添加的服务器
     */
    public synchronized void addServer(Server server) {
        if (!servers.contains(server)) {
            servers.add(server);

            if (server.isMaster() && !masters.contains(server)) {
                masters.add(server);
            }
        }
    }

    /**
     * 删除一台服务器
     *
     * @param server 要删除的服务器
     */
    public synchronized void removeServer(Server server) {
        masters.remove(server);
        servers.remove(server);
    }

    /**
     * 获取一台主服务器（用于写入）。所有的服务器中必须至少有一台设置为 master=true
     *
     * @return 获取到的主服务器，如果找不到则抛出异常
     */
    public Server getMaster() {
        if (masters.isEmpty()) {
            throw new SsdbNoServerAvailableException(
                    "Unable to find master server in cluster '" + id + "'");
        }

        if (masters.size() == 1) {
            return masters.get(0);
        }

        return masters.get(RANDOM.nextInt(masters.size()));
    }

    /**
     * 获取一个随机的服务器（用于读取）
     *
     * @return 一个随机的服务器
     */
    public Server getRandomServer() {
        if (servers.isEmpty()) {
            throw new SsdbNoServerAvailableException(
                    "Unable to find server in cluster '" + id + "'");
        }

        if (servers.size() == 1) {
            return servers.get(0);
        }

        return servers.get(RANDOM.nextInt(servers.size()));
    }

    /**
     * 将服务器标记为无效的
     *
     * @param invalid 需要被标记的服务器
     */
    public synchronized void markInvalid(Server invalid) {
        LOG.error("Server unavailable: " + invalid);

        this.servers.remove(invalid);
        if (invalid.isMaster()) {
            this.masters.remove(invalid);
        }

        if (!this.invalidServers.contains(invalid)) {
            this.invalidServers.add(invalid);
        }

        ServerMonitorDaemon.addInvalidServer(invalid, this);
    }

    /**
     * 将服务器标记为有效的
     *
     * @param server 恢复正常的服务器
     */
    public synchronized void markValid(Server server) {
        LOG.info("Server restored: " + server);

        this.servers.add(server);
        if (server.isMaster()) {
            this.masters.add(server);
        }

        this.invalidServers.remove(server);
        this.invalid = false;
    }

    /**
     * 返回指定 hash 所属的 Cluster
     *
     * @param hash 要查询的 hash
     *
     * @return 如果当前 Cluster 的哈希段包含该值，则返回当前 Cluster 或托管的 Cluster，否则返回 null
     */
    public Cluster getHashHostingCluster(int hash) {
        if (this.hashRange.contains(hash)) {
            return thisOrTakenOver();
        } else {
            return null;
        }
    }

    private Cluster thisOrTakenOver() {
        return this.invalid ? this.takenOverBy.thisOrTakenOver() : this;
    }

    @Override
    public String toString() {
        return "Cluster{" +
                "id='" + id + '\'' +
                ", weight=" + weight +
                '}';
    }

    /**
     * 判断 server 是否属于当前 cluster
     *
     * @param server 要判断的 server
     *
     * @return 如果属于当前 cluster 则返回 true
     */
    public synchronized boolean containsServer(Server server) {
        return this.servers.contains(server) || this.invalidServers.contains(server);
    }

    public Server findServer(String host, int port) {
        for (Server server : servers) {
            if (server.getHost().equals(host) && server.getPort() == port) {
                return server;
            }
        }

        return null;
    }
}
 public class Server {

    private String host;            // 服务器地址

    private int port;               // 服务器端口

    private String pass;            // 服务器校验密码（可选）

    private boolean master = true;  // 是否是主服务器。

    private GenericObjectPoolConfig<Connection> poolConfig = createDefaultPoolConfig();     // 连接池配置参数

    private SocketConfig socketConfig = new SocketConfig();     // 网络配置参数

    public Server() {
    }

    public Server(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public Server(String host, int port, boolean master) {
        this.host = host;
        this.port = port;
        this.master = master;
    }

    public Server(String host, int port, String pass) {
        this.host = host;
        this.port = port;
        this.pass = pass;
    }

    public Server(String host, int port, int timeoutSeconds) {
        this.host = host;
        this.port = port;
        this.socketConfig.setSoTimeout(timeoutSeconds * 1000);
    }

    public Server(String host, int port, SocketConfig socketConfig) {
        this.host = host;
        this.port = port;
        this.socketConfig = socketConfig;
    }

    public Server(String host, int port, int timeoutSeconds, int bufferSize) {
        this.host = host;
        this.port = port;
        this.socketConfig.setSoTimeout(timeoutSeconds * 1000);
        this.socketConfig.setSoBufferSize(bufferSize);
    }

    public Server(String host, int port, String pass, boolean master) {
        this.host = host;
        this.port = port;
        this.pass = pass;
        this.master = master;
    }

    public Server(String host, int port, String pass, boolean master, int timeoutSeconds, int poolMaxTotal) {
        this.host = host;
        this.port = port;
        this.pass = pass;
        this.master = master;
        this.socketConfig.setSoTimeout(timeoutSeconds * 1000);
        this.poolConfig.setMaxTotal(poolMaxTotal);
    }

    public Server(String host, int port, String pass, boolean master, SocketConfig socketConfig) {
        this.host = host;
        this.port = port;
        this.pass = pass;
        this.master = master;

        if (socketConfig != null) {
            this.socketConfig = socketConfig;
        }
    }

    public Server(String host, int port, String pass, boolean master,
                  SocketConfig socketConfig, GenericObjectPoolConfig<Connection> poolConfig) {

        this.host = host;
        this.port = port;
        this.pass = pass;
        this.master = master;

        if (socketConfig != null) {
            this.socketConfig = socketConfig;
        }

        if (poolConfig != null) {
            this.poolConfig = poolConfig;
        }
    }


    private GenericObjectPoolConfig<Connection> createDefaultPoolConfig() {
        GenericObjectPoolConfig<Connection> config = new GenericObjectPoolConfig<>();
        config.setMaxIdle(1);
        return config;
    }

    public GenericObjectPoolConfig<Connection> getPoolConfig() {
        return poolConfig;
    }

    public void setPoolConfig(GenericObjectPoolConfig<Connection> poolConfig) {
        this.poolConfig = poolConfig;
    }

    public SocketConfig getSocketConfig() {
        return socketConfig;
    }

    public void setSocketConfig(SocketConfig socketConfig) {
        this.socketConfig = socketConfig;
    }

    public boolean isMaster() {
        return master;
    }

    public void setMaster(boolean master) {
        this.master = master;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getPass() {
        return pass;
    }

    public void setPass(String pass) {
        this.pass = pass;
    }

    @SuppressWarnings("SimplifiableIfStatement")
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof Server)) return false;

        Server server = (Server) o;

        if (getPort() != server.getPort()) return false;
        return getHost().equals(server.getHost());

    }

    @Override
    public int hashCode() {
        int result = getHost().hashCode();
        result = 31 * result + getPort();
        return result;
    }

    @Override
    public String toString() {
        return "Server{" +
                "host='" + host + '\'' +
                ", port=" + port +
                ", master=" + master +
                '}';
    }
} public abstract class Sharding {

    protected List<Cluster> clusters;

    /**
     * 构造方法
     *
     * @param cluster 集群配置，整个负载均衡当中只会有一个集群
     */
    public Sharding(Cluster cluster) {
        this(Collections.singletonList(cluster));
    }

    /**
     * 构造方法
     *
     * @param clusters 集群配置。注意本方法将会使用 clusters 的拷贝，因此调用本方法之后再操作
     *                 clusters（清空或增减元素），不会对 Sharding 有任何影响。
     */
    public Sharding(List<Cluster> clusters) {

        // 清掉可能的 null 元素
        clusters.removeAll(Collections.singleton((Cluster) null));

        if (clusters.isEmpty()) {
            throw new SsdbClientException("clusters is empty");
        }

        this.clusters = new ArrayList<Cluster>(clusters);
    }

    //////////////////////////////////////////////////////////////

    /**
     * 对 {@link #clusters} 进行初始化配置，或执行其他初始化工作
     */
    public abstract void initClusters();

    /**
     * 根据 key 获取所对应的 Cluster
     *
     * @param key 键
     *
     * @return 对应的 Cluster
     */
    public abstract Cluster getClusterByKey(String key);

    /**
     * 当某个 Cluster 下线（即 Cluster 中的所有的服务器都不可用）时的处理（交给子类实现）
     * ServerMonitorDaemon 会检查下线的服务器，发现有服务器恢复运作的话，会修改对应的
     * Cluster 状态为上线
     *
     * @param invalidCluster 下线的 Cluster
     *
     * @return 是否允许选择其他的 Cluster
     */
    public abstract boolean clusterFailed(Cluster invalidCluster);

    //////////////////////////////////////////////////////////////

    public List<Cluster> getClusters() {
        return Collections.unmodifiableList(clusters);
    }

    public Cluster getClusterById(String id) {
        for (Cluster cluster : clusters) {
            if (cluster.getId().equals(id)) {
                return cluster;
            }
        }

        return null;
    }
}
 public class SocketConfig {

    /**
     * 缺省 Socket 超时时间
     */
    public static final int DEFAULT_SO_TIMEOUT = 1000;

    /**
     * 缺省缓冲区大小
     */
    public static final int DEFAULT_SO_BUFFER_SIZE = 8 * 1024;

    //////////////////////////////////////////////////////////////

    private int soTimeout = DEFAULT_SO_TIMEOUT;         // 超时时间 (ms)

    private int soBufferSize = DEFAULT_SO_BUFFER_SIZE;  // 缓存大小 (byte)

    public SocketConfig() {
    }

    public SocketConfig(int soTimeout) {
        this.soTimeout = soTimeout;
    }

    public SocketConfig(int soTimeout, int soBufferSize) {
        this.soTimeout = soTimeout;
        this.soBufferSize = soBufferSize;
    }

    /**
     * 获取 Socket 超时时间
     *
     * @return Socket 超时时间
     */
    public int getSoTimeout() {
        return soTimeout;
    }

    /**
     * 设置 Socket 超时时间
     *
     * @param soTimeout Socket 超时时间
     */
    public void setSoTimeout(int soTimeout) {
        this.soTimeout = soTimeout;
    }


    /**
     * 获取读取数据使用 Buffer 的大小
     *
     * @return Buffer 大小
     */
    public int getSoBufferSize() {
        return soBufferSize;
    }

    /**
     * 设置读取数据使用 Buffer 大小
     *
     * @param soBufferSize Buffer 大小
     */
    public void setSoBufferSize(int soBufferSize) {
        this.soBufferSize = soBufferSize;
    }
} public class Response {

    private Block head;

    private List<Block> body = new ArrayList<>();

    private Charset charset;

    public Response(Charset charset) {
        this.charset = charset;
    }

    public Block getHead() {
        return head;
    }

    public void setHead(Block head) {
        this.head = head;
    }

    public List<Block> getBody() {
        return body;
    }

    public void setBody(List<Block> body) {
        this.body = body;
    }

    public void addBodyBlock(Block block) {
        this.body.add(block);
    }

    public String firstBlock() {
        return this.body.isEmpty() ? null : this.body.get(0).toString();
    }

    public String joinBlocks(char joint) {

        if (this.body.isEmpty()) {
            return "";

        } else if (this.body.size() == 1) {
            return this.body.get(0).toString();

        } else {
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < this.body.size() - 1; i++) {
                sb.append(this.body.get(i)).append(joint);
            }
            sb.append(this.body.get(this.body.size() - 1));
            return sb.toString();
        }
    }

    // 以字节的方式返回数据
    public byte[] getBytes() {
        return this.body.isEmpty() ? new byte[0] : this.body.get(0).getData();
    }

    public Integer getIntResult() {
        if (this.head.toString().equals("not_found")) {
            return null;
        }
        return this.body.isEmpty() ? 0 : Integer.parseInt(this.body.get(0).toString());
    }

    public int getIntResult(int defaultValue) {
        Integer result = getIntResult();
        return result == null ? defaultValue : result;
    }

    public Long getLongResult() {
        if (this.head.toString().equals("not_found")) {
            return null;
        }
        return this.body.isEmpty() ? 0 : Long.parseLong(this.body.get(0).toString());
    }

    public long getLongResult(long defaultValue) {
        Long result = getLongResult();
        return result == null ? defaultValue : result;
    }

    public List<String> getBlocks() {
        ArrayList<String> blocks = new ArrayList<>();
        for (Block block : body) {
            blocks.add(block.toString(charset));
        }
        return blocks;
    }

    public List<KeyValue> getKeyValues() {
        List<KeyValue> keyValues = new ArrayList<>();

        for (int i = 0; i + 1 < body.size(); i += 2) {
            keyValues.add(new KeyValue(
                body.get(i).getData(), body.get(i + 1).getData(), charset
            ));
        }

        return keyValues;
    }

    public List<IdScore> getIdScores() {
        List<IdScore> idScores = new ArrayList<>();

        for (int i = 0; i + 1 < body.size(); i += 2) {
            String key = body.get(i).toString();
            String value = body.get(i + 1).toString();
            idScores.add(new IdScore(key, Long.parseLong(value)));
        }

        return idScores;
    }

    public List<String> getIds() {
        List<String> ids = new ArrayList<>();

        for (int i = 0; i + 1 < body.size(); i += 2) {
            String key = body.get(i).toString();
            ids.add(key);
        }

        return ids;
    }

    public Map<byte[], byte[]> getBlocksAsMap() {

        Map<byte[], byte[]> map = new HashMap<>();
        List<KeyValue> keyValues = getKeyValues();

        for (KeyValue keyValue : keyValues) {
            map.put(keyValue.getKey(), keyValue.getValue());
        }

        return map;
    }

    public Map<String, String> getBlocksAsStringMap(Charset charset) {

        Map<String, String> map = new HashMap<>();
        List<KeyValue> keyValues = getKeyValues();

        for (KeyValue keyValue : keyValues) {
            map.put(
                new String(keyValue.getKey(), charset).intern(),
                new String(keyValue.getValue(), charset).intern()
            );
        }

        return map;
    }
}
 public class ConsistentHashSharding extends Sharding {

    private static final Logger LOG = LoggerFactory.getLogger(ConsistentHashSharding.class);

    /**
     * 单点故障处理策略，参考 {@link SPOFStrategy}
     */
    private SPOFStrategy spofStrategy = SPOFStrategy.AutoExpandStrategy;

    public ConsistentHashSharding(Cluster cluster) {
        super(cluster);
    }

    //////////////////////////////////////////////////////////////

    public ConsistentHashSharding(List<Cluster> clusters) {
        super(clusters);
    }

    public ConsistentHashSharding(Cluster... clusters) {
        super(Arrays.asList(clusters));
    }

    public SPOFStrategy getSpofStrategy() {
        return spofStrategy;
    }

    public void setSpofStrategy(SPOFStrategy spofStrategy) {
        this.spofStrategy = spofStrategy;
    }

    public Map<String, Range<Integer>> getRangeMap() {
        HashMap<String, Range<Integer>> map = new HashMap<String, Range<Integer>>();
        for (Cluster cluster : clusters) {
            map.put(cluster.getId() + "(" + !cluster.isInvalid() + ")",
                cluster.getHashRange().duplicate());
        }
        return map;
    }

    public void removeCluster(String clusterId) {

        Cluster cluster = getClusterById(clusterId);
        if (cluster == null) {
            throw new SsdbNoClusterAvailableException("Cluster " + clusterId + " not found.");
        }

        int minHash = cluster.getHashRange().getMin();
        int maxHash = cluster.getHashRange().getMax();

        // 如果是第一个 Cluster，则将后面的向前扩展；否则令前面的向后扩展
        boolean isFirstCluster = minHash == Integer.MIN_VALUE;

        if (isFirstCluster) {
            Cluster secondCluster = clusters.get(1);
            secondCluster.getHashRange().setMin(minHash);
            LOG.debug("Expand cluster " + secondCluster.getId() + " left to ring start.");
            return;
        } else {
            for (Cluster c : clusters) {
                if (c.getHashRange().getMax() + 1 == minHash) {   // 找到上一个 cluster
                    c.getHashRange().setMax(maxHash);
                    LOG.debug("Expand cluster " + cluster.getId() + " right to " + maxHash);
                    return;
                }
            }
        }

        throw new SsdbClientException("should not be here");
    }

    /**
     * 添加一个 Cluster
     *
     * @param newCluster  要加入的 Cluster
     * @param prevCluster 需要被分担负载的 Cluster
     */
    public synchronized void addCluster(Cluster newCluster, Cluster prevCluster) {

        // check args
        if (clusters.contains(newCluster)) {
            return;
        }

        if (prevCluster == null) {
            throw new IllegalArgumentException("Argument prevCluster cannot be null");
        }

        if (!clusters.contains(prevCluster)) {
            throw new IllegalArgumentException("Argument prevCluster not found");
        }

        // insert into right position
        int prevClusterIndex = clusters.indexOf(prevCluster);
        if (prevClusterIndex == clusters.size() - 1) {
            clusters.add(newCluster);
        } else {
            clusters.add(prevClusterIndex + 1, newCluster);
        }

        // newCluster 和 prevCluster 的哈希段分配仍然依据各自的 weight 权重
        splitRangeToRight(newCluster, prevCluster);
    }

    // 将 toSplitCluster 的右边部分划分给 newCluster
    private void splitRangeToRight(Cluster c2, Cluster c1) {
        Range<Integer> r1 = c1.getHashRange();

        long split = r1.getMin() +
            (r1.getMax().longValue() - r1.getMin().longValue())
                * c1.getWeight() / (c1.getWeight() + c2.getWeight());

        // 更新范围
        setClusterRange(c1, r1.getMin(), (int) split);
        setClusterRange(c2, (int) split + 1, r1.getMax());
    }

    @Override
    public void initClusters() {

        if (clusters.isEmpty()) {
            return;
        } else {
            setClusterIdsIfDefault();
        }

        if (clusters.size() == 1) {
            setClusterRange(clusters.get(0), Integer.MIN_VALUE, Integer.MAX_VALUE);
            return;
        }

        // 计算所有权重总和
        int maxWeight = 0;
        for (Cluster cluster : clusters) {
            maxWeight += cluster.getWeight();
        }

        // 为每个权重分配 hash 段
        long maxrange = (long) Integer.MAX_VALUE - (long) Integer.MIN_VALUE;
        long weightcounter = 0;
        long pointer = 0;

        for (int i = 0; i < clusters.size(); i++) {
            Cluster cluster = clusters.get(i);
            int min, max;

            if (i == clusters.size() - 1) {
                // 如果是最后一个节点，则 max 设为 Integer 的最大值
                min = ((int) (pointer + Integer.MIN_VALUE) + 1);
                max = (Integer.MAX_VALUE);
                setClusterRange(cluster, min, max);

            } else {
                // 非最后一个节点，则
                // 如果是第一个节点，则 min 为 Integer 的最小值，否则 min 为上一个节点最大值 + 1
                min = i == 0 ? Integer.MIN_VALUE : (int) (pointer + Integer.MIN_VALUE) + 1;

                // 非最后节点的 max 值计算：根据 weight 计算节点的哈希段占比（含值个数），加上 min 值即可
                weightcounter += cluster.getWeight();
                pointer = maxrange * weightcounter / maxWeight;
                max = ((int) (pointer + Integer.MIN_VALUE));

                setClusterRange(cluster, min, max);
            }
        }
    }

    private void setClusterIdsIfDefault() {
        for (int i = 0; i < clusters.size(); i++) {
            Cluster cluster = clusters.get(i);
            if (String.valueOf(cluster.hashCode()).equals(cluster.getId())) {
                cluster.setId("Cluster" + i);
            }
        }
    }

    private void setClusterRange(Cluster cluster, int minValue, int maxValue) {
        Range<Integer> range = new Range<Integer>(minValue, maxValue);
        cluster.setHashRange(range);
    }

    @Override
    public synchronized boolean clusterFailed(Cluster invalidCluster) {
        if (invalidCluster == null) {
            return true;
        }

        if (this.spofStrategy == SPOFStrategy.AutoExpandStrategy) {
            invalidCluster.setInvalid(true);
            autoExpand(invalidCluster);
            return true;
        } else {
            // 保留 key 空间，直到 Cluster 恢复上线
            // 在此之前，对该 Cluster 的读写都将失败
            return false;
        }
    }

    /**
     * 自动扩展 key 空间，将 invalidCluster 的 key 空间交给其他 Cluster
     *
     * @param invalidCluster 已下线的 Cluster
     */
    private synchronized void autoExpand(Cluster invalidCluster) {

        if (!clusters.contains(invalidCluster)) {
            return;
        }

        if (noClusterAvailable()) {
            throw new SsdbNoClusterAvailableException("No cluster exists");
        }

        int minHash = invalidCluster.getHashRange().getMin();
        int maxHash = invalidCluster.getHashRange().getMax();

        // 如果是第一个 Cluster，则将后面的向前扩展；否则令前面的向后扩展
        // 扩展本身不会修改 Cluster 的 range，
        boolean isFirstCluster = minHash == Integer.MIN_VALUE;

        if (isFirstCluster) {
            Cluster secondCluster = clusters.get(1);
            invalidCluster.setTakenOverBy(secondCluster);
            LOG.debug("Cluster " + secondCluster.getId() + " takes over " + invalidCluster.getId());
            return;
        } else {
            for (Cluster cluster : clusters) {
                if (cluster.getHashRange().getMax() + 1 == minHash) {   // 找到上一个 cluster
                    invalidCluster.setTakenOverBy(cluster);
                    LOG.debug("Cluster " + cluster.getId() + " takes over " + invalidCluster.getId());
                    return;
                }
            }
        }

        throw new SsdbClientException("should not be here");
    }

    private int getAvailableClusterCount() {
        int n = 0;
        for (Cluster cluster : clusters) {
            if (!cluster.isInvalid()) {
                n += 1;
            }
        }
        return n;
    }

    private boolean noClusterAvailable() {
        for (Cluster cluster : clusters) {
            if (!cluster.isInvalid()) {
                return false;
            }
        }
        return true;
    }

    @Override
    public Cluster getClusterByKey(String key) {

        if (noClusterAvailable()) {
            throw new SsdbNoClusterAvailableException("NO CLUSTER AVAILABLE");
        }

        int hash = MD5.md5Hash(key);

        for (Cluster cluster : clusters) {
            Cluster hostingCluster = cluster.getHashHostingCluster(hash);
            if (hostingCluster != null) {
                return hostingCluster;
            }
        }

        // 理论上 clusters 列表一定会包含 Integer 的所有值，
        // 所以执行到这里表示所有的 Cluster 都不可用
        throw new SsdbException("Unable to choose a cluster for key '" + key + "'");
    }
} public class IdScore {

    private String id;

    private long score;

    public IdScore() {
    }

    public IdScore(String id, long score) {
        this.id = id;
        this.score = score;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public long getScore() {
        return score;
    }

    public void setScore(long score) {
        this.score = score;
    }

    @Override
    public String toString() {
        return "IdScore{" +
                "id='" + id + '\'' +
                ", score=" + score +
                '}';
    }
} public class KeyValue {

    private Charset charset;

    private byte[] key;

    private byte[] value;

    public KeyValue(byte[] key, byte[] value, Charset charset) {
        this.key = key;
        this.value = value;
        this.charset = charset;
    }

    public KeyValue(String key, byte[] value, Charset charset) {
        this((charset == null? key.getBytes(): key.getBytes(charset)), value, charset);
    }

    public KeyValue(String key, String value) {
        this(key, value, AbstractClient.DEFAULT_CHARSET);
    }

    public KeyValue(String key, String value, Charset charset) {
        this(key.getBytes(charset), value.getBytes(charset), charset);
    }

    public byte[] getKey() {
        return key;
    }

    public String getKeyString() {
        return this.charset == null? new String(this.key) :  new String(this.key, this.charset);
    }

    public void setKey(byte[] key) {
        this.key = key;
    }

    public byte[] getValue() {
        return value;
    }

    public String getValueString() {
        return new String(this.value, this.charset);
    }

    public void setValue(byte[] value) {
        this.value = value;
    }
} public class Num {

    public static String ifNull(Long l, String value) {
        return l == null ? value : String.valueOf(l);
    }
}
 public interface Processor<T> {

    void process(T t);
}
 public class Str {

    public static String ifBlank(String... values) {
        for (String value : values) {
            if (!isBlank(value)) {
                return value;
            }
        }

        return "";
    }

    public static boolean isBlank(String s) {
        return s == null || s.trim().length() == 0;
    }
}

import com.hyd.ssdb.conf.Cluster;
import com.hyd.ssdb.conf.Server;
import com.hyd.ssdb.conf.Sharding;
import com.hyd.ssdb.conf.SocketConfig;
import com.hyd.ssdb.protocol.Response;
import com.hyd.ssdb.sharding.ConsistentHashSharding;
import com.hyd.ssdb.util.IdScore;
import com.hyd.ssdb.util.KeyValue;
import com.hyd.ssdb.util.Num;
import com.hyd.ssdb.util.Processor;
import com.hyd.ssdb.util.Str;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
package com.hyd.ssdb;


/**
 * 包含连接池的客户端类，对于一个 SSDB 服务器只需要创建一个 SsdbClient 客户端。
 * <p>
 * 应用关闭时，需要调用 {@link #close()} 方法释放资源。
 *
 * @author Yiding
 */
@SuppressWarnings({"WeakerAccess", "unused", "RedundantCast"})
public class SsdbClient extends AbstractClient {

    /**
     * 构造方法
     *
     * @param host 服务器地址
     * @param port 服务器端口
     *
     * @throws SsdbException 如果连接服务器失败
     */
    public SsdbClient(String host, int port) throws SsdbException {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(host, port)));
    }

    // 创建只连接到一台服务器的 SsdbClient 对象
    public SsdbClient(String host, int port, int timeoutSeconds) throws SsdbException {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(host, port, timeoutSeconds)));
    }

    // 创建只连接到一台服务器的 SsdbClient 对象
    public SsdbClient(String host, int port, int timeoutSeconds, int bufferSize) throws SsdbException {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(host, port, timeoutSeconds, bufferSize)));
    }

    // 创建只连接到一台服务器的，带密码的 SsdbClient 对象
    public SsdbClient(String host, int port, int timeoutSeconds, String pass) throws SsdbException {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(
            host, port, pass, timeoutSeconds, SocketConfig.DEFAULT_SO_BUFFER_SIZE)));
    }

    // 创建只连接到一台服务器的，带密码的 SsdbClient 对象
    public SsdbClient(String host, int port, String pass) throws SsdbException {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(host, port, pass)));
    }

    // 基于一台服务器的配置创建 SsdbClient 对象
    public SsdbClient(Server server) {
        super(new ConsistentHashSharding(Cluster.fromSingleServer(server)));
    }

    // 创建连接到被 Sharding 托管的服务器集群的 SsdbClient 对象
    public SsdbClient(Sharding sharding) {