springboot 集成 etcd

往期內(nèi)容

• ETCD 簡介
• docker部署ETCD

前言

好久不見各位小伙伴們，上兩期內(nèi)容中，我們對于分布式kv存儲中間件有了簡單的認(rèn)識，完成了docker-compose 部署etcd集群以及可視化工具 etcd Keeper，既然有了認(rèn)識，完成了部署，那么當(dāng)然要用起來啦

那么本期我們簡單使用springboot 集成etcd 實(shí)現(xiàn)一些簡單的數(shù)據(jù)操作

---

1-創(chuàng)建springboot工程

對于java開發(fā)的小伙伴來說，springboot項(xiàng)目的創(chuàng)建這塊，我們不在過多贅述，簡單貼一下我的版本依賴，我們直接跳轉(zhuǎn)至下一環(huán)節(jié)

<properties><java.version>1.8</java.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><spring-boot.version>2.6.13</spring-boot.version></properties>

2-集成依賴

簡單的開發(fā)需要集成ETCD的依賴

那么直接pom.xml

<!-- https://mvnrepository.com/artifact/io.etcd/jetcd-core 主要依賴--><dependency><groupId>io.etcd</groupId><artifactId>jetcd-core</artifactId><version>0.7.5</version></dependency><!-- (可選依賴)-><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.26</version><scope>provided</scope></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.21</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.21</version></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.52</version></dependency>

3-全局配置

application.yml

#etcd
etcd:host: http://XXXXXX #你自己的宿主機(jī)port: 12379  #etcd的宿主機(jī)port 上一節(jié)中我的容器中的2379映射到了宿主機(jī)上的12379 所以配置12379

增加全局配置文件讀取yml配置信息

@Configuration
@Data
public class EtcdConfig {@Value("${etcd.host}")private String etcdHost;@Value("${etcd.port}")private int etcdPort;}

創(chuàng)建etcd連接工廠

 @Beanpublic Client etcdFactory() {return Client.builder().endpoints(etcdHost + ":" + etcdPort).build();}

4-接入使用

創(chuàng)建service類進(jìn)行簡單的操作，這里簡單的存儲和redis一樣，也是key-value的形式，不過多贅述

對于租約lease 而言類似一redis的鍵設(shè)置過期時間，過期時間內(nèi)可以續(xù)期，過期時間之后etcd將會刪除存儲的內(nèi)容

package com.jerry.springetcd.service;import com.alibaba.fastjson2.JSON;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseTimeToLiveResponse;
import io.etcd.jetcd.options.LeaseOption;
import io.etcd.jetcd.options.PutOption;
import org.springframework.stereotype.Service;import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;@Servicepublic class EtcdService {@Resourceprivate Client etcdClient;/**** Put a key-value pair to etcd* @param key* @param value* @return*/public CompletableFuture<PutResponse> putValueWithOutLease(String key, String value) {ByteSequence keyByte = ByteSequence.from(key, StandardCharsets.UTF_8);ByteSequence valueByte = ByteSequence.from(value, StandardCharsets.UTF_8);return etcdClient.getKVClient().put(keyByte, valueByte);}/**** Get the value of a key from etcd* @param key* @return*/public CompletableFuture<String> getValue(String key) {ByteSequence keyByte = ByteSequence.from(key, StandardCharsets.UTF_8);CompletableFuture<GetResponse> getFuture = etcdClient.getKVClient().get(keyByte);return getFuture.thenApply(getResponse -> {if (getResponse.getKvs().isEmpty()) {return null;}return getResponse.getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);});}/**** Delete a key-value pair from etcd* @param key* @return*/public CompletableFuture<Void> deleteValue(String key) {ByteSequence keyByte = ByteSequence.from(key, StandardCharsets.UTF_8);return etcdClient.getKVClient().delete(keyByte).thenAccept(deleteResponse -> {});}/**** Put a key-value pair to etcd with a lease* @param key* @param value* @param leaseTime* @param timeout* @return* @throws ExecutionException* @throws InterruptedException* @throws TimeoutException*/public CompletableFuture<PutResponse> putValueWithLease(String key, String value, long leaseTime, long timeout) throws ExecutionException, InterruptedException, TimeoutException {ByteSequence keyByte = ByteSequence.from(key, StandardCharsets.UTF_8);ByteSequence valueByte = ByteSequence.from(value, StandardCharsets.UTF_8);Lease leaseClient = etcdClient.getLeaseClient();long leaseId = leaseClient.grant(leaseTime).get(timeout, TimeUnit.SECONDS).getID();System.out.println("Lease ID: " + leaseId);PutOption putOption = PutOption.newBuilder().withLeaseId(leaseId).build();// put value with leaseCompletableFuture<PutResponse> putResponse = etcdClient.getKVClient().put(keyByte, valueByte, putOption);return putResponse;}/**** 續(xù)租一個已存在的租約* @param leaseId* @return*/public CompletableFuture<Void> renewLease(long leaseId) {Lease leaseClient = etcdClient.getLeaseClient();return leaseClient.keepAliveOnce(leaseId).thenAccept(keepAliveResponse -> {if (keepAliveResponse.getTTL() == -1) {// lease has expiredSystem.out.println("Lease has expired");}else {System.out.println("Lease is renewed");}});}/**** 獲取租約的信息* @param leaseId* @return*/public CompletableFuture<LeaseTimeToLiveResponse> getLeaseInfo(long leaseId) throws ExecutionException, InterruptedException {Lease leaseClient = etcdClient.getLeaseClient();LeaseTimeToLiveResponse lTRes = leaseClient.timeToLive(leaseId, LeaseOption.newBuilder().withAttachedKeys().build()).get();return CompletableFuture.completedFuture(lTRes);}
}

創(chuàng)建簡單的controller 進(jìn)行請求嘗試

//存 
@PostMapping("/putValueWithOutLease")public CompletableFuture<PutResponse> putValueWithOutLease(@RequestParam String key, @RequestParam String value) {return etcdService.putValueWithOutLease(key, value);}

//取
@GetMapping("/get")public CompletableFuture<String> getValueWithOutLease(@RequestParam String key) {return etcdService.getValue(key);}

//刪除@DeleteMapping("/deleteValue")public CompletableFuture<Void> deleteValue(@RequestParam String key) {return etcdService.deleteValue(key);}

//帶租約的存儲@PostMapping("/putValueWithLease")public CompletableFuture<PutResponse> putValueWithLease(@RequestParam String key,@RequestParam String value,@RequestParam Long leaseTime,@RequestParam Long timeOut) throws ExecutionException, InterruptedException, TimeoutException {return etcdService.putValueWithLease(key, value,leaseTime, timeOut);}

//順序存儲@PutMapping("/writeWithOutLease")public String writeValue(@RequestParam int num) {logger.info("write value to etcd");StopWatch stopWatch = new StopWatch();stopWatch.start("write");for (int i = 0; i < num; i++) {etcdService.putValueWithOutLease("/test/" + i, String.valueOf(i));}stopWatch.stop();logger.info("write value to etcd, time: " + stopWatch.prettyPrint());return "success";}

//查看租約狀態(tài)@PostMapping("/viewLeaseDetail")public CompletableFuture<LeaseTimeToLiveResponse> getLeaseInfo(@RequestParam Long leaseId) throws ExecutionException, InterruptedException {return etcdService.getLeaseInfo(leaseId);}

//續(xù)租@PostMapping("/renewLease")public CompletableFuture<Void> renewLease(@RequestParam long leaseId) throws ExecutionException, InterruptedException, TimeoutException {return etcdService.renewLease(leaseId);}

5-接口模擬驗(yàn)證

使用apipost 進(jìn)行接口調(diào)用，查看執(zhí)行情況

• 不帶租約存儲
  

• 不帶租約查詢

• 手動刪除key

• 刪除后重新查詢，返回空值

• 帶租約存儲值

設(shè)置key=/acc/test val=hello etcd 租約時間為2min的數(shù)據(jù)存儲

• 查詢數(shù)據(jù)信息
  

---

demo gitee地址