第三方框架整合
API 文档
你当前正在浏览历史版本 Vert.x 4.5.35 的文档。 访问本页面的 最新版本。

Vert.x Consul 客户端

Consul 是一个用于在你的基础设施中发现和配置服务的工具。 Vert.x 客户端允许应用程序通过阻塞和非阻塞 HTTP API 与 Consul 系统进行交互。

使用 Vert.x Consul 客户端

要使用本项目，请在构建描述文件的 dependencies 部分添加以下依赖项：

  • Maven（在你的 pom.xml 中）：

<dependency>
 <groupId>io.vertx</groupId>
 <artifactId>vertx-consul-client</artifactId>
 <version>4.5.35</version>
</dependency>

  • Gradle（在你的 build.gradle 文件中）：

compile 'io.vertx:vertx-consul-client:4.5.35'

创建客户端

只需使用工厂方法：

ConsulClient client = ConsulClient.create(vertx);

客户端也可以通过配置选项进行配置。

ConsulClientOptions options = new ConsulClientOptions()
  .setHost("consul.example.com");

ConsulClient client = ConsulClient.create(vertx, options);

Consul 客户端支持以下配置项：

host

Consul 主机地址。默认为 localhost

port

Consul HTTP API 端口。默认为 8500

timeout

设置超时时间（单位：毫秒）。如果在超时时间内请求未返回任何数据，则会将失败传递给处理器（Handler），并且请求将被关闭。

aclToken

ACL Token。提供该参数时，客户端在向 Consul 发起请求时会带上 "?token" 查询参数。未提供时，将使用空 Token（对应于 "anonymous" ACL 策略）。

dc

数据中心名称。提供该参数时，客户端在向 Consul 发起请求时会带上 "?dc" 查询参数。未提供时，将查询 Consul Agent 所在的数据中心。

ConsulClient 配置选项继承自 vertx-web-client 模块中的 WebClientOptions， 因此提供了大量可用配置。详情请参阅相关文档。

使用 API

客户端 API 由 ConsulClient 表示。该 API 与 Consul API 文档 中描述的 Consul HTTP API 非常相似。

阻塞查询

某些端点支持称为“阻塞查询”（blocking query）的特性。阻塞查询用于通过长轮询（long polling）等待潜在的变更。任何支持阻塞的端点都会提供一个表示所请求资源当前状态的唯一标识符（Index）。可以使用以下配置来执行阻塞查询：

index

表示客户端希望等待该 Index 之后发生的任何变更。

wait

指定阻塞请求的最大持续时间。最大限制为 10 分钟。

BlockingQueryOptions opts = new BlockingQueryOptions()
  .setIndex(lastIndex)
  .setWait("1m");

需要特别注意的是，阻塞请求返回*并不保证*一定发生了变更。 有可能是达到了超时时间，或者是发生了幂等写入但并未影响查询结果。

Key/Value 存储

KV 端点用于访问 Consul 的简易 Key-Value 存储，适用于存储服务配置或其他元数据。 支持以下端点操作：

  • 管理单个键的更新、单个键或键前缀的删除，以及单个键或键前缀的获取

  • 在单个原子事务中管理多个键的更新或获取

从存储中获取键值对

Consul 客户端可以返回特定键的值

consulClient.getValue("key").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("retrieved value: " + res.result().getValue());
    System.out.println("modify index: " + res.result().getModifyIndex());
  } else {
    res.cause().printStackTrace();
  }
});

……或者返回具有给定前缀的所有键值对

consulClient.getValues("prefix").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("modify index: " + res.result().getIndex());
    for (KeyValue kv : res.result().getList()) {
      System.out.println("retrieved value: " + kv.getValue());
    }
  } else {
    res.cause().printStackTrace();
  }
});

返回的 Key-Value 对象包含以下字段（参见 相关文档）：

createIndex

表示该条目创建时对应的内部 Index 值。

modifyIndex

最后一次修改该键的 Index

lockIndex

该键成功获取锁的次数

key

键名

flags

附加到该条目的标志。客户端可以根据应用程序的实际需要自行定义和使用

value

键值

session

持有该锁的 Session

modifyIndex 可用于阻塞查询：

BlockingQueryOptions opts = new BlockingQueryOptions()
  .setIndex(modifyIndex)
  .setWait("1m");

consulClient.getValueWithOptions("key", opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("retrieved value: " + res.result().getValue());
    System.out.println("new modify index: " + res.result().getModifyIndex());
  } else {
    res.cause().printStackTrace();
  }
});

向存储中存入键值对

consulClient.putValue("key", "value").onComplete(res -> {
  if (res.succeeded()) {
    String opResult = res.result() ? "success" : "fail";
    System.out.println("result of the operation: " + opResult);
  } else {
    res.cause().printStackTrace();
  }
});

PUT 请求也支持传入配置选项

KeyValueOptions opts = new KeyValueOptions()
  .setFlags(42)
  .setCasIndex(modifyIndex)
  .setAcquireSession("acquireSessionID")
  .setReleaseSession("releaseSessionID");

consulClient.putValueWithOptions("key", "value", opts).onComplete(res -> {
  if (res.succeeded()) {
    String opResult = res.result() ? "success" : "fail";
    System.out.println("result of the operation: " + opResult);
  } else {
    res.cause().printStackTrace();
  }
});

可在 PUT 请求中使用的查询选项列表：

flags

可用于指定介于 0 到 264-1 之间的无符号数值。客户端可以根据自身应用程序的需要来使用。

casIndex

该标志用于将 PUT 操作转为 Check-And-Set 操作。这是构建更复杂同步原语非常有用的基础构建块。如果 Index 为 0，Consul 仅在该键尚不存在时才会存入。如果 Index 非零，则仅在 Index 与该键当前的 ModifyIndex 匹配时才设置该键。

acquireSession

该标志用于将 PUT 操作转为获取锁的操作。这非常有用，因为它允许在 Consul 之上实现 Leader 选举。如果锁未被持有且 Session 有效，则除了更新键的内容外，还会递增 LockIndex 并设置键的 Session 值。即使某个键尚不存在，也可以对其进行锁获取。如果锁已经被给定的 Session 持有，则 LockIndex 不会递增，但键的内容会被更新。这允许当前的锁持有者更新键的内容，而无需先放弃锁再重新获取。

releaseSession

该标志用于将 PUT 操作转为释放锁的操作。当与 acquireSession 搭配使用时非常有用，允许客户端主动让出锁。这不会改变 LockIndex，但会清除键关联的 Session。必须由持有该锁的 Session 才能将其解锁。

事务

连接到 Consul 0.7 及更高版本时，客户端允许在单个原子事务中管理多个键的更新或获取。KV 是当前唯一可用的操作类型，不过在 Consul 的未来版本中可能会加入其他操作类型，以便与 Key-Value 操作混合使用（参见 相关文档）。

TxnRequest request = new TxnRequest()
  .addOperation(new TxnKVOperation().setKey("key1").setValue("value1").setType(TxnKVVerb.SET))
  .addOperation(new TxnKVOperation().setKey("key2").setValue("value2").setType(TxnKVVerb.SET));

consulClient.transaction(request).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("succeeded results: " + res.result().getResults().size());
    System.out.println("errors: " + res.result().getErrors().size());
  } else {
    res.cause().printStackTrace();
  }
});

删除键值对

最后，Consul 客户端支持从存储中删除键值对：

consulClient.deleteValue("key").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("complete");
  } else {
    res.cause().printStackTrace();
  }
});

……或者删除匹配相应键前缀的所有键值对

consulClient.deleteValues("prefix").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("complete");
  } else {
    res.cause().printStackTrace();
  }
});

服务

Service Discovery 的核心目标之一是提供可用服务的目录（Catalog）。 为此，Agent 提供了简明的服务定义格式，用于声明服务的可用性，并可选择将其与 Health Check 进行关联。

服务注册

服务定义必须包含 name，并可选择性提供 id、tags、address、port 和 checks。

ServiceOptions opts = new ServiceOptions()
  .setName("serviceName")
  .setId("serviceId")
  .setTags(Arrays.asList("tag1", "tag2"))
  .setCheckOptions(new CheckOptions().setTtl("10s"))
  .setAddress("10.0.0.1")
  .setPort(8048);
name

服务名称

id

若未提供，id 默认设为 name。每个节点上的所有服务必须具有唯一的 ID，因此如果名称可能冲突，则应提供唯一的 ID。

tags

对 Consul 而言不透明的标签值列表，可用于区分主从节点、不同版本或任何其他服务层级的标签。

address

用于指定服务特定的 IP 地址。默认情况下使用 Agent 的 IP 地址，因此该字段不必提供。

port

用于简化面向服务架构的配置；通过这种方式可以发现服务的地址和端口。

checks

关联的 Health Check

使用这些配置项在目录中注册服务：

consulClient.registerService(opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Service successfully registered");
  } else {
    res.cause().printStackTrace();
  }

});

服务发现

Consul 客户端支持获取提供某项服务的实际节点列表

consulClient.catalogServiceNodes("serviceName").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("found " + res.result().getList().size() + " services");
    System.out.println("consul state index: " + res.result().getIndex());
    for (Service service : res.result().getList()) {
      System.out.println("Service node: " + service.getNode());
      System.out.println("Service address: " + service.getAddress());
      System.out.println("Service port: " + service.getPort());
    }
  } else {
    res.cause().printStackTrace();
  }
});

获取该列表时还可以附带关联 Health Check 的状态。 查询结果可以按检查状态进行过滤。

consulClient.healthServiceNodes("serviceName", passingOnly).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("found " + res.result().getList().size() + " services");
    System.out.println("consul state index: " + res.result().getIndex());
    for (ServiceEntry entry : res.result().getList()) {
      System.out.println("Service node: " + entry.getNode());
      System.out.println("Service address: " + entry.getService().getAddress());
      System.out.println("Service port: " + entry.getService().getPort());
    }
  } else {
    res.cause().printStackTrace();
  }
});

服务查询还支持以下附加参数

ServiceQueryOptions queryOpts = new ServiceQueryOptions()
  .setTag("tag1")
  .setNear("_agent")
  .setBlockingOptions(new BlockingQueryOptions().setIndex(lastIndex));
tag

默认返回与该服务匹配的所有节点。可以通过 tag 查询参数按标签对列表进行过滤。

near

添加可选的带有节点名称的 near 参数后，将根据与该节点的预估往返时间（RTT）对节点列表进行升序排序。传入 near=_agent 时，将使用 Agent 所在的节点进行排序。

blockingOptions

阻塞查询选项

然后请求形式如下

consulClient.healthServiceNodesWithOptions("serviceName", passingOnly, queryOpts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("found " + res.result().getList().size() + " services");
  } else {
    res.cause().printStackTrace();
  }

});

注销服务

可以通过服务 ID 注销服务：

consulClient.deregisterService("serviceId").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Service successfully deregistered");
  } else {
    res.cause().printStackTrace();
  }
});

Health Check

Agent 的首要职责之一就是管理系统级和应用级的 Health Check。 如果 Health Check 与某个服务关联，则视为应用级检查。 若未与服务关联，则该检查负责监控整个节点的健康状态。

CheckOptions opts = new CheckOptions()
  .setTcp("localhost:4848")
  .setInterval("1s");

Consul 客户端支持的检查配置项列表如下：

id

检查 ID

name

检查名称

script

检查脚本的本地路径。同时还需设置检查间隔时间

http

用于检查的 HTTP 地址。同时还需设置检查间隔时间

ttl

检查的存活时间（TTL）

tcp

用于检查的 TCP 地址。同时还需设置检查间隔时间

interval

检查间隔时间，使用 Go 的时间格式（由十进制数字序列组成，每个数字均带有可选的小数部分和单位后缀，如 "300ms"、"-1.5h" 或 "2h45m"）。有效的时间单位为 "ns"、"us"（或 "µs"）、"ms"、"s"、"m"、"h"

notes

检查备注说明

serviceId

用于将注册的检查与 Agent 提供的现有服务进行关联的服务 ID。

deregisterAfter

注销超时时间。这是一个可选字段，其超时时间格式与 Interval 和 TTL 相同。如果检查关联到某个服务，且处于严重（critical）状态的时间超过了该配置值，则其关联的服务（以及该服务的所有关联检查）将被自动注销。最小超时时间为 1 分钟，且清理处于 critical 状态服务的进程每 30 秒运行一次，因此触发注销所需的时间可能会略长于配置的超时时间。通常应将此值配置为远大于该服务任何预期可恢复故障的超时时间。

status

用于指定 Health Check 初始状态的检查状态

Name 字段是必填项，同时 Script、HTTP、TCP 或 TTL 中必须指定其中之一。Script、TCP 和 HTTP 还需要设置 Interval。如果未提供 ID，则默认设为 Name。 每个 Agent 不允许出现重复的 ID 条目，因此可能需要提供显式的 ID。

consulClient.registerCheck(opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("check successfully registered");
  } else {
    res.cause().printStackTrace();
  }
});

Event

Consul 提供了向整个数据中心触发自定义用户 Event 的机制。 这些 Event 对 Consul 而言是不透明的，但可以用于构建脚本基础设施以进行自动化部署、重启服务或执行任何其他编排操作。

发送用户 Event 时只需指定其名称

consulClient.fireEvent("eventName").onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Event sent");
    System.out.println("id: " + res.result().getId());
  } else {
    res.cause().printStackTrace();
  }
});

此外还可以指定附加配置选项。

node

用于按节点名称过滤接收者的正则表达式

service

用于按服务过滤接收者的正则表达式

tag

用于按标签过滤接收者的正则表达式

payload

Event 的可选消息体。消息体内容对 Consul 而言是不透明的，并作为 Event 的 "payload"

EventOptions opts = new EventOptions()
  .setTag("tag")
  .setPayload("message");

consulClient.fireEventWithOptions("eventName", opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Event sent");
    System.out.println("id: " + res.result().getId());
  } else {
    res.cause().printStackTrace();
  }
});

Consul 客户端支持查询 Agent 所知晓的最新 Event。Event 使用 Gossip 协议进行广播，因此没有全局顺序，也不保证一定送达。Agent 仅缓冲最新的条目。当前缓冲区大小为 256，但此值未来可能会发生变化。

consulClient.listEvents().onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Consul index: " + res.result().getIndex());
    for(Event event: res.result().getList()) {
      System.out.println("Event id: " + event.getId());
      System.out.println("Event name: " + event.getName());
      System.out.println("Event payload: " + event.getPayload());
    }
  } else {
    res.cause().printStackTrace();
  }
});

Consul Index 可用于准备阻塞请求：

EventListOptions opts = new EventListOptions()
  .setName("eventName")
  .setBlockingOptions(new BlockingQueryOptions().setIndex(lastIndex));

consulClient.listEventsWithOptions(opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Consul index: " + res.result().getIndex());
    for(Event event: res.result().getList()) {
      System.out.println("Event id: " + event.getId());
    }
  } else {
    res.cause().printStackTrace();
  }
});

Session

Consul 提供了 Session 机制，可用于构建分布式锁。 Session 充当节点、Health Check 与 Key-Value 数据之间的绑定层。 在构建 Session 时，可以提供节点名称、Health Check 列表、行为模式、TTL 以及 Lock-Delay。

SessionOptions opts = new SessionOptions()
  .setNode("nodeId")
  .setBehavior(SessionBehavior.RELEASE);
lockDelay

可指定为带有 's' 秒数后缀的持续时间字符串。默认为 '15s'。

name

可用于为 Session 提供一个可读性较好的名称。

node

若指定，必须引用已注册的节点。默认使用 Agent 自身的节点名称。

checks

用于提供关联 Health Check 的列表。如果覆盖该列表，强烈建议包含默认的 serfHealth。

behavior

可设置为 release 或 delete。这控制 Session 失效时的行为。默认为 release，使得持有的任何锁都被释放。若改为 delete，则持有的任何锁都会被删除。delete 模式对于创建临时的 Key-Value 条目非常有用。

ttl

是一个持续时间字符串，与 LockDelay 类似，可以使用 's' 作为秒数后缀。如果指定，目前必须介于 10s 到 86400s 之间。提供后，若在 TTL 到期前未续订，Session 将会失效。

完整信息请参阅 Consul Sessions 内部原理。

新构建的 Session 会分配一个可用于标识它的 ID。 该 ID 可以与 KV 存储配合使用来获取锁：用于互斥的建议性锁机制（advisory mechanisms）。

consulClient.createSessionWithOptions(opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Session successfully created");
    System.out.println("id: " + res.result());
  } else {
    res.cause().printStackTrace();
  }
});

以及销毁 Session

consulClient.destroySession(sessionId).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Session successfully destroyed");
  } else {
    res.cause().printStackTrace();
  }
});

列出属于某个节点的 Session

consulClient.listNodeSessions("nodeId").onComplete(res -> {
  if (res.succeeded()) {
    for(Session session: res.result().getList()) {
      System.out.println("Session id: " + session.getId());
      System.out.println("Session node: " + session.getNode());
      System.out.println("Session create index: " + session.getCreateIndex());
    }
  } else {
    res.cause().printStackTrace();
  }
});

所有读取 Session 的端点都支持阻塞查询和所有一致性模式。

BlockingQueryOptions blockingOpts = new BlockingQueryOptions()
  .setIndex(lastIndex);

consulClient.listSessionsWithOptions(blockingOpts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Found " + res.result().getList().size() + " sessions");
  } else {
    res.cause().printStackTrace();
  }
});

数据中心中的节点

consulClient.catalogNodes().onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("found " + res.result().getList().size() + " nodes");
    System.out.println("consul state index " + res.result().getIndex());
  } else {
    res.cause().printStackTrace();
  }
});

该端点支持阻塞查询以及按与指定节点的距离进行排序

NodeQueryOptions opts = new NodeQueryOptions()
  .setNear("_agent")
  .setBlockingOptions(new BlockingQueryOptions().setIndex(lastIndex));

consulClient.catalogNodesWithOptions(opts).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("found " + res.result().getList().size() + " nodes");
  } else {
    res.cause().printStackTrace();
  }
});

Prepared Query

该端点用于创建、更新、销毁和执行 Prepared Query。 Prepared Query 允许你注册复杂的服务查询，之后通过其 ID 或名称来执行该查询，以获取提供指定服务的一组健康节点。这与 Consul 的 DNS 接口结合使用时尤为有用，因为与 DNS 暴露的有限入口相比，它允许丰富得多的查询方式。

创建 Prepared Query 时有许多可用参数。 完整详细信息请 查阅官方文档。

dc

指定要查询的数据中心。默认查询被请求 Agent 所在的数据中心。该参数作为查询参数在 URL 中指定。

name

指定一个可选的易记名称，可用于代替 ID 来执行查询。

session

指定现有 Session 的 ID。这提供了一种在给定 Session 失效时自动移除 Prepared Query 的途径。如果未指定，当不再需要 Prepared Query 时必须手动移除。

token

指定每次执行查询时使用的 ACL Token。这允许拥有更少甚至没有 ACL Token 的客户端执行查询，因此应谨慎使用。Token 本身只能由拥有 Management Token 的客户端查看。如果 Token 字段留空或省略，将使用客户端自身的 ACL Token 来判断其是否对被查询服务具有访问权限。如果客户端未提供 ACL Token，则将使用 Anonymous Token。

service

指定要查询的服务名称。这是必填字段。

failover

包含两个字段，两者都是可选的，用于决定在执行查询时如果本地数据中心中没有健康节点该如何处理。它允许通过极少配置即可使用其他数据中心中的节点。

nearestN

指定根据来自 WAN Gossip 池的网络坐标（Network Coordinates）估算的网络往返时间，将查询转发给最多 NearestN 个其他数据中心。通过处理查询的服务器到远程数据中心中服务器的往返时间中位数来决定优先级。

datacenters

指定一个固定的远程数据中心列表，用于在本地数据中心没有健康节点时转发查询。数据中心将按列表中给出的顺序进行查询。如果此选项与 NearestN 结合使用，则将先执行 NearestN 查询，随后按 Datacenters 给出的列表进行查询。在故障转移期间，即使某个数据中心既被 NearestN 选中又列在 Datacenters 中，也只会被查询一次。

onlyPassing

指定查询的 Health Check 过滤行为。如果设置为 false，结果将包括检查状态为 passing 以及 warning 的节点。如果设置为 true，则仅返回检查状态为 passing 的节点。

tags

指定用于过滤查询结果的服务标签列表。服务必须拥有所有必需的标签，且不能包含任何被排除的标签（以 ! 为前缀），才能通过标签过滤器。

nodeMeta

指定用户自定义的 Key-Value 对列表，用于将查询结果过滤为包含给定元数据值的节点。

dnsTtl

指定通过 DNS 提供查询结果时的 TTL 持续时间。如果指定了该值，它将优先于任何 Consul Agent 特定的配置。

templateType

查询类型，必须为 name_prefix_match。这意味着模板将应用于名称前缀与模板的 Name 字段相匹配的任何查询查找。在此示例中，针对 geo-db 的任何查询都将匹配该查询。查询模板使用最长前缀匹配来解析，因此可以定义通用的高层模板并在针对特定服务时进行覆盖。静态查询始终优先解析，因此它们也可以覆盖模板。

templateRegexp

一个可选的正则表达式，一旦选中此模板，该表达式将用于从整个名称中提取字段。在此示例中，正则表达式将 "-" 之后的第一项作为数据库名称，其余内容作为标签。关于该正则表达式的语法，请参阅 RE2 参考手册。

PreparedQueryDefinition def = new PreparedQueryDefinition()
  .setName("Query name")
  .setService("service-${match(1)}-${match(2)}")
  .setDcs(Arrays.asList("dc1", "dc42"))
  .setTemplateType("name_prefix_match")
  .setTemplateRegexp("^find_(.+?)_(.+?)$");

如果查询创建成功，将返回其 ID

consulClient.createPreparedQuery(def).onComplete(res -> {
  if (res.succeeded()) {
    String queryId = res.result();
    System.out.println("Query created: " + queryId);
  } else {
    res.cause().printStackTrace();
  }
});

可以通过 ID 执行 Prepared Query

consulClient.executePreparedQuery(id).onComplete(res -> {
  if (res.succeeded()) {
    PreparedQueryExecuteResponse response = res.result();
    System.out.println("Found " + response.getNodes().size() + " nodes");
  } else {
    res.cause().printStackTrace();
  }
});

或者通过必须与模板正则表达式匹配的查询字符串执行

consulClient.executePreparedQuery("find_1_2").onComplete(res -> {
  // 匹配模板正则表达式 "^find_(.+?)_(.+?)$"
  if (res.succeeded()) {
    PreparedQueryExecuteResponse response = res.result();
    System.out.println("Found " + response.getNodes().size() + " nodes");
  } else {
    res.cause().printStackTrace();
  }
});

最后，ConsulClient 允许你修改、获取或删除 Prepared Query

consulClient.deletePreparedQuery(query).onComplete(res -> {
  if (res.succeeded()) {
    System.out.println("Query deleted");
  } else {
    res.cause().printStackTrace();
  }
});

Watch

Watch 是一种指定数据视图（例如节点列表、KV 对、Health Check）并监控其更新的方法。 当检测到更新时，将调用带有 WatchResult 的 Handler。 例如，你可以监控 Health Check 的状态，并在检查处于 critical 状态时发出通知。

Watch.key("foo/bar", vertx)
  .setHandler(res -> {
    if (res.succeeded()) {
      System.out.println("value: " + res.nextResult().getValue());
    } else {
      res.cause().printStackTrace();
    }
  })
  .start();