当前位置：首页 > article >正文

springboot 集成 etcd

article 2025/1/9 14:39:22

springboot 集成 etcd

往期内容

ETCD 简介
docker部署ETCD

前言

好久不见各位小伙伴们，上两期内容中，我们对于分布式kv存储中间件有了简单的认识，完成了docker-compose 部署etcd集群以及可视化工具 etcd Keeper，既然有了认识，完成了部署，那么当然要用起来啦

那么本期我们简单使用springboot 集成etcd 实现一些简单的数据操作

1-创建springboot工程

对于java开发的小伙伴来说，springboot项目的创建这块，我们不在过多赘述，简单贴一下我的版本依赖，我们直接跳转至下一环节

<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-集成依赖

简单的开发需要集成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 #你自己的宿主机
  port: 12379  #etcd的宿主机port 上一节中我的容器中的2379映射到了宿主机上的12379 所以配置12379

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

@Configuration
@Data
public class EtcdConfig {

    @Value("${etcd.host}")
    private String etcdHost;

    @Value("${etcd.port}")
    private int etcdPort;

}

创建etcd连接工厂

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

4-接入使用

创建service类进行简单的操作，这里简单的存储和redis一样，也是key-value的形式，不过多赘述

对于租约lease 而言类似一redis的键设置过期时间，过期时间内可以续期，过期时间之后etcd将会删除存储的内容

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;

@Service

public class EtcdService {

    @Resource
    private 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 lease
        CompletableFuture<PutResponse> putResponse = etcdClient.getKVClient().put(keyByte, valueByte, putOption);
       return putResponse;
    }




    /***
     * 续租一个已存在的租约
     * @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 expired
                System.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);
    }
}

创建简单的controller 进行请求尝试

//存 
@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";
    }

//查看租约状态
 @PostMapping("/viewLeaseDetail")
    public CompletableFuture<LeaseTimeToLiveResponse> getLeaseInfo(@RequestParam Long leaseId) throws ExecutionException, InterruptedException {
        return etcdService.getLeaseInfo(leaseId);
    }

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