快捷導(dǎo)航

高可用架構(gòu)etcd選主故障主備秒級(jí)切換實(shí)現(xiàn)

更新時(shí)間：2022年02月24日 17:02:52 作者：kl

這篇文章主要為大家介紹了高可用架構(gòu)etcd選主故障主備秒級(jí)切換的實(shí)現(xiàn)，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

什么是Etcd？

etcd是一個(gè)強(qiáng)大的一致性的分布式鍵值存儲(chǔ)，它提供了一種可靠的方式來(lái)存儲(chǔ)需要由分布式系統(tǒng)或機(jī)器群訪問(wèn)的數(shù)據(jù)。它優(yōu)雅地處理網(wǎng)絡(luò)分區(qū)期間的領(lǐng)導(dǎo)者選舉，并且可以容忍機(jī)器故障，即使在領(lǐng)導(dǎo)者節(jié)點(diǎn)中也是如此。從簡(jiǎn)單的Web應(yīng)用程序到Kubernetes，任何復(fù)雜的應(yīng)用程序都可以讀取數(shù)據(jù)并將數(shù)據(jù)寫(xiě)入etcd。這是官方對(duì)Etcd的描述，基于這些特性，Etcd常用于分布式配置、分布式鎖、分布式服務(wù)協(xié)調(diào)者以及分布式注冊(cè)。從功能上來(lái)說(shuō)和zookeeper是一類項(xiàng)目，但是相比而言etcd更現(xiàn)代，etcd使用go語(yǔ)言開(kāi)發(fā)，編譯后生成了系統(tǒng)可執(zhí)行的二進(jìn)制產(chǎn)物，跨平臺(tái)性更好，更易維護(hù)。etcd直接提供http的接口，非常方便各大語(yǔ)言封裝自己的client sdk，在易用性方面也更好一點(diǎn)。下面也主要使用java的客戶端jetcd，解決主備服務(wù)的協(xié)調(diào)問(wèn)題。

etcd官網(wǎng)：https://etcd.io

主備服務(wù)場(chǎng)景描述

很多時(shí)候?yàn)榱朔?wù)的高可用，除了有個(gè)在工作的主服務(wù)外，還需要多啟用幾個(gè)備用服務(wù)，這樣，在主服務(wù)出現(xiàn)故障時(shí)，備用服務(wù)能夠馬上頂上。這個(gè)場(chǎng)景有個(gè)很明顯的特征就是同一時(shí)間只能有一個(gè)主服務(wù)。常見(jiàn)的如mysql主從切換等，同一時(shí)間只能有一個(gè)msyql負(fù)責(zé)寫(xiě)數(shù)據(jù)。在我們這邊的場(chǎng)景是，有一個(gè)binlog解析服務(wù)，實(shí)時(shí)解析mysql 的binlog，將解析到的數(shù)據(jù)傳遞到kafka中，kafka消費(fèi)端有一個(gè)Flink job去消費(fèi)解析的數(shù)據(jù)。最終這些數(shù)據(jù)會(huì)下層到數(shù)據(jù)中臺(tái)中，提供給中臺(tái)系統(tǒng)做基礎(chǔ)的業(yè)務(wù)數(shù)據(jù)。很多在線的服務(wù)查詢的數(shù)據(jù)就是來(lái)源binlog解析的數(shù)據(jù)，所以binlog解析的服務(wù)不能存在單點(diǎn)故障，在架構(gòu)上只能是一主多備的模式，主服務(wù)故障時(shí)，備用服務(wù)實(shí)時(shí)頂上。同時(shí)binlog服務(wù)也不能同時(shí)多個(gè)解析。所以，這個(gè)場(chǎng)景使用etcd來(lái)做主備架構(gòu)再好不過(guò)了。

jetcd具體實(shí)現(xiàn)

首先引入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();

關(guān)鍵api介紹

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

Lease提供授予，撤銷(xiāo)和保持租約的方法，其中有兩個(gè)關(guān)鍵方法grant(long ttl)和keepAlive()。grant用于授予租約，入?yún)樽饧s的時(shí)間，即如果創(chuàng)建帶租約的key值，ttl秒后即自動(dòng)刪除，返回租約的id。keepAlive()方法用于保持租約有效，即如果租約即將到期時(shí)，keepAlive能夠自動(dòng)續(xù)租ttl時(shí)間。
Lock有兩個(gè)方法，lock(ByteSequence name, long leaseId)和unlock(ByteSequence lockKey)。來(lái)實(shí)現(xiàn)分布式鎖的功能，其中加鎖時(shí)，入?yún)easeid為續(xù)約對(duì)象的id，即定義了持有鎖的時(shí)間

通過(guò)這Lease和Lock的功能，很容易實(shí)現(xiàn)主備服務(wù)的切換。關(guān)鍵代碼如下：

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();

首先申請(qǐng)授予續(xù)約獲取到leaseId，其中l(wèi)ockttl為1，單位秒，etcd的租約是秒級(jí)的。在這里ttl的設(shè)置是有講究的，取決于當(dāng)主服務(wù)故障時(shí)，你想多快讓從服務(wù)感知并頂上。當(dāng)然，受限于etcd本身租約秒級(jí)限制，最快也只能是1秒。
然后調(diào)用keepAlive方法，使授予到的leaseid?；睿@樣，只要應(yīng)用還存活就會(huì)自動(dòng)續(xù)約
接著調(diào)用lock方法，傳入leaseid。只有首次啟動(dòng)的服務(wù)會(huì)獲取到鎖，而且在運(yùn)行期間，會(huì)不斷的續(xù)約。當(dāng)從服務(wù)運(yùn)行到此處時(shí)，會(huì)阻塞住。這樣就能保證多個(gè)服務(wù)同時(shí)運(yùn)行，只有一個(gè)服務(wù)真正工作的目的。當(dāng)獲取到鎖的主服務(wù)出現(xiàn)問(wèn)題時(shí)，原先的只有鎖的續(xù)約在1秒內(nèi)就會(huì)到期，從服務(wù)會(huì)馬上獲取到鎖執(zhí)行工作代碼

完整的測(cè)試用例

/**
 * @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("我是主服務(wù)開(kāi)始工作了");
                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("我是備用服務(wù)，我開(kāi)始工作了，估計(jì)主服務(wù)已經(jīng)掛了");
                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("我是備用服務(wù)，我開(kāi)始工作了，估計(jì)主服務(wù)已經(jīng)掛了");
                TimeUnit.SECONDS.sleep(1);
            }
        });
        TimeUnit.DAYS.sleep(1);
    }
}

上面測(cè)試用例模擬了一主兩備的高可用架構(gòu)。分別執(zhí)行l(wèi)ockTest1toMaster()、lockTest2toStandby()、lockTest3toStandby()服務(wù)，會(huì)發(fā)現(xiàn)只有一個(gè)服務(wù)會(huì)打印。然后手動(dòng)關(guān)閉這個(gè)服務(wù)，從服務(wù)馬上會(huì)接著打印。在關(guān)閉這個(gè)從服務(wù)，另外一個(gè)從服務(wù)就會(huì)接著打印。很好的模擬了主備故障切換的效果。

以上就是高可用架構(gòu)etcd選主故障主備秒級(jí)切換實(shí)現(xiàn)的詳細(xì)內(nèi)容，更多關(guān)于etcd主備故障秒級(jí)切換的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: