記錄是一種精神，是加深理解最好的方式之一樱报。

最近深入研究了Jedis的源碼葬项，對Jedis的使用進(jìn)行深入理解，提筆記錄迹蛤。
曹金桂 cao_jingui@163.com(如有遺漏之處還請指教)
時間：2016年12月06日15:00

概述

分片允許數(shù)據(jù)存放在不同的物理機(jī)器上民珍，以適應(yīng)數(shù)據(jù)量過大的場景，克服單臺機(jī)器內(nèi)存或者磁盤空間的限制盗飒。Redis在3.0之前是沒有集群的嚷量，就算現(xiàn)在3.0推出3.0集群版本，也沒有幾家公司用于生產(chǎn)逆趣。Jedis作為Java的Redis客戶端工具蝶溶，其內(nèi)部也提供了客戶的分片的實現(xiàn)。具體可以參考我的文章《Jedis使用教程完整版》宣渗，講解了Jedis客戶端使用抖所，包括了Jedis的客戶端分片的使用方法。

Jedis客戶端分片存在問題

Jedis的客戶端分片采用一致性hash算法痕囱，實現(xiàn)數(shù)據(jù)的分配存儲田轧。但當(dāng)某一個分片服務(wù)掛了之后，客戶端需要修改服務(wù)的指向才能動態(tài)的切換到當(dāng)前分片的從服務(wù)去鞍恢。其實完全可以采用和JedisSentinel一樣的方法傻粘，利用Jedis的Sentinel構(gòu)建高可用的服務(wù)每窖。

實現(xiàn)

個人參照J(rèn)edisSentinelPool的實現(xiàn)，完成了對構(gòu)建Jedis分片的高可用的修改弦悉。源碼如下：

package redis.clients.jedis;

import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.Hashing;
import redis.clients.util.Pool;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;

public class SentinelShardedJedisPool extends Pool<SentinelShardedJedis> {

    protected Logger log = Logger.getLogger(getClass().getName());

    private final Map<String, JedisShardInfo> shardInfoMap = new ConcurrentHashMap<>(); // key=masterName
    private final Map<String, String> sentinels = new ConcurrentHashMap<>(); // key=host+port value=masterName

    private SentinelShardedJedisFactory factory;
    private Set<MasterListener> masterListeners = new HashSet<>();

    public SentinelShardedJedisPool(Set<SentinelInfo> sentinelInfos, final GenericObjectPoolConfig poolConfig, Hashing algo, Pattern keyTagPattern) {
        initShards(sentinelInfos);
        factory = new SentinelShardedJedisFactory(shardInfoMap.values(), algo, keyTagPattern);
        super.initPool(poolConfig, factory);

    }

    private void initShards(Set<SentinelInfo> sentinelInfos) {
        for (SentinelInfo info : sentinelInfos) {
            JedisShardInfo shardInfo = getJedisSharedInfoFromSentinelInfo(info);
            sentinels.put(shardInfo.getHost() + ":" + shardInfo.getPort(), info.getMasterName());
            shardInfoMap.put(info.getMasterName(), shardInfo);

            for (String sentinel : info.getSentinels()) {
                HostAndPort hap = HostAndPort.parseString(sentinel);
                MasterListener masterListener = new MasterListener(info.getMasterName(), hap.getHost(), hap.getPort());
                masterListeners.add(masterListener);
                masterListener.setDaemon(true);
                masterListener.start();
            }
        }
    }

    private JedisShardInfo getJedisSharedInfoFromSentinelInfo(SentinelInfo info) {
        log.info("Trying to find master from available Sentinels...");
        for (String sentinel : info.getSentinels()) {
            final HostAndPort hap = HostAndPort.parseString(sentinel);
            log.fine("Connecting to Sentinel " + hap);
            Jedis jedis = null;
            try {
                jedis = new Jedis(hap.getHost(), hap.getPort());
                List<String> masterAddr = jedis.sentinelGetMasterAddrByName(info.getMasterName());
                if (masterAddr == null || masterAddr.size() != 2) {
                    log.warning("Can not get master addr, master name: " + info.getMasterName() + ". Sentinel: " + hap + ".");
                    continue;
                }

                String host = masterAddr.get(0);
                int port = Integer.parseInt(masterAddr.get(1));
                JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port);
                log.fine("Found Redis master at " + masterAddr);
                return jedisShardInfo;
            } catch (JedisException e) {
                log.warning("Cannot get master address from sentinel running @ " + hap + ". Reason: " + e + ". Trying next one.");
            } finally {
                if (jedis != null) {
                    jedis.close();
                }
            }
        }
        throw new JedisConnectionException("All sentinels down, cannot determine where is " + info.getMasterName() + " master is running...");
    }

    private void reInitPool() {
        internalPool.clear();
    }

    @Override
    public void destroy() {
        for (MasterListener m : masterListeners) {
            m.shutdown();
        }
        super.destroy();
    }

    @Override
    public SentinelShardedJedis getResource() {
        while (true) {
            SentinelShardedJedis jedis = super.getResource();
            for (Jedis shard : jedis.getAllShards()) {
                String host = shard.getClient().getHost();
                int port = shard.getClient().getPort();
                if (sentinels.containsKey(host + ":" + port)) {
                    jedis.setDataSource(this);
                    return jedis;
                }
            }
            returnBrokenResource(jedis);
        }
    }

    @Override
    protected void returnBrokenResource(final SentinelShardedJedis resource) {
        if (resource != null) {
            returnBrokenResourceObject(resource);
        }
    }

    @Override
    protected void returnResource(final SentinelShardedJedis resource) {
        if (resource != null) {
            resource.resetState();
            returnResourceObject(resource);
        }
    }

    private class MasterListener extends Thread {
        private String masterName;
        private String sentinelHost;
        private int sentinelPort;
        private volatile Jedis j;
        private AtomicBoolean running = new AtomicBoolean(false);

        public MasterListener(String masterName, String host, int port) {
            this.masterName = masterName;
            this.sentinelHost = host;
            this.sentinelPort = port;
            this.setName(String.format("MasterListener-%s-[%s:%d]", masterName, host, port));
        }

        @Override
        public void run() {
            running.set(true);
            while (running.get()) {
                j = new Jedis(sentinelHost, sentinelPort);
                try {
                    if (!running.get()) {
                        break;
                    }
                    j.subscribe(new JedisPubSub() {
                        @Override
                        public void onMessage(String channel, String message) {
                            log.fine("Sentinel " + sentinelHost + ":" + sentinelPort + " published: " + message + ".");
                            String[] switchMasterMsg = message.split(" ");
                            if (switchMasterMsg.length > 3) {
                                if (masterName.equals(switchMasterMsg[0])) {
                                    JedisShardInfo oldInfo = shardInfoMap.get(masterName);
                                    sentinels.remove(oldInfo.getHost() + ":" + oldInfo.getPort());

                                    JedisShardInfo newInfo = new JedisShardInfo(switchMasterMsg[3], switchMasterMsg[4]);
                                    newInfo.setConnectionTimeout(oldInfo.getConnectionTimeout());
                                    newInfo.setPassword(oldInfo.getPassword());
                                    newInfo.setSoTimeout(oldInfo.getSoTimeout());
                                    shardInfoMap.put(masterName, newInfo);
                                    sentinels.put(newInfo.getHost() + ":" + newInfo.getPort(), masterName);
                                    reInitPool();
                                } else {
                                    log.fine("Ignoring message on +switch-master for master name " + switchMasterMsg[0] + ", our master name is " + masterName);
                                }

                            } else {
                                log.severe("Invalid message received on Sentinel " + sentinelHost + ":" + sentinelPort + " on channel +switch-master: " + message);
                            }
                        }
                    }, "+switch-master");
                } catch (JedisConnectionException e) {
                    if (running.get()) {
                        log.log(Level.SEVERE, "Lost connection to Sentinel at " + sentinelHost + ":" + sentinelPort + ". Sleeping 5000ms and retrying.", e);
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e1) {
                            log.log(Level.SEVERE, "Sleep interrupted: ", e1);
                        }
                    } else {
                        log.fine("Unsubscribing from Sentinel at " + sentinelHost + ":" + sentinelPort);
                    }
                } finally {
                    j.close();
                }
            }
        }

        public void shutdown() {
            try {
                log.fine("Shutting down listener on " + sentinelHost + ":" + sentinelPort);
                running.set(false);
                if (j != null) {
                    j.disconnect();
                }
            } catch (Exception e) {
                log.log(Level.SEVERE, "Caught exception while shutting down: ", e);
            }
        }
    }

    /**
     * PoolableObjectFactory custom impl.
     */
    private class SentinelShardedJedisFactory implements PooledObjectFactory<SentinelShardedJedis> {

        private Collection<JedisShardInfo> shards;
        private Hashing algo;
        private Pattern keyTagPattern;

        public SentinelShardedJedisFactory(Collection<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
            this.shards = shards;
            this.algo = algo;
            this.keyTagPattern = keyTagPattern;
        }

        @Override
        public PooledObject<SentinelShardedJedis> makeObject() throws Exception {
            SentinelShardedJedis jedis = new SentinelShardedJedis(new ArrayList<>(shards), algo, keyTagPattern);
            return new DefaultPooledObject<SentinelShardedJedis>(jedis);
        }

        @Override
        public void destroyObject(PooledObject<SentinelShardedJedis> pooledShardedJedis) throws Exception {
            final SentinelShardedJedis shardedJedis = pooledShardedJedis.getObject();
            for (Jedis jedis : shardedJedis.getAllShards()) {
                try {
                    try {
                        jedis.quit();
                    } catch (Exception e) {

                    }
                    jedis.disconnect();
                } catch (Exception e) {

                }
            }
        }

        @Override
        public boolean validateObject(PooledObject<SentinelShardedJedis> pooledShardedJedis) {
            try {
                SentinelShardedJedis jedis = pooledShardedJedis.getObject();
                for (Jedis shard : jedis.getAllShards()) {
                    String host = shard.getClient().getHost();
                    int port = shard.getClient().getPort();
                    if (!shardInfoMap.containsKey(host + port)) {
                        return false;
                    }

                    if (!shard.ping().equals("PONG")) {
                        return false;
                    }
                }
                return true;
            } catch (Exception ex) {
                return false;
            }
        }

        @Override
        public void activateObject(PooledObject<SentinelShardedJedis> p) throws Exception {

        }

        @Override
        public void passivateObject(PooledObject<SentinelShardedJedis> p) throws Exception {

        }
    }

    public static class SentinelInfo {
        private String masterName;
        private String clientName;
        private Set<String> sentinels;

        public SentinelInfo(String masterName, String clientName, Set<String> sentinels) {
            this.masterName = masterName;
            this.clientName = clientName;
            this.sentinels = sentinels;
        }

        public String getMasterName() {
            return masterName;
        }

        public String getClientName() {
            return clientName;
        }

        public Set<String> getSentinels() {
            return sentinels;
        }
    }
}

package redis.clients.jedis;

import redis.clients.util.Hashing;

import java.util.List;
import java.util.regex.Pattern;

public class SentinelShardedJedis extends ShardedJedis {

    protected SentinelShardedJedisPool dataSource = null;

    public SentinelShardedJedis(List<JedisShardInfo> shards) {
        super(shards);
    }

    public SentinelShardedJedis(List<JedisShardInfo> shards, Hashing algo) {
        super(shards, algo);
    }

    public SentinelShardedJedis(List<JedisShardInfo> shards, Pattern keyTagPattern) {
        super(shards, keyTagPattern);
    }

    public SentinelShardedJedis(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
        super(shards, algo, keyTagPattern);
    }

    public void setDataSource(SentinelShardedJedisPool sentinelShardedJedisPool) {
        this.dataSource = sentinelShardedJedisPool;
    }

}

使用方法

JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMinIdle(4);
poolConfig.setMaxTotal(100);
poolConfig.setMaxWaitMillis(30000);

Set<String> sentinels = new HashSet<>();
sentinels.add("192.168.2.122:26379");
sentinels.add("192.168.2.128:26379");
Set<SentinelShardedJedisPool.SentinelInfo> sentinelInfos = new HashSet<>();
sentinelInfos.add(new SentinelShardedJedisPool.SentinelInfo("mymaster1", "m1", sentinels));
sentinelInfos.add(new SentinelShardedJedisPool.SentinelInfo("mymaster2", "m2", sentinels));
SentinelShardedJedisPool pool = new SentinelShardedJedisPool(sentinelInfos, poolConfig, Hashing.MURMUR_HASH, null);
SentinelShardedJedis jedis = pool.getResource();
jedis.set("key", "value");
jedis.close();
pool.close();

小結(jié)

本實現(xiàn)個人已經(jīng)測試通過窒典。但若采用本方案必須要經(jīng)過一定的性能測試。如果問題歡迎一起討論警绩。