Redis学习之三 —— 基于Redis Sentinel的Redis集群(主从&分片)高可用客户端方案

以你之姓@ 2022-03-02 02:22 364阅读 0赞

## Redis集群（主从、分片）概念 ##

之前讲过一主多从的哨兵（Sentinel）模式，只有一个主，称为单实例的Redis，现在讲的是多主多从方案。

现在公司用到的就是 多个一主一从组成的Redis集群，通过 Sentinel监控多个主从，多个主就称为Redis数据分片，数据被分不到多个实例上，每个实例都是主从结构，以下是Java客户端怎么用这种集群方案。

## Sentinel出现的问题 ##

Sentinel&Jedis看上去是个完美的解决方案，这句话只说对了一半，在无分片的情况是这样，但我们的应用使用了数据分片-sharing，数据被平均分布到4个不同的实例上，每个实例以主从结构部署，Jedis没有提供基于Sentinel的ShardedJedisPool，也就是说在4个分片中，如果其中一个分片发生主从切换，应用所使用的ShardedJedisPool无法获得通知，所有对那个分片的操作将会失败。   
本文提供一个基于Sentinel的ShardedJedisPool，能及时感知所有分片主从切换行为，进行连接池重建，源码见[https://github.com/warmbreeze/sharded-jedis-sentinel-pool/blob/master/src/main/java/redis/clients/jedis/ShardedJedisSentinelPool.java][https_github.com_warmbreeze_sharded-jedis-sentinel-pool_blob_master_src_main_java_redis_clients_jedis_ShardedJedisSentinelPool.java]

## ShardedJedisSentinelPool实现分析 ##

类似之前的Jedis Pool的构造方法，需要参数poolConfig提供诸如maxIdle，maxTotal之类的配置，masters是一个List，用来保存所有分片Master在Sentinel中配置的名字（注意master的顺序不能改变，因为Shard算法是依据分片位置进行计算，如果顺序错误将导致数据存储混乱），sentinels是一个Set，其中存放所有Sentinel的地址(格式：IP:PORT，如127.0.0.1:26379)，顺序无关；

public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
                                        final GenericObjectPoolConfig poolConfig, int timeout,
                                        final String password, final int database) {
            this.poolConfig = poolConfig;
            this.timeout = timeout;
            this.password = password;
            this.database = database;
    
            List<HostAndPort> masterList = initSentinels(sentinels, masters);
            initPool(masterList);
        }

## **从sentinel集群获取所有分片的master地址** ##

**循环所有分片，循环所有sentinel，获取master地址，给initPool**

private List<HostAndPort> initSentinels(Set<String> sentinels, final List<String> masters) {
    
            Map<String, HostAndPort> masterMap = new HashMap<String, HostAndPort>();
            List<HostAndPort> shardMasters = new ArrayList<HostAndPort>();
    
            log.info("Trying to find all master from available Sentinels...");
    
            for (String masterName : masters) {
                HostAndPort master = null;
                boolean fetched = false;
    
                while (!fetched && sentinelRetry < MAX_RETRY_SENTINEL) {
                    for (String sentinel : sentinels) {
                        final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
    
                        log.info("Connecting to Sentinel " + hap);
    
                        try {
                            Jedis jedis = new Jedis(hap.getHost(), hap.getPort());
                            master = masterMap.get(masterName);
                            if (master == null) {
                                List<String> hostAndPort = jedis.sentinelGetMasterAddrByName(masterName);
                                if (hostAndPort != null && hostAndPort.size() > 0) {
                                    master = toHostAndPort(hostAndPort);
                                    log.info("Found Redis master at " + master);
                                    shardMasters.add(master);
                                    masterMap.put(masterName, master);
                                    fetched = true;
                                    jedis.disconnect();
                                    break;
                                }
                            }
                        } catch (JedisConnectionException e) {
                            log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
                        }
                    }
    
                    if (null == master) {
                        try {
                            log.warn("All sentinels down, cannot determine where is "
                                    + masterName + " master is running... sleeping 1000ms, Will try again.");
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        fetched = false;
                        sentinelRetry++;
                    }
                }
    
                // Try MAX_RETRY_SENTINEL times.
                if (!fetched && sentinelRetry >= MAX_RETRY_SENTINEL) {
                    log.warn("All sentinels down and try " + MAX_RETRY_SENTINEL + " times, Abort.");
                    throw new JedisConnectionException("Cannot connect all sentinels, Abort.");
                }
            }
    
            // All shards master must been accessed.
            if (masters.size() != 0 && masters.size() == shardMasters.size()) {
    
                log.info("Starting Sentinel listeners...");
                for (String sentinel : sentinels) {
                    final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
                    MasterListener masterListener = new MasterListener(masters, hap.getHost(), hap.getPort());
                    masterListeners.add(masterListener);
                    masterListener.start();
                }
            }
    
            return shardMasters;
        }

private void initPool(List<HostAndPort> masters) {
            if (!equals(currentHostMasters, masters)) {
                StringBuffer sb = new StringBuffer();
                for (HostAndPort master : masters) {
                    sb.append(master.toString());
                    sb.append(" ");
                }
                log.info("Created ShardedJedisPool to master at [" + sb.toString() + "]");
                List<JedisShardInfo> shardMasters = makeShardInfoList(masters);
                // 这个initPool是redis.clients.util.Pool的方法
                initPool(poolConfig, new ShardedJedisFactory(shardMasters, Hashing.MURMUR_HASH, null));
                currentHostMasters = masters;
            }
        }

这样就完成了连接池初始化了。

### 监控每个Sentinel，当出现主从切换时，重新初始化连接池 ###

在方法  
![e6f32bb5-f983-3ffe-97f2-a7fa7ce9d26f.png][]  
 最后，会为每个Sentinel启动一个Thread来监控Sentinel做出的更改：   
![1507c655-a4e7-33c5-bb22-38edb76fb1d5.png][]  
该线程的run方法通过Jedis Pub/Sub API（实现JedisPubSub接口，并通过jedis.subscribe进行订阅）向Sentinel实例订阅**“+switch-master”**频道，当Sentinel进行主从切换时，该线程会得到新Master地址的通知，通过master name判断哪个分片进行了切换，将新master地址替换原来位置的地址，并调用initPool(List masters)进行**Jedis连接池重建**；后续所有通过该连接池取得的连接都指向新Master地址，对应用程序透明；

全部代码：

package com.artisan.redis.pool;
    
    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 org.apache.log4j.Logger;
    import redis.clients.jedis.*;
    import redis.clients.jedis.exceptions.JedisConnectionException;
    import redis.clients.util.Hashing;
    import redis.clients.util.Pool;
    
    import java.util.*;
    import java.util.concurrent.atomic.AtomicBoolean;
    import java.util.regex.Pattern;
    
    /**
     * Created by 18121254 on 2019/3/19.
     */
    public class ShardedJedisSentinelPool extends Pool<ShardedJedis> {
        public static final int MAX_RETRY_SENTINEL = 10;
    
        protected final Logger log = Logger.getLogger(getClass().getName());
    
        protected GenericObjectPoolConfig poolConfig;
    
        protected int timeout = Protocol.DEFAULT_TIMEOUT;
    
        private int sentinelRetry = 0;
    
        protected String password;
    
        protected int database = Protocol.DEFAULT_DATABASE;
    
        protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();
    
        private volatile List<HostAndPort> currentHostMasters;
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels) {
            this(masters, sentinels, new GenericObjectPoolConfig(),
                    Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
        }
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels, String password) {
            this(masters, sentinels, new GenericObjectPoolConfig(),
                    Protocol.DEFAULT_TIMEOUT, password);
        }
    
        public ShardedJedisSentinelPool(final GenericObjectPoolConfig poolConfig, List<String> masters, Set<String> sentinels) {
            this(masters, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null,
                    Protocol.DEFAULT_DATABASE);
        }
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
                                        final GenericObjectPoolConfig poolConfig, int timeout,
                                        final String password) {
            this(masters, sentinels, poolConfig, timeout, password,
                    Protocol.DEFAULT_DATABASE);
        }
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
                                        final GenericObjectPoolConfig poolConfig, final int timeout) {
            this(masters, sentinels, poolConfig, timeout, null,
                    Protocol.DEFAULT_DATABASE);
        }
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
                                        final GenericObjectPoolConfig poolConfig, final String password) {
            this(masters, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT,
                    password);
        }
    
        public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
                                        final GenericObjectPoolConfig poolConfig, int timeout,
                                        final String password, final int database) {
            this.poolConfig = poolConfig;
            this.timeout = timeout;
            this.password = password;
            this.database = database;
    
            List<HostAndPort> masterList = initSentinels(sentinels, masters);
            initPool(masterList);
        }
    
        public void destroy() {
            for (MasterListener m : masterListeners) {
                m.shutdown();
            }
    
            super.destroy();
        }
    
        public List<HostAndPort> getCurrentHostMaster() {
            return currentHostMasters;
        }
    
        private void initPool(List<HostAndPort> masters) {
            if (!equals(currentHostMasters, masters)) {
                StringBuffer sb = new StringBuffer();
                for (HostAndPort master : masters) {
                    sb.append(master.toString());
                    sb.append(" ");
                }
                log.info("Created ShardedJedisPool to master at [" + sb.toString() + "]");
                List<JedisShardInfo> shardMasters = makeShardInfoList(masters);
                initPool(poolConfig, new ShardedJedisFactory(shardMasters, Hashing.MURMUR_HASH, null));
                currentHostMasters = masters;
            }
        }
    
        private boolean equals(List<HostAndPort> currentShardMasters, List<HostAndPort> shardMasters) {
            if (currentShardMasters != null && shardMasters != null) {
                if (currentShardMasters.size() == shardMasters.size()) {
                    for (int i = 0; i < currentShardMasters.size(); i++) {
                        if (!currentShardMasters.get(i).equals(shardMasters.get(i))) return false;
                    }
                    return true;
                }
            }
            return false;
        }
    
        private List<JedisShardInfo> makeShardInfoList(List<HostAndPort> masters) {
            List<JedisShardInfo> shardMasters = new ArrayList<JedisShardInfo>();
            for (HostAndPort master : masters) {
                JedisShardInfo jedisShardInfo = new JedisShardInfo(master.getHost(), master.getPort(), timeout);
                jedisShardInfo.setPassword(password);
    
                shardMasters.add(jedisShardInfo);
            }
            return shardMasters;
        }
    
        private List<HostAndPort> initSentinels(Set<String> sentinels, final List<String> masters) {
    
            Map<String, HostAndPort> masterMap = new HashMap<String, HostAndPort>();
            List<HostAndPort> shardMasters = new ArrayList<HostAndPort>();
    
            log.info("Trying to find all master from available Sentinels...");
    
            for (String masterName : masters) {
                HostAndPort master = null;
                boolean fetched = false;
    
                while (!fetched && sentinelRetry < MAX_RETRY_SENTINEL) {
                    for (String sentinel : sentinels) {
                        final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
    
                        log.info("Connecting to Sentinel " + hap);
    
                        try {
                            Jedis jedis = new Jedis(hap.getHost(), hap.getPort());
                            master = masterMap.get(masterName);
                            if (master == null) {
                                List<String> hostAndPort = jedis.sentinelGetMasterAddrByName(masterName);
                                if (hostAndPort != null && hostAndPort.size() > 0) {
                                    master = toHostAndPort(hostAndPort);
                                    log.info("Found Redis master at " + master);
                                    shardMasters.add(master);
                                    masterMap.put(masterName, master);
                                    fetched = true;
                                    jedis.disconnect();
                                    break;
                                }
                            }
                        } catch (JedisConnectionException e) {
                            log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
                        }
                    }
    
                    if (null == master) {
                        try {
                            log.warn("All sentinels down, cannot determine where is "
                                    + masterName + " master is running... sleeping 1000ms, Will try again.");
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        fetched = false;
                        sentinelRetry++;
                    }
                }
    
                // Try MAX_RETRY_SENTINEL times.
                if (!fetched && sentinelRetry >= MAX_RETRY_SENTINEL) {
                    log.warn("All sentinels down and try " + MAX_RETRY_SENTINEL + " times, Abort.");
                    throw new JedisConnectionException("Cannot connect all sentinels, Abort.");
                }
            }
    
            // All shards master must been accessed.
            if (masters.size() != 0 && masters.size() == shardMasters.size()) {
    
                log.info("Starting Sentinel listeners...");
                for (String sentinel : sentinels) {
                    final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
                    MasterListener masterListener = new MasterListener(masters, hap.getHost(), hap.getPort());
                    masterListeners.add(masterListener);
                    masterListener.start();
                }
            }
    
            return shardMasters;
        }
    
        private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
            String host = getMasterAddrByNameResult.get(0);
            int port = Integer.parseInt(getMasterAddrByNameResult.get(1));
    
            return new HostAndPort(host, port);
        }
    
        /**
         * PoolableObjectFactory custom impl.
         */
        protected static class ShardedJedisFactory implements PooledObjectFactory<ShardedJedis> {
            private List<JedisShardInfo> shards;
            private Hashing algo;
            private Pattern keyTagPattern;
    
            public ShardedJedisFactory(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
                this.shards = shards;
                this.algo = algo;
                this.keyTagPattern = keyTagPattern;
            }
    
            public PooledObject<ShardedJedis> makeObject() throws Exception {
                ShardedJedis jedis = new ShardedJedis(shards, algo, keyTagPattern);
                return new DefaultPooledObject<ShardedJedis>(jedis);
            }
    
            public void destroyObject(PooledObject<ShardedJedis> pooledShardedJedis) throws Exception {
                final ShardedJedis shardedJedis = pooledShardedJedis.getObject();
                for (Jedis jedis : shardedJedis.getAllShards()) {
                    try {
                        try {
                            jedis.quit();
                        } catch (Exception e) {
    
                        }
                        jedis.disconnect();
                    } catch (Exception e) {
    
                    }
                }
            }
    
            public boolean validateObject(PooledObject<ShardedJedis> pooledShardedJedis) {
                try {
                    ShardedJedis jedis = pooledShardedJedis.getObject();
                    for (Jedis shard : jedis.getAllShards()) {
                        if (!shard.ping().equals("PONG")) {
                            return false;
                        }
                    }
                    return true;
                } catch (Exception ex) {
                    return false;
                }
            }
    
            public void activateObject(PooledObject<ShardedJedis> p) throws Exception {
    
            }
    
            public void passivateObject(PooledObject<ShardedJedis> p) throws Exception {
    
            }
        }
    
        protected class JedisPubSubAdapter extends JedisPubSub {
            @Override
            public void onMessage(String channel, String message) {
            }
    
            @Override
            public void onPMessage(String pattern, String channel, String message) {
            }
    
            @Override
            public void onPSubscribe(String pattern, int subscribedChannels) {
            }
    
            @Override
            public void onPUnsubscribe(String pattern, int subscribedChannels) {
            }
    
            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
            }
    
            @Override
            public void onUnsubscribe(String channel, int subscribedChannels) {
            }
        }
    
        protected class MasterListener extends Thread {
    
            protected List<String> masters;
            protected String host;
            protected int port;
            protected long subscribeRetryWaitTimeMillis = 5000;
            protected Jedis jedis;
            protected AtomicBoolean running = new AtomicBoolean(false);
    
            protected MasterListener() {
            }
    
            public MasterListener(List<String> masters, String host, int port) {
                this.masters = masters;
                this.host = host;
                this.port = port;
            }
    
            public MasterListener(List<String> masters, String host, int port,
                                  long subscribeRetryWaitTimeMillis) {
                this(masters, host, port);
                this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
            }
    
            public void run() {
    
                running.set(true);
    
                while (running.get()) {
    
                    jedis = new Jedis(host, port);
    
                    try {
                        jedis.subscribe(new JedisPubSubAdapter() {
                            @Override
                            public void onMessage(String channel, String message) {
                                log.info("Sentinel " + host + ":" + port + " published: " + message + ".");
    
                                String[] switchMasterMsg = message.split(" ");
    
                                if (switchMasterMsg.length > 3) {
    
                                    int index = masters.indexOf(switchMasterMsg[0]);
                                    if (index >= 0) {
                                        HostAndPort newHostMaster = toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4]));
                                        List<HostAndPort> newHostMasters = new ArrayList<HostAndPort>();
                                        for (int i = 0; i < masters.size(); i++) {
                                            newHostMasters.add(null);
                                        }
                                        Collections.copy(newHostMasters, currentHostMasters);
                                        newHostMasters.set(index, newHostMaster);
    
                                        initPool(newHostMasters);
                                    } else {
                                        StringBuffer sb = new StringBuffer();
                                        for (String masterName : masters) {
                                            sb.append(masterName);
                                            sb.append(",");
                                        }
                                        log.info("Ignoring message on +switch-master for master name "
                                                + switchMasterMsg[0]
                                                + ", our monitor master name are ["
                                                + sb + "]");
                                    }
    
                                } else {
                                    log.warn("Invalid message received on Sentinel "
                                            + host
                                            + ":"
                                            + port
                                            + " on channel +switch-master: "
                                            + message);
                                }
                            }
                        }, "+switch-master");
    
                    } catch (JedisConnectionException e) {
    
                        if (running.get()) {
                            log.warn("Lost connection to Sentinel at " + host
                                    + ":" + port
                                    + ". Sleeping 5000ms and retrying.");
                            try {
                                Thread.sleep(subscribeRetryWaitTimeMillis);
                            } catch (InterruptedException e1) {
                                e1.printStackTrace();
                            }
                        } else {
                            log.info("Unsubscribing from Sentinel at " + host + ":"
                                    + port);
                        }
                    }
                }
            }
    
            public void shutdown() {
                try {
                    log.info("Shutting down listener on " + host + ":" + port);
                    running.set(false);
                    // This isn't good, the Jedis object is not thread safe
                    jedis.disconnect();
                } catch (Exception e) {
                    log.warn("Caught exception while shutting down: " + e.getMessage());
                }
            }
        }
    }

[https_github.com_warmbreeze_sharded-jedis-sentinel-pool_blob_master_src_main_java_redis_clients_jedis_ShardedJedisSentinelPool.java]: https://github.com/warmbreeze/sharded-jedis-sentinel-pool/blob/master/src/main/java/redis/clients/jedis/ShardedJedisSentinelPool.java
[e6f32bb5-f983-3ffe-97f2-a7fa7ce9d26f.png]: /images/20220302/b6f758c217324b8590553e74476d33b2.png
[1507c655-a4e7-33c5-bb22-38edb76fb1d5.png]: /images/20220302/1a256b33afc14b0f930bb51e9041eda6.png