`

JedisSentinelPool源代码

 
阅读更多

概述

Jedis是Redis官方推荐的Java客户端,更多Redis的客户端可以参考Redis官网客户端列表。Redis-Sentinel作为官方推荐的HA解决方案,Jedis也在客户端角度实现了对Sentinel的支持,主要实现在JedisSentinelPool.java这个类中,下文会分析这个类的实现。

属性

JedisSentinelPool类里有以下的属性:

    //基于apache的commom-pool2的对象池配置
    protected GenericObjectPoolConfig poolConfig;

    //超时时间,默认是2000
    protected int timeout = Protocol.DEFAULT_TIMEOUT;

    //sentinel的密码
    protected String password;

    //redis数据库的数目
    protected int database = Protocol.DEFAULT_DATABASE;

    //master监听器,当master的地址发生改变时,会触发这些监听者
    protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();

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

    //Jedis实例创建工厂
    private volatile JedisFactory factory;

    //当前的master,HostAndPort是一个简单的包装了ip和port的模型类
    private volatile HostAndPort currentHostMaster;

构造器

构造器的代码如下:

public JedisSentinelPool(String masterName, 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;

        HostAndPort master = initSentinels(sentinels, masterName);
        initPool(master);
}

构造器一开始对实例变量进行赋值,参数sentinels是客户端所需要打交道的Redis-Sentinel,允许有多个,用一个集合来盛装。

然后通过initSentinels方法与sentinel沟通后,确定当前sentinel所监视的master是哪一个。然后通过master来创建好对象池,以便后续从对象池中取出一个Jedis实例,来对master进行操作。

initSentinels方法

initSentinels方法的代码如下所示,我加了一些注释:

    private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {

        HostAndPort master = null;
        boolean sentinelAvailable = false;

        log.info("Trying to find master from available Sentinels...");

        // 有多个sentinels,遍历这些个sentinels
        for (String sentinel : sentinels) {
            // host:port表示的sentinel地址转化为一个HostAndPort对象。
            final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));

            log.fine("Connecting to Sentinel " + hap);

            Jedis jedis = null;
            try {
                // 连接到sentinel
                jedis = new Jedis(hap.getHost(), hap.getPort());

                // 根据masterName得到master的地址,返回一个list,host= list[0], port =
                // list[1]
                List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);

                // connected to sentinel...
                sentinelAvailable = true;

                if (masterAddr == null || masterAddr.size() != 2) {
                    log.warning("Can not get master addr, master name: " + masterName
                            + ". Sentinel: " + hap + ".");
                    continue;
                }

                master = toHostAndPort(masterAddr);
                log.fine("Found Redis master at " + master);
                // 如果在任何一个sentinel中找到了master,不再遍历sentinels
                break;
            } catch (JedisConnectionException e) {
                log.warning("Cannot connect to sentinel running @ " + hap
                        + ". Trying next one.");
            } finally {
                // 关闭与sentinel的连接
                if (jedis != null) {
                    jedis.close();
                }
            }
        }

        // 到这里,如果master为null,则说明有两种情况:
// 一种是所有的sentinels节点都down掉了一种是master节点没有被存活的sentinels监控到
        if (master == null) {
            if (sentinelAvailable) {
                // can connect to sentinel, but master name seems to not
                // monitored
                throw new JedisException("Can connect to sentinel, but " + masterName
                        + " seems to be not monitored...");
            } else {
                throw new JedisConnectionException(
                        "All sentinels down, cannot determine where is " + masterName
                                + " master is running...");
            }
        }

        //如果走到这里,说明找到了master的地址
        log.info("Redis master running at " + master + ", starting Sentinel listeners...");

        //启动对每个sentinels的监听
        for (String sentinel : sentinels) {
            final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
            MasterListener masterListener = new MasterListener(masterName, hap.getHost(),
                    hap.getPort());
            masterListeners.add(masterListener);
            masterListener.start();
        }

        return master;
    }

可以看到initSentinels方法的参数有一个masterName,就是我们所需要查找的master的名字。
一开始,遍历多个sentinels,一个一个连接到sentinel,去询问关于masterName的消息,可以看到是通过jedis.sentinelGetMasterAddrByName()方法去连接sentinel,并询问当前的master的地址。点进这个方法去看看,源代码是这样写的:

/**
   * <pre>
   * redis 127.0.0.1:26381> sentinel get-master-addr-by-name mymaster
   * 1) "127.0.0.1"
   * 2) "6379"
   * </pre>
   * @param masterName
   * @return two elements list of strings : host and port.
   */
  public List<String> sentinelGetMasterAddrByName(String masterName) {
    client.sentinel(Protocol.SENTINEL_GET_MASTER_ADDR_BY_NAME, masterName);
    final List<Object> reply = client.getObjectMultiBulkReply();
    return BuilderFactory.STRING_LIST.build(reply);
  }

调用的是与Jedis绑定的client去发送一个"get-master-addr-by-name"命令。

回到initSentinels方法中,如果没有询问到master的地址,那就询问下一个sentinel。如果询问到了master的地址,那么将不再遍历sentinel集合,直接break退出循环遍历。

如果循环结束后,master的值为null,那么有两种可能:

  • 一种是所有的sentinel实例都不可用了
  • 另外一种是,sentinel实例有可用的,但是没有监控名字为masterName的Redis。

如果master为null,程序会抛出异常,不再往下走了。如果master不为null呢,继续往下走。

可以从代码中看到,为每个sentinel都启动了一个监听者MasterListener。MasterListener本身是一个线程,它会去订阅sentinel上关于master节点地址改变的消息。

接下来先分析构造方法中的另外一个方法:initPool。之后再看MasterListener的实现。

initPool方法

initPool的实现源代码如下所示:

private void initPool(HostAndPort master) {
        if (!master.equals(currentHostMaster)) {
            currentHostMaster = master;
            if (factory == null) {
                factory = new JedisFactory(master.getHost(), master.getPort(), timeout,
                        password, database);
                initPool(poolConfig, factory);
            } else {
                factory.setHostAndPort(currentHostMaster);
                // although we clear the pool, we still have to check the
                // returned object
                // in getResource, this call only clears idle instances, not
                // borrowed instances
                internalPool.clear();
            }

            log.info("Created JedisPool to master at " + master);
        }
    }

可以看到,作为参数传进来的master会与实例变量currentHostMaster作比较,看看是否是相同的,为什么要作这个比较呢,因为前文中提到的MasterListener会在发现master地址改变以后,去调用initPool方法。
如果是第一次调用initPool方法(构造函数中调用),那么会初始化Jedis实例创建工厂,如果不是第一次调用(MasterListener中调用),那么只对已经初始化的工厂进行重新设置。
从以上也可以看出为什么currentHostMasterfactory这两个变量为什么要声明为volatile,它们会在多线程环境下被访问和修改,因此必须保证可见性
第一次调用时,会调用initPool(poolConfig, factory)方法。
看看这个方法的源代码:

public void initPool(final GenericObjectPoolConfig poolConfig,
            PooledObjectFactory<T> factory) {

        if (this.internalPool != null) {
            try {
                closeInternalPool();
            } catch (Exception e) {
            }
        }

        this.internalPool = new GenericObjectPool<T>(factory, poolConfig);
    }

基本上只干了一件事:初始化内部对象池。

MasterListener监听者线程

直接看它的run方法实现吧:

        public void run() {

            running.set(true);

            while (running.get()) {

                j = new Jedis(host, port);

                try {
                    //订阅sentinel上关于master地址改变的消息
                    j.subscribe(new JedisPubSub() {
                        @Override
                        public void onMessage(String channel, String message) {
                            log.fine("Sentinel " + host + ":" + port + " published: "
                                    + message + ".");

                            String[] switchMasterMsg = message.split(" ");

                            if (switchMasterMsg.length > 3) {

                                if (masterName.equals(switchMasterMsg[0])) {
                                    initPool(toHostAndPort(Arrays.asList(
                                            switchMasterMsg[3], switchMasterMsg[4])));
                                } 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 " + 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);
                    }
                }
            }
        }

可以看到它依然委托了Jedis去与sentinel打交道,订阅了关于master地址变换的消息,当master地址变换时,就会再调用一次initPool方法,重新设置对象池相关的设置。

尾声

Jedis的JedisSentinelPool的实现仅仅适用于单个master-slave。
现在有了更多的需求,既需要sentinel提供的自动主备切换机制,又需要客户端能够做数据分片(Sharding),类似于memcached用一致性哈希进行数据分片。
接下来可能会自己在现有Jedis上实现一个支持一致性哈希分片的ShardedJedisSentinelPool。

分享到:
评论

相关推荐

    jedis通过JedisSentinelPool获取jedis操作redis

    jedis通过JedisSentinelPool获取jedis操作redis,通过配置sentienl,获取sentinel连接池,通过sentinel,获取jedis信息

    redis,jedis,sentinel需要的两个jar包

    JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels); Jedis jedis = pool.getResource(); String value = jedis.get("key"); System.out.println(value); jedis.close(); pool.close(); ``` ...

    jedis-jedis-1.5.0-RC1.zip

    `jedis-jedis-1.5.0-RC1.zip`是一个包含Jedis 1.5.0 Release Candidate 1版本的压缩包,它可能包含了源代码、文档、示例和测试用例等内容,用于开发者在项目中集成或测试Jedis的最新功能。 Jedis的主要特点包括: 1...

    jedis-jedis-3.6.0.tar.gz

    标题"jedis-jedis-3.6.0.tar.gz"表明这是Jedis的一个版本3.6.0的归档文件,通常包含源代码、文档、示例和构建脚本等内容。描述中的"jedis-jedis-3.6.0.tar"进一步确认了这是一个tar文件,通常在Unix或Linux环境中...

    redis in action java代码

    这本书通过丰富的实例和Java代码来解释Redis的各种功能,帮助开发者更好地理解和应用Redis。然而,提供的压缩包中并未包含第三章的Java代码,可能是因为该部分的内容涉及到特定的话题或者在其他资源中单独提供。 ...

    Redis单机、主从、哨兵Jave-Jedis连接代码

    JedisSentinelPool sentinelPool = new JedisSentinelPool( masterGroupName, sentinels, new SentinelConfig(), null); Jedis jedis = sentinelPool.getResource(); jedis.set("key", "value"); String value...

    jedis-jedis-3.5.2.zip

    Jedis的源代码是开放的,开发者可以通过阅读源码了解其内部实现,也可以根据需求进行定制化开发。社区对Jedis的持续维护和更新,使得它成为了Java开发者首选的Redis客户端工具。 总之,Jedis 3.5.2是一个强大的...

    jedis-jedis-2.9.3.tar.gz

    在此次提供的"jedis-jedis-2.9.3.tar.gz"压缩包中,包含了Jedis的2.9.3版本源代码和其他相关资源。这个版本的发布通常意味着对之前的版本进行了优化、修复了已知问题,同时也可能添加了一些新的特性和功能。 首先,...

    jedispool连redis高并发卡死的问题

    下面是一个简单的示例代码: ```java public class RedisUtil { protected static ReentrantLock lockPool = new ReentrantLock(); protected static ReentrantLock lockJedis = new ReentrantLock(); protected...

    java客户端Jedis操作Redis Sentinel 连接池的实现方法

    private static JedisSentinelPool pool = null; public static Properties getJedisProperties() { Properties config = new Properties(); InputStream is = null; try { is = JedisPoolUtil.class....

    集群redis哨兵模式连接方式,解决database不生效问题(附源码)

    但在哨兵模式下,如果客户端直接连接到哨兵,哨兵不会自动处理`SELECT`命令,因此需要在客户端代码中处理数据库切换。 例如,以下是一个Java客户端连接Redis哨兵并选择数据库的源码示例: ```java import redis....

    java连接Redis所需jar包

    Jedis提供了一系列的接口和类,如Jedis、JedisSentinelPool、SentinelManager等,以供开发者使用。例如,`Jedis`是基础的客户端接口,通过它可以执行基本的Redis命令,如设置和获取字符串、操作列表、集合、哈希表等...

    Redis连接测试工程,支持Sentinel和Cluster模式,可直接运行

    在Java应用中,使用JedisSentinelPool管理连接池,可以方便地实现对Sentinel的交互和故障转移。 3. **Cluster集群模式**:Redis Cluster实现了数据的分布式存储,将数据分散在多个节点上,增强了系统的扩展性。每个...

    redis中master及哨兵调用demo

    JedisSentinelPool pool = new JedisSentinelPool("myMaster", sentinels); Jedis resource = pool.getResource(); resource.set("key", "value"); ``` 3. 注意事项:在Sentinel模式下,应用不再直接与Master交互,...

    Redis Sentinel主从高可用方案1

    这段代码创建了一个 Sentinel 连接池,`"mymaster"` 是主服务器的名称,`sentinels` 集合包含了 Sentinel 实例的地址。`pool.getResource()` 返回的 Jedis 实例将始终保持与当前主服务器的连接。 总结来说,Redis ...

    redis(哨兵模式配置)

    2. **配置 Sentinel**:创建 JedisSentinelPool 实例,传入哨兵节点列表、主服务的名称以及连接池配置。 3. **获取 Redis 连接**:从 JedisSentinelPool 获取 Jedis 实例,进行读写操作。 4. **关闭连接**:使用完 ...

    redis哨兵的使用

    ##### 示例代码: ```java public class App { public static void main(String[] args) { Set&lt;String&gt; sentinels = new HashSet(); sentinels.add("172.18.18.207:26379"); // 哨兵地址 JedisSentinelPool ...

    jedis-jedis-3.4.0.zip

    - 用户可以通过配置文件或者代码设置Jedis的超时时间、客户端名称、密码认证等参数,以适应不同的应用场景。 9. **性能优化**: - 每个新版本的Jedis都会针对性能进行优化,3.4.0也不例外。这可能包括减少内存...

    Jedis出现connection timeout问题解决方法(JedisPool连接池使用实例)

    在使用Redis客户端Jedis时,有时会遇到"connection timeout"的问题,这通常意味着客户端尝试连接到Redis服务器时超时,无法建立有效的网络连接。解决这个问题的一种常见策略是使用Jedis的连接池JedisPool,它提供了...

Global site tag (gtag.js) - Google Analytics