基於Redis Sentinel主從切換以及Sharding分片的Redis叢集高可用方案
阿新 • • 發佈：2018-12-22
本文主要介紹一種通過Jedis&Sentinel實現Redis叢集高可用方案，該方案需要使用Jedis2.2.2及以上版本（強制），Redis2.8及以上版本(可選，Sentinel最早出現在Redis2.4中，Redis2.8中Sentinel更加穩定)，Redis叢集是以分片(Sharding)加主從的方式搭建，滿足可擴充套件性的要求；
Redis Sentinel介紹
Redis Sentinel是Redis官方提供的叢集管理工具，主要有三大功能：
監控，能持續監控Redis的主從例項是否正常工作；
通知，當被監控的Redis例項出問題時，能通過API通知系統管理員或其他程式；
自動故障恢復，如果主例項無法正常工作，Sentinel將啟動故障恢復機制把一個從例項提升為主例項，其他的從例項將會被重新配置到新的主例項，且應用程式會得到一個更換新地址的通知。
Redis Sentinel是一個分散式系統，可以部署多個Sentinel例項來監控同一組Redis例項，它們通過Gossip協議來確定一個主例項宕機，通過Agreement協議來執行故障恢復和配置變更，一般在生產環境中部署多個例項來提高系統可用性，只要有一個Sentinel例項執行正常，就能保證被監控的Redis例項執行正常（類似Zookeeper，通過多個Zookeeper來提高系統可用性）；
本文不涉及Sentinel的實現細節和工作原理，讀者可以閱讀其他文章瞭解；
Redis HA方案
HA的關鍵在於避免單點故障及故障恢復，在Redis Cluster未釋出之前，Redis一般以主/從方式部署（這裡討論的應用從例項主要用於備份，主例項提供讀寫，有不少應用是讀寫分離的，讀寫操作需要取不同的Redis例項，該方案也可用於此種應用，原理都是相通的，區別在於資料操作層如何封裝），該方式要實現HA主要有如下幾種方案：
1，keepalived：通過keepalived的虛擬IP，提供主從的統一訪問，在主出現問題時，通過keepalived執行指令碼將從提升為主，待主恢復後先同步後自動變為主，該方案的好處是主從切換後，應用程式不需要知道(因為訪問的虛擬IP不變)，壞處是引入keepalived增加部署複雜性；
2，zookeeper：通過zookeeper來監控主從例項，維護最新有效的IP，應用通過zookeeper取得IP，對Redis進行訪問；
3，sentinel：通過Sentinel監控主從例項，自動進行故障恢復，該方案有個缺陷：因為主從例項地址(IP&PORT)是不同的，當故障發生進行主從切換後，應用程式無法知道新地址，故在Jedis2.2.2中新增了對Sentinel的支援，應用通過redis.clients.jedis.JedisSentinelPool.getResource()取得的Jedis例項會及時更新到新的主例項地址。
筆者所在的公司先使用了方案1一段時間後，發現keepalived在有些情況下會導致資料丟失，keepalived通過shell指令碼進行主從切換，配置複雜，而且keepalived成為新的單點，後來選用了方案3，使用Redis官方解決方案；（方案2需要編寫大量的監控程式碼，沒有方案3簡便，網上有人使用方案2讀者可自行檢視）
選用Sentinel出現的問題
Sentinel&Jedis看上去是個完美的解決方案，這句話只說對了一半，在無分片的情況是這樣，但我們的應用使用了資料分片-sharing，資料被平均分佈到4個不同的例項上，每個例項以主從結構部署，Jedis沒有提供基於Sentinel的ShardedJedisPool，也就是說在4個分片中，如果其中一個分片發生主從切換，應用所使用的ShardedJedisPool無法獲得通知，所有對那個分片的操作將會失敗。
本文提供一個基於Sentinel的ShardedJedisPool，能及時感知所有分片主從切換行為，進行連線池重建，原始碼見ShardedJedisSentinelPool.java
ShardedJedisSentinelPool實現分析
建構函式
類似之前的Jedis Pool的構造方法，需要引數poolConfig提供諸如maxIdle，maxTotal之類的配置，masters是一個List，用來儲存所有分片Master在Sentinel中配置的名字（注意master的順序不能改變，因為Shard演算法是依據分片位置進行計算，如果順序錯誤將導致資料儲存混亂），sentinels是一個Set，其中存放所有Sentinel的地址(格式：IP:PORT，如127.0.0.1:26379)，順序無關；
初始化連線池
在建構函式中，通過方法
取得當前所有分片的master地址(IP&PORT)，對每個分片，通過順次連線Sentinel例項，獲取該分片的master地址，如果無法獲得，即所有Sentinel都無法連線，將休眠1秒後繼續重試，直到取得所有分片的master地址，程式碼塊如下： 
通過
初始化連線池，到此連線池中的所有連線都指向分片的master；
監控每個Sentinel
在方法
最後，會為每個Sentinel啟動一個Thread來監控Sentinel做出的更改： 
該執行緒的run方法通過Jedis Pub/Sub API（實現JedisPubSub介面，並通過jedis.subscribe進行訂閱）向Sentinel例項訂閱“+switch-master”頻道，當Sentinel進行主從切換時，該執行緒會得到新Master地址的通知，通過master name判斷哪個分片進行了切換，將新master地址替換原來位置的地址，並呼叫initPool(List masters)進行Jedis連線池重建；後續所有通過該連線池取得的連線都指向新Master地址，對應用程式透明；
應用示例
maven:
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.0</version>
<type>jar</type>
<scope>compile</scope>
</dependency>程式碼使用如下：
GenericObjectPoolConfig config = new GenericObjectPoolConfig();
config.setMaxTotal(100);
config.setMaxIdle(10);
List<String> masters = new ArrayList<String>();
masters.add("master-1");//sentinel.conf檔案中的名字
masters.add("master-2");
Set<String> sentinels = new HashSet<String>();
sentinels.add("192.168.248.127:63792");
sentinels.add("192.168.248.128:63792");
ShardedJedisSentinelPool pool = new ShardedJedisSentinelPool(masters, sentinels, config, 60000);
ShardedJedis j = null;
for (int i = 11; i < 15; i++) {
try {
j = pool.getResource();
j.set("KEY: " + i, "" + i);
pool.returnResource(j);
} catch (JedisConnectionException e) {
e.printStackTrace();
}
}
pool.destroy();ShardedJedisSentinelPool檔案內容如下：
/**
* 2017/4/30.
*/
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;
import java.util.regex.Pattern;
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.*;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.Hashing;
import redis.clients.util.Pool;
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.fine("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.fine("Found Redis master at " + master);
shardMasters.add(master);
masterMap.put(masterName, master);
fetched = true;
jedis.disconnect();
break;
}
}
} catch (JedisConnectionException e) {
log.warning("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
}
}
if (null == master) {
try {
log.severe("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.severe("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.fine("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.fine("Ignoring message on +switch-master for master name "
+ switchMasterMsg[0]
+ ", our monitor master name are ["
+ sb + "]");
}
} else {
log.severe("Invalid message received on Sentinel "
+ host
+ ":"
+ port
+ " on channel +switch-master: "
+ message);
}
}
}, "+switch-master");
} catch (JedisConnectionException e) {
if (running.get()) {
log.severe("Lost connection to Sentinel at " + host
+ ":" + port
+ ". Sleeping 5000ms and retrying.");
try {
Thread.sleep(subscribeRetryWaitTimeMillis);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
} else {
log.fine("Unsubscribing from Sentinel at " + host + ":"
+ port);
}
}
}
}
public void shutdown() {
try {
log.fine("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.severe("Caught exception while shutting down: " + e.getMessage());
}
}
}
}
總結
本文通過現實中遇到的問題，即在Redis資料分片的情況下，在使用Sentinel做HA時，如何做到主從的切換對應用程式透明，通過Jedis的Pub/Sub功能，能同時監控多個分片的主從切換情況，並通過監聽到的新地址重新構造連線池，後續從連線池中取得的所有連線都指向新地址。該方案的關鍵是：使用sentinel做HA，Jedis版本必須2.2.2及以上，所有訪問Redis例項的連線都必須從連線池中獲取；
連線洩露Bug
迴圈時發現ShardedJedisSentinelPool連線池有時竟然滿了！畢竟預設是8個連線，但為什麼會這樣？查了下程式碼，果然是這個原因！這個新擴充套件的pool沒有為getResource()出來的ShardedJedis設定DataSource，所以關閉ShardedJedis時沒法正常還回到連線池中。
補丁很簡單，就是參照Jedis的ShardedJedisPool，覆寫一些資源管理的方法。
@Override
public ShardedJedis getResource() {
ShardedJedis jedis = super.getResource();
jedis.setDataSource(this);
return jedis;
}
@Override
public void returnBrokenResource(final ShardedJedis resource) {
if (resource != null) {
returnBrokenResourceObject(resource);
}
}
@Override
public void returnResource(final ShardedJedis resource) {
if (resource != null) {
resource.resetState();
returnResourceObject(resource);
}
}