什么是Etcd？

etcd是一个强大的一致性的分布式键值存储，它提供了一种可靠的方式来存储需要由分布式系统或机器群访问的数据。它优雅地处理网络分区期间的领导者选举，并且可以容忍机器故障，即使在领导者节点中也是如此。从简单的Web应用程序到Kubernetes，任何复杂的应用程序都可以读取数据并将数据写入etcd。这是官方对Etcd的描述，基于这些特性，Etcd常用于分布式配置、分布式锁、分布式服务协调者以及分布式注册。从功能上来说和zookeeper是一类项目，但是相比而言etcd更现代，etcd使用go语言开发，编译后生成了系统可执行的二进制产物，跨平台性更好，更易维护。etcd直接提供http的接口，非常方便各大语言封装自己的client sdk，在易用性方面也更好一点。下面也主要使用java的客户端jetcd，解决主备服务的协调问题。

etcd项目地址：https://github.com/etcd-io/etcd

etcd官网：https://etcd.io

jetcd地址：https://github.com/etcd-io/jetcd

主备服务场景描述

很多时候为了服务的高可用，除了有个在工作的主服务外，还需要多启用几个备用服务，这样，在主服务出现故障时，备用服务能够马上顶上。这个场景有个很明显的特征就是同一时间只能有一个主服务。常见的如mysql主从切换等，同一时间只能有一个msyql负责写数据。在我们这边的场景是，有一个binlog解析服务，实时解析mysql 的binlog，将解析到的数据传递到kafka中，kafka消费端有一个Flink job去消费解析的数据。最终这些数据会下层到数据中台中，提供给中台系统做基础的业务数据。很多在线的服务查询的数据就是来源binlog解析的数据，所以binlog解析的服务不能存在单点故障，在架构上只能是一主多备的模式，主服务故障时，备用服务实时顶上。同时binlog服务也不能同时多个解析。所以，这个场景使用etcd来做主备架构再好不过了。

jetcd具体实现

首先引入jetcd依赖

        <dependency>
            <groupId>io.etcd</groupId>
            <artifactId>jetcd-core</artifactId>
            <version>0.3.0</version>
        </dependency>

初始化客户端

     Client client = Client.builder().endpoints(
                "http://127.0.0.1:2379",
                "http://127.0.0.1:3379",
                "http://127.0.0.1:4379"
        ).build();

关键api介绍

        Lock lock = client.getLockClient();
        Lease lease = client.getLeaseClient();

Lease提供授予，撤销和保持租约的方法，其中有两个关键方法grant(long ttl)和keepAlive()。grant用于授予租约，入参为租约的时间，即如果创建带租约的key值，ttl秒后即自动删除，返回租约的id。keepAlive()方法用于保持租约有效，即如果租约即将到期时，keepAlive能够自动续租ttl时间。
Lock有两个方法，lock(ByteSequence name, long leaseId)和unlock(ByteSequence lockKey)。来实现分布式锁的功能，其中加锁时，入参leaseid为续约对象的id，即定义了持有锁的时间

通过这Lease和Lock的功能，很容易实现主备服务的切换。关键代码如下：

        ByteSequence lockKey = ByteSequence.from("/root/lock", StandardCharsets.UTF_8);
        Lock lock = client.getLockClient();
        Lease lease = client.getLeaseClient();
        long leaseId = lease.grant(lockTTl).get().getID();
        lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
            @Override
            public void onNext(LeaseKeepAliveResponse value) {
                System.err.println("LeaseKeepAliveResponse value:" + value.getTTL());
            }
            @Override
            public void onError(Throwable t) { t.printStackTrace(); }
            @Override
            public void onCompleted() { }
        });
        lock.lock(lockKey, leaseId).get().getKey();

首先申请授予续约获取到leaseId，其中lockttl为1，单位秒，etcd的租约是秒级的。在这里ttl的设置是有讲究的，取决于当主服务故障时，你想多快让从服务感知并顶上。当然，受限于etcd本身租约秒级限制，最快也只能是1秒。
然后调用keepAlive方法，使授予到的leaseid保活，这样，只要应用还存活就会自动续约
接着调用lock方法，传入leaseid。只有首次启动的服务会获取到锁，而且在运行期间，会不断的续约。当从服务运行到此处时，会阻塞住。这样就能保证多个服务同时运行，只有一个服务真正工作的目的。当获取到锁的主服务出现问题时，原先的只有锁的续约在1秒内就会到期，从服务会马上获取到锁执行工作代码

完整的测试用例

/**
 * @author: kl @kailing.pub
 * @date: 2019/7/22
 */
public class JEtcdTest {

    private Client client;
    private Lock lock;
    private Lease lease;
    //单位：秒
    private long lockTTl = 1;
    private ByteSequence lockKey = ByteSequence.from("/root/lock", StandardCharsets.UTF_8);
    private ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);

    @Before
    public void setUp() {
         client = Client.builder().endpoints(
                "http://127.0.0.1:2379",
                "http://127.0.0.1:3379",
                "http://127.0.0.1:4379"
        ).build();
         lock = client.getLockClient();
         lease = client.getLeaseClient();
    }

    @Test
    public void lockTest1toMaster() throws InterruptedException, ExecutionException {
        long leaseId = lease.grant(lockTTl).get().getID();
         lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
             @Override
             public void onNext(LeaseKeepAliveResponse value) {
                 System.err.println("LeaseKeepAliveResponse value:"+ value.getTTL());
             }

             @Override
             public void onError(Throwable t) {
                 scheduledThreadPool.shutdownNow();
                 t.printStackTrace();
                    
             }

             @Override
             public void onCompleted() {
                 scheduledThreadPool.shutdownNow();
             }
         });
        lock.lock(lockKey, leaseId).get().getKey();

        scheduledThreadPool.submit(() -> {
            while (true) {
                System.err.println("我是主服务开始工作了");
                TimeUnit.SECONDS.sleep(1);
            }
        });
        TimeUnit.DAYS.sleep(1);
    }

    @Test
    public void lockTest2toStandby() throws InterruptedException, ExecutionException {
        long leaseId = lease.grant(lockTTl).get().getID();
        lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
            @Override
            public void onNext(LeaseKeepAliveResponse value) {
                System.err.println("LeaseKeepAliveResponse value:"+ value.getTTL());
            }

            @Override
            public void onError(Throwable t) {
                scheduledThreadPool.shutdownNow();
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                 scheduledThreadPool.shutdownNow();
            }
        });
        lock.lock(lockKey, leaseId).get().getKey();
        scheduledThreadPool.submit(() -> {
            while (true) {
                System.err.println("我是备用服务，我开始工作了，估计主服务已经挂了");
                TimeUnit.SECONDS.sleep(1);
            }
        });
        TimeUnit.DAYS.sleep(1);
    }

    @Test
    public void lockTest3toStandby() throws InterruptedException, ExecutionException {
        long leaseId = lease.grant(lockTTl).get().getID();
        lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
            @Override
            public void onNext(LeaseKeepAliveResponse value) {
                System.err.println("LeaseKeepAliveResponse value:"+ value.getTTL());
            }

            @Override
            public void onError(Throwable t) {
                scheduledThreadPool.shutdownNow();
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                scheduledThreadPool.shutdownNow();
            }
        });
        lock.lock(lockKey, leaseId).get().getKey();

        scheduledThreadPool.submit(() -> {
            while (true) {
                System.err.println("我是备用服务，我开始工作了，估计主服务已经挂了");
                TimeUnit.SECONDS.sleep(1);
            }
        });
        TimeUnit.DAYS.sleep(1);
    }
}

上面测试用例模拟了一主两备的高可用架构。分别执行lockTest1toMaster()、lockTest2toStandby()、lockTest3toStandby()服务，会发现只有一个服务会打印。然后手动关闭这个服务，从服务马上会接着打印。在关闭这个从服务，另外一个从服务就会接着打印。很好的模拟了主备故障切换的效果

etcd选主实现故障主备秒级切换高可用架构

什么是Etcd？

主备服务场景描述

jetcd具体实现

首先引入jetcd依赖

初始化客户端

关键api介绍

完整的测试用例

Redis 流量复制、流量回放、流量镜像

Redis 的数据清理策略详解

基于 Prometheus + Grafana 实现 Nexus 监控观测

jedis 连接池 jedisPool 的实现原理浅析

xxl-job 登入功能集成 OIDC 统一认证

2023 KL的年终总结

记一次 golang 的 zstd 压缩、解压缩优化

Redis 流量复制、流量回放、流量镜像

Redis 的数据清理策略详解

Sentry Snuba Clickhoue 数据源分析

文章分类

友情链接

文章归档

随机文章