聊聊lettuce的sentinel连接

本文主要研究一下lettuce的sentinel连接

RedisClient.connectSentinel

lettuce-core-5.0.4.RELEASE-sources.jar!/io/lettuce/core/RedisClient.java

    private <K, V> StatefulRedisSentinelConnection<K, V> connectSentinel(RedisCodec<K, V> codec, RedisURI redisURI,
            Duration timeout) {
        assertNotNull(codec);
        checkValidRedisURI(redisURI);

        ConnectionBuilder connectionBuilder = ConnectionBuilder.connectionBuilder();
        connectionBuilder.clientOptions(ClientOptions.copyOf(getOptions()));
        connectionBuilder.clientResources(clientResources);

        DefaultEndpoint endpoint = new DefaultEndpoint(clientOptions);

        StatefulRedisSentinelConnectionImpl<K, V> connection = newStatefulRedisSentinelConnection(endpoint, codec, timeout);

        logger.debug("Trying to get a Redis Sentinel connection for one of: " + redisURI.getSentinels());

        connectionBuilder.endpoint(endpoint).commandHandler(() -> new CommandHandler(clientOptions, clientResources, endpoint))
                .connection(connection);
        connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, redisURI);

        if (clientOptions.isPingBeforeActivateConnection()) {
            connectionBuilder.enablePingBeforeConnect();
        }

        if (redisURI.getSentinels().isEmpty() && (isNotEmpty(redisURI.getHost()) || !isEmpty(redisURI.getSocket()))) {
            channelType(connectionBuilder, redisURI);
            try {
                getConnection(initializeChannelAsync(connectionBuilder));
            } catch (RuntimeException e) {
                connection.close();
                throw e;
            }
        } else {

            boolean connected = false;
            boolean first = true;
            Exception causingException = null;
            validateUrisAreOfSameConnectionType(redisURI.getSentinels());

            for (RedisURI uri : redisURI.getSentinels()) {
                if (first) {
                    channelType(connectionBuilder, uri);
                    first = false;
                }
                connectionBuilder.socketAddressSupplier(getSocketAddressSupplier(uri));

                if (logger.isDebugEnabled()) {
                    SocketAddress socketAddress = SocketAddressResolver.resolve(uri, clientResources.dnsResolver());
                    logger.debug("Connecting to Redis Sentinel, address: " + socketAddress);
                }
                try {
                    getConnection(initializeChannelAsync(connectionBuilder));
                    connected = true;
                    break;
                } catch (Exception e) {
                    logger.warn("Cannot connect Redis Sentinel at " + uri + ": " + e.toString());
                    causingException = e;
                }
            }

            if (!connected) {
                connection.close();
                throw new RedisConnectionException("Cannot connect to a Redis Sentinel: " + redisURI.getSentinels(),
                        causingException);
            }
        }

        if (LettuceStrings.isNotEmpty(redisURI.getClientName())) {
            connection.setClientName(redisURI.getClientName());
        }

        return connection;
    }
  • connectSentinel方法,会遍历sentinel,挨个取master获取连接,如果连接不上或抛异常则继续用下一个sentinel获取
  • 如果遍历完sentinel都抛异常,则最后抛出RedisConnectionException(“Cannot connect to a Redis Sentinel: “ + redisURI.getSentinels(),causingException)
  • 这里会调用AbstractRedisClient的initializeChannelAsync方法

AbstractRedisClient.initializeChannelAsync

lettuce-core-5.0.4.RELEASE-sources.jar!/io/lettuce/core/AbstractRedisClient.java

    /**
     * Connect and initialize a channel from {@link ConnectionBuilder}.
     *
     * @param connectionBuilder must not be {@literal null}.
     * @return the {@link ConnectionFuture} to synchronize the connection process.
     * @since 4.4
     */
    @SuppressWarnings("unchecked")
    protected <K, V, T extends RedisChannelHandler<K, V>> ConnectionFuture<T> initializeChannelAsync(
            ConnectionBuilder connectionBuilder) {

        SocketAddress redisAddress = connectionBuilder.socketAddress();

        if (clientResources.eventExecutorGroup().isShuttingDown()) {
            throw new IllegalStateException("Cannot connect, Event executor group is terminated.");
        }

        logger.debug("Connecting to Redis at {}", redisAddress);

        CompletableFuture<Channel> channelReadyFuture = new CompletableFuture<>();
        Bootstrap redisBootstrap = connectionBuilder.bootstrap();

        RedisChannelInitializer initializer = connectionBuilder.build();
        redisBootstrap.handler(initializer);

        clientResources.nettyCustomizer().afterBootstrapInitialized(redisBootstrap);
        CompletableFuture<Boolean> initFuture = initializer.channelInitialized();
        ChannelFuture connectFuture = redisBootstrap.connect(redisAddress);

        connectFuture.addListener(future -> {

            if (!future.isSuccess()) {

                logger.debug("Connecting to Redis at {}: {}", redisAddress, future.cause());
                connectionBuilder.endpoint().initialState();
                channelReadyFuture.completeExceptionally(future.cause());
                return;
            }

            initFuture.whenComplete((success, throwable) -> {

                if (throwable == null) {
                    logger.debug("Connecting to Redis at {}: Success", redisAddress);
                    RedisChannelHandler<?, ?> connection = connectionBuilder.connection();
                    connection.registerCloseables(closeableResources, connection);
                    channelReadyFuture.complete(connectFuture.channel());
                    return;
                }

                logger.debug("Connecting to Redis at {}, initialization: {}", redisAddress, throwable);
                connectionBuilder.endpoint().initialState();
                Throwable failure;

                if (throwable instanceof RedisConnectionException) {
                    failure = throwable;
                } else if (throwable instanceof TimeoutException) {
                    failure = new RedisConnectionException("Could not initialize channel within "
                            + connectionBuilder.getTimeout(), throwable);
                } else {
                    failure = throwable;
                }
                channelReadyFuture.completeExceptionally(failure);

                CompletableFuture<Boolean> response = new CompletableFuture<>();
                response.completeExceptionally(failure);

            });
        });

        return new DefaultConnectionFuture<T>(redisAddress, channelReadyFuture.thenApply(channel -> (T) connectionBuilder
                .connection()));
    }
  • 这里initializeChannelAsync的时候,会调用connectionBuilder.socketAddress()方法,进而调用RedisClient的getSocketAddress方法

RedisClient.getSocketAddress

lettuce-core-5.0.4.RELEASE-sources.jar!/io/lettuce/core/RedisClient.java

    protected SocketAddress getSocketAddress(RedisURI redisURI) throws InterruptedException, TimeoutException,
            ExecutionException {
        SocketAddress redisAddress;

        if (redisURI.getSentinelMasterId() != null && !redisURI.getSentinels().isEmpty()) {
            logger.debug("Connecting to Redis using Sentinels {}, MasterId {}", redisURI.getSentinels(),
                    redisURI.getSentinelMasterId());
            redisAddress = lookupRedis(redisURI);

            if (redisAddress == null) {
                throw new RedisConnectionException("Cannot provide redisAddress using sentinel for masterId "
                        + redisURI.getSentinelMasterId());
            }

        } else {
            redisAddress = SocketAddressResolver.resolve(redisURI, clientResources.dnsResolver());
        }
        return redisAddress;
    }

    private SocketAddress lookupRedis(RedisURI sentinelUri) throws InterruptedException, TimeoutException, ExecutionException {
        try (StatefulRedisSentinelConnection<String, String> connection = connectSentinel(sentinelUri)) {
            return connection.async().getMasterAddrByName(sentinelUri.getSentinelMasterId())
                    .get(timeout.toNanos(), TimeUnit.NANOSECONDS);
        }
    }
  • getSocketAddress方法会调用lookupRedis方法,而lookupRedis方法则调用getMasterAddrByName方法,通过sentinel来获取master的ip地址

小结

  • redis的sentinel类似于一个master的服务发现中心,假设master有故障,则通过sentinel获取新的master实现failover。
  • 而sentinel部署多个来实现高可用,假设一个sentinel挂了,则client端使用下一个sentinel来获取master地址

doc

  • lettuce Redis-Sentinel

原文发布于微信公众号 - 码匠的流水账(geek_luandun)

原文发表时间:2018-09-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏c#开发者

在DataGrid中创建一个弹出式Details窗口

在DataGrid中创建一个弹出式Details窗口 这篇文章来自DotNetJunkie的提议。他最初写信要求我们提供一个关于如何创建在DataGrid 中...

3838
来自专栏Golang语言社区

Golang语言 实现线程池

1 type GoroutinePool struct { 2 Queue chan func() error 3 Number int ...

5405
来自专栏菩提树下的杨过

ExtJs学习笔记(6)_可分页的GridPanel

一.WCF部分 1.通过查看官方的示例得知,分页数据源需要一个记录总数值,为保持通用性,这里借鉴jillZhang的文章,把他写的通用类PageData拿过来...

2178
来自专栏蛋未明的专栏

Build a JavaScript Compressor tool using NodeJS, ExpressJS, Jade, UglifyJS tutorial Read more: http

1202
来自专栏大内老A

谈谈WCF中的Data Contract (1):Data Contract Overview

Contract in SO:Contract是对操作和数据的抽象 在我们看来,Service Orientation提供了一种对业务、功能进行分解的方式。针对...

1966
来自专栏Ryan Miao

MongoDB - basic

mongoDB basic from:http://www.tutorialspoint.com/mongodb prject:https://github....

3266
来自专栏Java成神之路

Java微信公众平台开发_03_消息管理之被动回复消息

上一节,我们启用服务器配置的时候,填写了一个服务器地址(url),如下图,这个url就是回调url,是开发者用来接收微信消息和事件的接口URL 。也就是说,用户...

1.4K5
来自专栏码匠的流水账

聊聊eureka的delta配置

eureka-client-1.8.8-sources.jar!/com/netflix/discovery/DiscoveryClient.java

961
来自专栏小巫技术博客

Retrofit2 &amp; RxJava2实现单文件和多文件上传

5744
来自专栏码匠的流水账

聊聊spring cloud gateway的RetryGatewayFilter

本文主要研究一下spring cloud gateway的RetryGatewayFilter

1882

扫码关注云+社区

领取腾讯云代金券