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

以你之姓@ 2022-03-02 02:22 409阅读 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

ShardedJedisSentinelPool实现分析

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

  1. public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
  2.                                     final GenericObjectPoolConfig poolConfig, int timeout,
  3.                                     final String password, final int database) {
  4.         this.poolConfig = poolConfig;
  5.         this.timeout = timeout;
  6.         this.password = password;
  7.         this.database = database;
  9.         List<HostAndPort> masterList = initSentinels(sentinels, masters);
  10.         initPool(masterList);
  11.     }

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

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

  1. private List<HostAndPort> initSentinels(Set<String> sentinels, final List<String> masters) {
  3.         Map<String, HostAndPort> masterMap = new HashMap<String, HostAndPort>();
  4.         List<HostAndPort> shardMasters = new ArrayList<HostAndPort>();
  6.         log.info("Trying to find all master from available Sentinels...");
  8.         for (String masterName : masters) {
  9.             HostAndPort master = null;
  10.             boolean fetched = false;
  12.             while (!fetched && sentinelRetry < MAX_RETRY_SENTINEL) {
  13.                 for (String sentinel : sentinels) {
  14.                     final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
  16.                     log.info("Connecting to Sentinel " + hap);
  18.                     try {
  19.                         Jedis jedis = new Jedis(hap.getHost(), hap.getPort());
  20.                         master = masterMap.get(masterName);
  21.                         if (master == null) {
  22.                             List<String> hostAndPort = jedis.sentinelGetMasterAddrByName(masterName);
  23.                             if (hostAndPort != null && hostAndPort.size() > 0) {
  24.                                 master = toHostAndPort(hostAndPort);
  25.                                 log.info("Found Redis master at " + master);
  26.                                 shardMasters.add(master);
  27.                                 masterMap.put(masterName, master);
  28.                                 fetched = true;
  29.                                 jedis.disconnect();
  30.                                 break;
  31.                             }
  32.                         }
  33.                     } catch (JedisConnectionException e) {
  34.                         log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
  35.                     }
  36.                 }
  38.                 if (null == master) {
  39.                     try {
  40.                         log.warn("All sentinels down, cannot determine where is "
  41.                                 + masterName + " master is running... sleeping 1000ms, Will try again.");
  42.                         Thread.sleep(1000);
  43.                     } catch (InterruptedException e) {
  44.                         e.printStackTrace();
  45.                     }
  46.                     fetched = false;
  47.                     sentinelRetry++;
  48.                 }
  49.             }
  51.             // Try MAX_RETRY_SENTINEL times.
  52.             if (!fetched && sentinelRetry >= MAX_RETRY_SENTINEL) {
  53.                 log.warn("All sentinels down and try " + MAX_RETRY_SENTINEL + " times, Abort.");
  54.                 throw new JedisConnectionException("Cannot connect all sentinels, Abort.");
  55.             }
  56.         }
  58.         // All shards master must been accessed.
  59.         if (masters.size() != 0 && masters.size() == shardMasters.size()) {
  61.             log.info("Starting Sentinel listeners...");
  62.             for (String sentinel : sentinels) {
  63.                 final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
  64.                 MasterListener masterListener = new MasterListener(masters, hap.getHost(), hap.getPort());
  65.                 masterListeners.add(masterListener);
  66.                 masterListener.start();
  67.             }
  68.         }
  70.         return shardMasters;
  71.     }
  73. private void initPool(List<HostAndPort> masters) {
  74.         if (!equals(currentHostMasters, masters)) {
  75.             StringBuffer sb = new StringBuffer();
  76.             for (HostAndPort master : masters) {
  77.                 sb.append(master.toString());
  78.                 sb.append(" ");
  79.             }
  80.             log.info("Created ShardedJedisPool to master at [" + sb.toString() + "]");
  81.             List<JedisShardInfo> shardMasters = makeShardInfoList(masters);
  82.             // 这个initPool是redis.clients.util.Pool的方法
  83.             initPool(poolConfig, new ShardedJedisFactory(shardMasters, Hashing.MURMUR_HASH, null));
  84.             currentHostMasters = masters;
  85.         }
  86.     }

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

监控每个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地址，对应用程序透明；

全部代码：

  1. package com.artisan.redis.pool;
  3. import org.apache.commons.pool2.PooledObject;
  4. import org.apache.commons.pool2.PooledObjectFactory;
  5. import org.apache.commons.pool2.impl.DefaultPooledObject;
  6. import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
  8. import org.apache.log4j.Logger;
  9. import redis.clients.jedis.*;
  10. import redis.clients.jedis.exceptions.JedisConnectionException;
  11. import redis.clients.util.Hashing;
  12. import redis.clients.util.Pool;
  14. import java.util.*;
  15. import java.util.concurrent.atomic.AtomicBoolean;
  16. import java.util.regex.Pattern;
  18. /**
  19.  * Created by 18121254 on 2019/3/19.
  20.  */
  21. public class ShardedJedisSentinelPool extends Pool<ShardedJedis> {
  22.     public static final int MAX_RETRY_SENTINEL = 10;
  24.     protected final Logger log = Logger.getLogger(getClass().getName());
  26.     protected GenericObjectPoolConfig poolConfig;
  28.     protected int timeout = Protocol.DEFAULT_TIMEOUT;
  30.     private int sentinelRetry = 0;
  32.     protected String password;
  34.     protected int database = Protocol.DEFAULT_DATABASE;
  36.     protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();
  38.     private volatile List<HostAndPort> currentHostMasters;
  40.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels) {
  41.         this(masters, sentinels, new GenericObjectPoolConfig(),
  42.                 Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
  43.     }
  45.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels, String password) {
  46.         this(masters, sentinels, new GenericObjectPoolConfig(),
  47.                 Protocol.DEFAULT_TIMEOUT, password);
  48.     }
  50.     public ShardedJedisSentinelPool(final GenericObjectPoolConfig poolConfig, List<String> masters, Set<String> sentinels) {
  51.         this(masters, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null,
  52.                 Protocol.DEFAULT_DATABASE);
  53.     }
  55.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
  56.                                     final GenericObjectPoolConfig poolConfig, int timeout,
  57.                                     final String password) {
  58.         this(masters, sentinels, poolConfig, timeout, password,
  59.                 Protocol.DEFAULT_DATABASE);
  60.     }
  62.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
  63.                                     final GenericObjectPoolConfig poolConfig, final int timeout) {
  64.         this(masters, sentinels, poolConfig, timeout, null,
  65.                 Protocol.DEFAULT_DATABASE);
  66.     }
  68.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
  69.                                     final GenericObjectPoolConfig poolConfig, final String password) {
  70.         this(masters, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT,
  71.                 password);
  72.     }
  74.     public ShardedJedisSentinelPool(List<String> masters, Set<String> sentinels,
  75.                                     final GenericObjectPoolConfig poolConfig, int timeout,
  76.                                     final String password, final int database) {
  77.         this.poolConfig = poolConfig;
  78.         this.timeout = timeout;
  79.         this.password = password;
  80.         this.database = database;
  82.         List<HostAndPort> masterList = initSentinels(sentinels, masters);
  83.         initPool(masterList);
  84.     }
  86.     public void destroy() {
  87.         for (MasterListener m : masterListeners) {
  88.             m.shutdown();
  89.         }
  91.         super.destroy();
  92.     }
  94.     public List<HostAndPort> getCurrentHostMaster() {
  95.         return currentHostMasters;
  96.     }
  98.     private void initPool(List<HostAndPort> masters) {
  99.         if (!equals(currentHostMasters, masters)) {
  100.             StringBuffer sb = new StringBuffer();
  101.             for (HostAndPort master : masters) {
  102.                 sb.append(master.toString());
  103.                 sb.append(" ");
  104.             }
  105.             log.info("Created ShardedJedisPool to master at [" + sb.toString() + "]");
  106.             List<JedisShardInfo> shardMasters = makeShardInfoList(masters);
  107.             initPool(poolConfig, new ShardedJedisFactory(shardMasters, Hashing.MURMUR_HASH, null));
  108.             currentHostMasters = masters;
  109.         }
  110.     }
  112.     private boolean equals(List<HostAndPort> currentShardMasters, List<HostAndPort> shardMasters) {
  113.         if (currentShardMasters != null && shardMasters != null) {
  114.             if (currentShardMasters.size() == shardMasters.size()) {
  115.                 for (int i = 0; i < currentShardMasters.size(); i++) {
  116.                     if (!currentShardMasters.get(i).equals(shardMasters.get(i))) return false;
  117.                 }
  118.                 return true;
  119.             }
  120.         }
  121.         return false;
  122.     }
  124.     private List<JedisShardInfo> makeShardInfoList(List<HostAndPort> masters) {
  125.         List<JedisShardInfo> shardMasters = new ArrayList<JedisShardInfo>();
  126.         for (HostAndPort master : masters) {
  127.             JedisShardInfo jedisShardInfo = new JedisShardInfo(master.getHost(), master.getPort(), timeout);
  128.             jedisShardInfo.setPassword(password);
  130.             shardMasters.add(jedisShardInfo);
  131.         }
  132.         return shardMasters;
  133.     }
  135.     private List<HostAndPort> initSentinels(Set<String> sentinels, final List<String> masters) {
  137.         Map<String, HostAndPort> masterMap = new HashMap<String, HostAndPort>();
  138.         List<HostAndPort> shardMasters = new ArrayList<HostAndPort>();
  140.         log.info("Trying to find all master from available Sentinels...");
  142.         for (String masterName : masters) {
  143.             HostAndPort master = null;
  144.             boolean fetched = false;
  146.             while (!fetched && sentinelRetry < MAX_RETRY_SENTINEL) {
  147.                 for (String sentinel : sentinels) {
  148.                     final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
  150.                     log.info("Connecting to Sentinel " + hap);
  152.                     try {
  153.                         Jedis jedis = new Jedis(hap.getHost(), hap.getPort());
  154.                         master = masterMap.get(masterName);
  155.                         if (master == null) {
  156.                             List<String> hostAndPort = jedis.sentinelGetMasterAddrByName(masterName);
  157.                             if (hostAndPort != null && hostAndPort.size() > 0) {
  158.                                 master = toHostAndPort(hostAndPort);
  159.                                 log.info("Found Redis master at " + master);
  160.                                 shardMasters.add(master);
  161.                                 masterMap.put(masterName, master);
  162.                                 fetched = true;
  163.                                 jedis.disconnect();
  164.                                 break;
  165.                             }
  166.                         }
  167.                     } catch (JedisConnectionException e) {
  168.                         log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
  169.                     }
  170.                 }
  172.                 if (null == master) {
  173.                     try {
  174.                         log.warn("All sentinels down, cannot determine where is "
  175.                                 + masterName + " master is running... sleeping 1000ms, Will try again.");
  176.                         Thread.sleep(1000);
  177.                     } catch (InterruptedException e) {
  178.                         e.printStackTrace();
  179.                     }
  180.                     fetched = false;
  181.                     sentinelRetry++;
  182.                 }
  183.             }
  185.             // Try MAX_RETRY_SENTINEL times.
  186.             if (!fetched && sentinelRetry >= MAX_RETRY_SENTINEL) {
  187.                 log.warn("All sentinels down and try " + MAX_RETRY_SENTINEL + " times, Abort.");
  188.                 throw new JedisConnectionException("Cannot connect all sentinels, Abort.");
  189.             }
  190.         }
  192.         // All shards master must been accessed.
  193.         if (masters.size() != 0 && masters.size() == shardMasters.size()) {
  195.             log.info("Starting Sentinel listeners...");
  196.             for (String sentinel : sentinels) {
  197.                 final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
  198.                 MasterListener masterListener = new MasterListener(masters, hap.getHost(), hap.getPort());
  199.                 masterListeners.add(masterListener);
  200.                 masterListener.start();
  201.             }
  202.         }
  204.         return shardMasters;
  205.     }
  207.     private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
  208.         String host = getMasterAddrByNameResult.get(0);
  209.         int port = Integer.parseInt(getMasterAddrByNameResult.get(1));
  211.         return new HostAndPort(host, port);
  212.     }
  214.     /**
  215.      * PoolableObjectFactory custom impl.
  216.      */
  217.     protected static class ShardedJedisFactory implements PooledObjectFactory<ShardedJedis> {
  218.         private List<JedisShardInfo> shards;
  219.         private Hashing algo;
  220.         private Pattern keyTagPattern;
  222.         public ShardedJedisFactory(List<JedisShardInfo> shards, Hashing algo, Pattern keyTagPattern) {
  223.             this.shards = shards;
  224.             this.algo = algo;
  225.             this.keyTagPattern = keyTagPattern;
  226.         }
  228.         public PooledObject<ShardedJedis> makeObject() throws Exception {
  229.             ShardedJedis jedis = new ShardedJedis(shards, algo, keyTagPattern);
  230.             return new DefaultPooledObject<ShardedJedis>(jedis);
  231.         }
  233.         public void destroyObject(PooledObject<ShardedJedis> pooledShardedJedis) throws Exception {
  234.             final ShardedJedis shardedJedis = pooledShardedJedis.getObject();
  235.             for (Jedis jedis : shardedJedis.getAllShards()) {
  236.                 try {
  237.                     try {
  238.                         jedis.quit();
  239.                     } catch (Exception e) {
  241.                     }
  242.                     jedis.disconnect();
  243.                 } catch (Exception e) {
  245.                 }
  246.             }
  247.         }
  249.         public boolean validateObject(PooledObject<ShardedJedis> pooledShardedJedis) {
  250.             try {
  251.                 ShardedJedis jedis = pooledShardedJedis.getObject();
  252.                 for (Jedis shard : jedis.getAllShards()) {
  253.                     if (!shard.ping().equals("PONG")) {
  254.                         return false;
  255.                     }
  256.                 }
  257.                 return true;
  258.             } catch (Exception ex) {
  259.                 return false;
  260.             }
  261.         }
  263.         public void activateObject(PooledObject<ShardedJedis> p) throws Exception {
  265.         }
  267.         public void passivateObject(PooledObject<ShardedJedis> p) throws Exception {
  269.         }
  270.     }
  272.     protected class JedisPubSubAdapter extends JedisPubSub {
  273.         @Override
  274.         public void onMessage(String channel, String message) {
  275.         }
  277.         @Override
  278.         public void onPMessage(String pattern, String channel, String message) {
  279.         }
  281.         @Override
  282.         public void onPSubscribe(String pattern, int subscribedChannels) {
  283.         }
  285.         @Override
  286.         public void onPUnsubscribe(String pattern, int subscribedChannels) {
  287.         }
  289.         @Override
  290.         public void onSubscribe(String channel, int subscribedChannels) {
  291.         }
  293.         @Override
  294.         public void onUnsubscribe(String channel, int subscribedChannels) {
  295.         }
  296.     }
  298.     protected class MasterListener extends Thread {
  300.         protected List<String> masters;
  301.         protected String host;
  302.         protected int port;
  303.         protected long subscribeRetryWaitTimeMillis = 5000;
  304.         protected Jedis jedis;
  305.         protected AtomicBoolean running = new AtomicBoolean(false);
  307.         protected MasterListener() {
  308.         }
  310.         public MasterListener(List<String> masters, String host, int port) {
  311.             this.masters = masters;
  312.             this.host = host;
  313.             this.port = port;
  314.         }
  316.         public MasterListener(List<String> masters, String host, int port,
  317.                               long subscribeRetryWaitTimeMillis) {
  318.             this(masters, host, port);
  319.             this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
  320.         }
  322.         public void run() {
  324.             running.set(true);
  326.             while (running.get()) {
  328.                 jedis = new Jedis(host, port);
  330.                 try {
  331.                     jedis.subscribe(new JedisPubSubAdapter() {
  332.                         @Override
  333.                         public void onMessage(String channel, String message) {
  334.                             log.info("Sentinel " + host + ":" + port + " published: " + message + ".");
  336.                             String[] switchMasterMsg = message.split(" ");
  338.                             if (switchMasterMsg.length > 3) {
  340.                                 int index = masters.indexOf(switchMasterMsg[0]);
  341.                                 if (index >= 0) {
  342.                                     HostAndPort newHostMaster = toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4]));
  343.                                     List<HostAndPort> newHostMasters = new ArrayList<HostAndPort>();
  344.                                     for (int i = 0; i < masters.size(); i++) {
  345.                                         newHostMasters.add(null);
  346.                                     }
  347.                                     Collections.copy(newHostMasters, currentHostMasters);
  348.                                     newHostMasters.set(index, newHostMaster);
  350.                                     initPool(newHostMasters);
  351.                                 } else {
  352.                                     StringBuffer sb = new StringBuffer();
  353.                                     for (String masterName : masters) {
  354.                                         sb.append(masterName);
  355.                                         sb.append(",");
  356.                                     }
  357.                                     log.info("Ignoring message on +switch-master for master name "
  358.                                             + switchMasterMsg[0]
  359.                                             + ", our monitor master name are ["
  360.                                             + sb + "]");
  361.                                 }
  363.                             } else {
  364.                                 log.warn("Invalid message received on Sentinel "
  365.                                         + host
  366.                                         + ":"
  367.                                         + port
  368.                                         + " on channel +switch-master: "
  369.                                         + message);
  370.                             }
  371.                         }
  372.                     }, "+switch-master");
  374.                 } catch (JedisConnectionException e) {
  376.                     if (running.get()) {
  377.                         log.warn("Lost connection to Sentinel at " + host
  378.                                 + ":" + port
  379.                                 + ". Sleeping 5000ms and retrying.");
  380.                         try {
  381.                             Thread.sleep(subscribeRetryWaitTimeMillis);
  382.                         } catch (InterruptedException e1) {
  383.                             e1.printStackTrace();
  384.                         }
  385.                     } else {
  386.                         log.info("Unsubscribing from Sentinel at " + host + ":"
  387.                                 + port);
  388.                     }
  389.                 }
  390.             }
  391.         }
  393.         public void shutdown() {
  394.             try {
  395.                 log.info("Shutting down listener on " + host + ":" + port);
  396.                 running.set(false);
  397.                 // This isn't good, the Jedis object is not thread safe
  398.                 jedis.disconnect();
  399.             } catch (Exception e) {
  400.                 log.warn("Caught exception while shutting down: " + e.getMessage());
  401.             }
  402.         }
  403.     }
  404. }

发表评论

表情：
评论列表 （有 0 条评论，409人围观）

还没有评论，来说两句吧...

相关阅读