亲宝软件园·资讯

展开

etcd主备故障秒级切换

kl 人气:3

什么是Etcd?

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

etcd官网:https://etcd.io

主备服务场景描述

很多时候为了服务的高可用,除了有个在工作的主服务外,还需要多启用几个备用服务,这样,在主服务出现故障时,备用服务能够马上顶上。这个场景有个很明显的特征就是同一时间只能有一个主服务。常见的如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和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();

完整的测试用例

/**
 * @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()服务,会发现只有一个服务会打印。然后手动关闭这个服务,从服务马上会接着打印。在关闭这个从服务,另外一个从服务就会接着打印。很好的模拟了主备故障切换的效果。

加载全部内容

相关教程
猜你喜欢
用户评论