登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 资源区 › 代码 ›etcd 入门实战(3)-java 操作 etcd

etcd 入门实战(3)-java 操作 etcd

铵滔 4 天前
本文主要介绍使用 coreos 提供的 Java 客户端(jetcd)来操作 etcd，文中所使用到的软件版本：etcd 3.5.18、jetcd 0.7.7。
1、引入依赖
  1. <dependency>
  2.     <groupId>io.etcd</groupId>
  3.     jetcd-core</artifactId>
  4.     <version>0.7.7</version>
  5. </dependency>
复制代码
2、jetcd 使用

2.1、初始化客户端
  1. @Before
  2. public void before() {
  3.     client = Client.builder()
  4.             .endpoints("http://10.49.196.33:2379")
  5.             //.endpoints("http://10.49.196.30:2379", "http://10.49.196.31:2379", "http://10.49.196.33:2379")
  6.             .connectTimeout(Duration.of(10, ChronoUnit.SECONDS))
  7.             .build();
  8. }
复制代码
2.2、键值操作

A、新增/修改
  1. @Test
  2. public void kvPut() throws Exception {
  3.     KV kv = client.getKVClient();
  4.     ByteSequence key = ByteSequence.from("key2", StandardCharsets.UTF_8);
  5.     ByteSequence value = ByteSequence.from("value2", StandardCharsets.UTF_8);
  6.     CompletableFuture<PutResponse> completableFuture = kv.put(key, value);
  7.     log.info("completableFuture={}", completableFuture.get());
  8. }
复制代码
B、查询
  1. @Test
  2. public void kvGet() throws Exception {
  3.     KV kv = client.getKVClient();
  4.     ByteSequence key = ByteSequence.from("key1", StandardCharsets.UTF_8);
  5.     CompletableFuture<GetResponse> completableFuture = kv.get(key);
  6.     GetResponse getResponse = completableFuture.get();
  7.     if (getResponse.getCount() > 0) {
  8.         log.info("value={}", getResponse.getKvs().get(0).getValue());
  9.     }
  11.     key = ByteSequence.from("key", StandardCharsets.UTF_8);
  12.     GetOption getOption = GetOption.builder().isPrefix(true).build();
  13.     completableFuture = kv.get(key, getOption);//查询健以”key“开头的数据
  14.     for (KeyValue keyValue : completableFuture.get().getKvs()) {
  15.         log.info("key={},value={}", keyValue.getKey(), keyValue.getValue());
  16.     }
  17. }
复制代码
C、删除
  1. @Test
  2. public void kvDelete() throws Exception {
  3.     KV kv = client.getKVClient();
  4.     ByteSequence key = ByteSequence.from("key1", StandardCharsets.UTF_8);
  5.     CompletableFuture<DeleteResponse> completableFuture = kv.delete(key);
  6.     log.info("completableFuture={}", completableFuture.get());
  7. }
复制代码
2.3、监控
  1. @Test
  2. public void watch() throws Exception {
  3.     Watch watch = client.getWatchClient();
  4.     watch.watch(ByteSequence.from("key1", StandardCharsets.UTF_8), new Watch.Listener() {
  5.         @Override
  6.         public void onNext(WatchResponse response) {
  7.             List<WatchEvent> events = response.getEvents();
  8.             for (WatchEvent watchEvent : events) {
  9.                 log.info("eventType={},value={}", watchEvent.getEventType(), watchEvent.getKeyValue().getValue());
  10.             }
  11.         }
  13.         @Override
  14.         public void onError(Throwable throwable) {
  15.             log.error("发生异常:{}", throwable.getMessage());
  16.         }
  18.         @Override
  19.         public void onCompleted() {
  20.             log.info("complete");
  21.         }
  22.     });
  24.     Thread.sleep(1000 * 60 * 5);
  25. }
复制代码
2.4、租约
  1. @Test
  2. public void lease() throws Exception {
  3.     Lease lease = client.getLeaseClient();
  5.     //创建租约
  6.     LeaseGrantResponse leaseGrantResponse = lease.grant(10).get();
  7.     long leaseId = leaseGrantResponse.getID();
  9.     //租约与键值数据绑定
  10.     ByteSequence key = ByteSequence.from("lease-key", StandardCharsets.UTF_8);
  11.     ByteSequence value = ByteSequence.from("lease-value", StandardCharsets.UTF_8);
  12.     PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
  13.     client.getKVClient().put(key, value, putOption).get();
  15.     Thread.sleep(1000);
  17.     //查看租约剩余时间
  18.     LeaseOption leaseOption = LeaseOption.builder().withAttachedKeys().build();
  19.     LeaseTimeToLiveResponse leaseTimeToLiveResponse = lease.timeToLive(leaseId, leaseOption).get();
  20.     log.info("leaseTimeToLiveResponse={}", leaseTimeToLiveResponse);
  22.     //使租约一直有效
  23.     lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
  24.         @Override
  25.         public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
  26.             log.info("Lease keep-alive response:{}", leaseGrantResponse.getTTL());
  27.         }
  29.         @Override
  30.         public void onError(Throwable throwable) {
  31.             log.info("发生异常:{}", throwable.getMessage());
  32.         }
  34.         @Override
  35.         public void onCompleted() {
  36.             log.info("Complete");
  37.         }
  38.     });
  40.     Thread.sleep(1000 * 30);
  42.     //撤销租约
  43.     LeaseRevokeResponse leaseRevokeResponse = lease.revoke(leaseId).get();
  44.     log.info("leaseRevokeResponse={}", leaseRevokeResponse);
  45. }
复制代码
2.5、锁
  1. @Test
  2. public void lock() throws Exception {
  3.     ByteSequence lockName = ByteSequence.from("my-lock", StandardCharsets.UTF_8);
  4.     for (int i = 1; i <= 3; i++) {
  5.         new Thread(() -> {
  6.             try {
  7.                 LeaseGrantResponse leaseGrantResponse = client.getLeaseClient().grant(10).get();
  8.                 long leaseId = leaseGrantResponse.getID();
  10.                 Lock lock = client.getLockClient();
  11.                 //阻塞获取锁
  12.                 LockResponse lockResponse = lock.lock(lockName, leaseId).get();
  13.                 ByteSequence lockKey = lockResponse.getKey();
  14.                 log.info("{} 获得锁 {}", Thread.currentThread().getName(), lockResponse.getKey());
  15.                 Thread.sleep(3000);
  16.                 //释放锁，租约撤销或到期也会释放锁
  17.                 lock.unlock(lockKey).get();
  18.             } catch (Exception e) {
  19.                 log.error("", e);
  20.             }
  21.         }).start();
  22.     }
  24.     Thread.sleep(1000 * 20);
  25. }
复制代码
2.6、选举
  1. @Test
  2. public void election() throws Exception {
  3.     ByteSequence electionName = ByteSequence.from("electionName", StandardCharsets.UTF_8);
  4.     ByteSequence proposal1 = ByteSequence.from("proposal1", StandardCharsets.UTF_8);
  5.     ByteSequence proposal2 = ByteSequence.from("proposal2", StandardCharsets.UTF_8);
  6.     ByteSequence proposal3 = ByteSequence.from("proposal3", StandardCharsets.UTF_8);
  7.     ByteSequence[] proposals = new ByteSequence[]{proposal1, proposal2, proposal3};
  9.     for (ByteSequence proposal : proposals) {
  10.         new Thread(() -> {
  11.             try {
  12.                 Election election = client.getElectionClient();
  13.                 //监听选举事件(可选)
  14.                 election.observe(electionName, new Election.Listener() {
  15.                     @Override
  16.                     public void onNext(LeaderResponse leaderResponse) {
  17.                         log.info("proposal={},key={},value={}", proposal, leaderResponse.getKv().getKey(), leaderResponse.getKv().getValue());
  18.                     }
  20.                     @Override
  21.                     public void onError(Throwable throwable) {
  22.                         log.error("发生异常:{}", throwable.getMessage());
  23.                     }
  25.                     @Override
  26.                     public void onCompleted() {
  27.                         log.info("complete");
  28.                     }
  29.                 });
  31.                 LeaseGrantResponse leaseGrantResponse = client.getLeaseClient().grant(5).get();
  32.                 long leaseId = leaseGrantResponse.getID();
  33.                 client.getLeaseClient().keepAlive(leaseId, null);
  35.                 //获得领导权限或租约到期退出等待
  36.                 CampaignResponse campaignResponse = election.campaign(electionName, leaseId, proposal).get();
  37.                 LeaderKey leaderKey = campaignResponse.getLeader();
  38.                 log.info("{},获得领导权,{}", proposal, leaderKey.getKey());
  39.                 //获取领导者，如果是租约到期则改行代码会抛出异常NoLeaderException
  40.                 LeaderResponse leaderResponse = election.leader(electionName).get();
  41.                 log.info("领导者:{}", leaderResponse.getKv().getValue());
  42.                 //TODO:业务处理
  43.                 Thread.sleep(1000 * 6);
  44.                 //释放领导权
  45.                 election.resign(leaderKey).get();
  46.                 client.getLeaseClient().revoke(leaseId);
  47.             } catch (Exception e) {
  48.                 log.error("", e);
  49.             }
  50.         }).start();
  51.     }
  53.     Thread.sleep(1000 * 30);
  54. }
复制代码
2.7、完整代码

1.gif
2.gif
  1. package com.abc.etcd;
  3. import io.etcd.jetcd.*;
  4. import io.etcd.jetcd.election.CampaignResponse;
  5. import io.etcd.jetcd.election.LeaderKey;
  6. import io.etcd.jetcd.election.LeaderResponse;
  7. import io.etcd.jetcd.kv.DeleteResponse;
  8. import io.etcd.jetcd.kv.GetResponse;
  9. import io.etcd.jetcd.kv.PutResponse;
  10. import io.etcd.jetcd.lease.LeaseGrantResponse;
  11. import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
  12. import io.etcd.jetcd.lease.LeaseRevokeResponse;
  13. import io.etcd.jetcd.lease.LeaseTimeToLiveResponse;
  14. import io.etcd.jetcd.lock.LockResponse;
  15. import io.etcd.jetcd.options.GetOption;
  16. import io.etcd.jetcd.options.LeaseOption;
  17. import io.etcd.jetcd.options.PutOption;
  18. import io.etcd.jetcd.watch.WatchEvent;
  19. import io.etcd.jetcd.watch.WatchResponse;
  20. import io.grpc.stub.StreamObserver;
  21. import lombok.extern.slf4j.Slf4j;
  22. import org.junit.After;
  23. import org.junit.Before;
  24. import org.junit.Test;
  26. import java.nio.charset.StandardCharsets;
  27. import java.time.Duration;
  28. import java.time.temporal.ChronoUnit;
  29. import java.util.List;
  30. import java.util.concurrent.CompletableFuture;
  32. @Slf4j
  33. public class EtcdCase {
  34.     private Client client;
  36.     @Before
  37.     public void before() {
  38.         client = Client.builder()
  39.                 .endpoints("http://10.49.196.33:2379")
  40.                 //.endpoints("http://10.49.196.30:2379", "http://10.49.196.31:2379", "http://10.49.196.33:2379")
  41.                 .connectTimeout(Duration.of(10, ChronoUnit.SECONDS))
  42.                 .build();
  43.     }
  45.     @After
  46.     public void after() {
  47.         client.close();
  48.     }
  50.     @Test
  51.     public void kvPut() throws Exception {
  52.         KV kv = client.getKVClient();
  53.         ByteSequence key = ByteSequence.from("key2", StandardCharsets.UTF_8);
  54.         ByteSequence value = ByteSequence.from("value2", StandardCharsets.UTF_8);
  55.         CompletableFuture<PutResponse> completableFuture = kv.put(key, value);
  56.         log.info("completableFuture={}", completableFuture.get());
  57.     }
  59.     @Test
  60.     public void kvGet() throws Exception {
  61.         KV kv = client.getKVClient();
  62.         ByteSequence key = ByteSequence.from("key1", StandardCharsets.UTF_8);
  63.         CompletableFuture<GetResponse> completableFuture = kv.get(key);
  64.         GetResponse getResponse = completableFuture.get();
  65.         if (getResponse.getCount() > 0) {
  66.             log.info("value={}", getResponse.getKvs().get(0).getValue());
  67.         }
  69.         key = ByteSequence.from("key", StandardCharsets.UTF_8);
  70.         GetOption getOption = GetOption.builder().isPrefix(true).build();
  71.         completableFuture = kv.get(key, getOption);//查询健以”key“开头的数据
  72.         for (KeyValue keyValue : completableFuture.get().getKvs()) {
  73.             log.info("key={},value={}", keyValue.getKey(), keyValue.getValue());
  74.         }
  75.     }
  77.     @Test
  78.     public void kvDelete() throws Exception {
  79.         KV kv = client.getKVClient();
  80.         ByteSequence key = ByteSequence.from("key1", StandardCharsets.UTF_8);
  81.         CompletableFuture<DeleteResponse> completableFuture = kv.delete(key);
  82.         log.info("completableFuture={}", completableFuture.get());
  83.     }
  85.     @Test
  86.     public void watch() throws Exception {
  87.         Watch watch = client.getWatchClient();
  88.         watch.watch(ByteSequence.from("key1", StandardCharsets.UTF_8), new Watch.Listener() {
  89.             @Override
  90.             public void onNext(WatchResponse response) {
  91.                 List<WatchEvent> events = response.getEvents();
  92.                 for (WatchEvent watchEvent : events) {
  93.                     log.info("eventType={},value={}", watchEvent.getEventType(), watchEvent.getKeyValue().getValue());
  94.                 }
  95.             }
  97.             @Override
  98.             public void onError(Throwable throwable) {
  99.                 log.error("发生异常:{}", throwable.getMessage());
  100.             }
  102.             @Override
  103.             public void onCompleted() {
  104.                 log.info("complete");
  105.             }
  106.         });
  108.         Thread.sleep(1000 * 60 * 5);
  109.     }
  111.     @Test
  112.     public void lease() throws Exception {
  113.         Lease lease = client.getLeaseClient();
  115.         //创建租约
  116.         LeaseGrantResponse leaseGrantResponse = lease.grant(10).get();
  117.         long leaseId = leaseGrantResponse.getID();
  119.         //租约与键值数据绑定
  120.         ByteSequence key = ByteSequence.from("lease-key", StandardCharsets.UTF_8);
  121.         ByteSequence value = ByteSequence.from("lease-value", StandardCharsets.UTF_8);
  122.         PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
  123.         client.getKVClient().put(key, value, putOption).get();
  125.         Thread.sleep(1000);
  127.         //查看租约剩余时间
  128.         LeaseOption leaseOption = LeaseOption.builder().withAttachedKeys().build();
  129.         LeaseTimeToLiveResponse leaseTimeToLiveResponse = lease.timeToLive(leaseId, leaseOption).get();
  130.         log.info("leaseTimeToLiveResponse={}", leaseTimeToLiveResponse);
  132.         //使租约一直有效
  133.         lease.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
  134.             @Override
  135.             public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
  136.                 log.info("Lease keep-alive response:{}", leaseGrantResponse.getTTL());
  137.             }
  139.             @Override
  140.             public void onError(Throwable throwable) {
  141.                 log.info("发生异常:{}", throwable.getMessage());
  142.             }
  144.             @Override
  145.             public void onCompleted() {
  146.                 log.info("Complete");
  147.             }
  148.         });
  150.         Thread.sleep(1000 * 30);
  152.         //撤销租约
  153.         LeaseRevokeResponse leaseRevokeResponse = lease.revoke(leaseId).get();
  154.         log.info("leaseRevokeResponse={}", leaseRevokeResponse);
  155.     }
  157.     @Test
  158.     public void lock() throws Exception {
  159.         ByteSequence lockName = ByteSequence.from("my-lock", StandardCharsets.UTF_8);
  160.         for (int i = 1; i <= 3; i++) {
  161.             new Thread(() -> {
  162.                 try {
  163.                     LeaseGrantResponse leaseGrantResponse = client.getLeaseClient().grant(10).get();
  164.                     long leaseId = leaseGrantResponse.getID();
  166.                     Lock lock = client.getLockClient();
  167.                     //阻塞获取锁
  168.                     LockResponse lockResponse = lock.lock(lockName, leaseId).get();
  169.                     ByteSequence lockKey = lockResponse.getKey();
  170.                     log.info("{} 获得锁 {}", Thread.currentThread().getName(), lockResponse.getKey());
  171.                     Thread.sleep(3000);
  172.                     //释放锁，租约撤销或到期也会释放锁
  173.                     lock.unlock(lockKey).get();
  174.                 } catch (Exception e) {
  175.                     log.error("", e);
  176.                 }
  177.             }).start();
  178.         }
  180.         Thread.sleep(1000 * 20);
  181.     }
  183.     @Test
  184.     public void election() throws Exception {
  185.         ByteSequence electionName = ByteSequence.from("electionName", StandardCharsets.UTF_8);
  186.         ByteSequence proposal1 = ByteSequence.from("proposal1", StandardCharsets.UTF_8);
  187.         ByteSequence proposal2 = ByteSequence.from("proposal2", StandardCharsets.UTF_8);
  188.         ByteSequence proposal3 = ByteSequence.from("proposal3", StandardCharsets.UTF_8);
  189.         ByteSequence[] proposals = new ByteSequence[]{proposal1, proposal2, proposal3};
  191.         for (ByteSequence proposal : proposals) {
  192.             new Thread(() -> {
  193.                 try {
  194.                     Election election = client.getElectionClient();
  195.                     //监听选举事件(可选)
  196.                     election.observe(electionName, new Election.Listener() {
  197.                         @Override
  198.                         public void onNext(LeaderResponse leaderResponse) {
  199.                             log.info("proposal={},key={},value={}", proposal, leaderResponse.getKv().getKey(), leaderResponse.getKv().getValue());
  200.                         }
  202.                         @Override
  203.                         public void onError(Throwable throwable) {
  204.                             log.error("发生异常:{}", throwable.getMessage());
  205.                         }
  207.                         @Override
  208.                         public void onCompleted() {
  209.                             log.info("complete");
  210.                         }
  211.                     });
  213.                     LeaseGrantResponse leaseGrantResponse = client.getLeaseClient().grant(5).get();
  214.                     long leaseId = leaseGrantResponse.getID();
  215.                     client.getLeaseClient().keepAlive(leaseId, null);
  217.                     //获得领导权限或租约到期退出等待
  218.                     CampaignResponse campaignResponse = election.campaign(electionName, leaseId, proposal).get();
  219.                     LeaderKey leaderKey = campaignResponse.getLeader();
  220.                     log.info("{},获得领导权,{}", proposal, leaderKey.getKey());
  221.                     //获取领导者，如果是租约到期则改行代码会抛出异常NoLeaderException
  222.                     LeaderResponse leaderResponse = election.leader(electionName).get();
  223.                     log.info("领导者:{}", leaderResponse.getKv().getValue());
  224.                     //TODO:业务处理
  225.                     Thread.sleep(1000 * 6);
  226.                     //释放领导权
  227.                     election.resign(leaderKey).get();
  228.                     client.getLeaseClient().revoke(leaseId);
  229.                 } catch (Exception e) {
  230.                     log.error("", e);
  231.                 }
  232.             }).start();
  233.         }
  235.         Thread.sleep(1000 * 30);
  236.     }
  237.    
  238. }
复制代码
EtcdCase.java 
参考：
https://github.com/etcd-io/jetcd。
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
铵滔
4 天前

0

粉丝关注

9

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫