基於Etcd/zookeeper等kv儲存實現服務發現

public interface KVStore extends AutoCloseable {
/**
* Get node and its data, return null when it does not exist
*
* @param key node name
* @return Node info
* @throws Exception exception during operation
*/
Node get(String key) throws Exception;

/**
* List all child nodes and their data, return an empty list when
* the parent node does not exist or has no children
*
* @param prefix 父節點
* @return List of Node
* @throws Exception exception during operation
*/
List<Node> list(String prefix) throws Exception;

/**
* List the names of all child nodes,
* return an empty list when the parent node does not exist or has no children
*
* @param prefix 父節點
* @return List of node name
* @throws Exception exception during operation
*/
List<String> listKeys(String prefix) throws Exception;

/**
* Update the data to the node and persist the storage.
* <p>
* Update if the node already exists,
* but does not modify the type of the node (temporary/persistent)
*
* @param keyNode name (configuration directory)
* @param value Node data (registered service information)
* @return reversion
* @throws Exception exception during operation
*/
long put(String key, String value) throws Exception;

/**
* Update data to nodes and persist storage.
* <p>
* Update if the node already exists,
* but does not modify the type of the node (temporary/persistent)
*
* @param keynode name (configuration directory)
* @param valuenode data (registered service information)
* @param ephemeral is a temporary node
* @return reversion
* @throws Exception exception during operation
*/
long put(String key, String value, boolean ephemeral) throws Exception;

/**
* Delete the node
*
* @param key node name
* @return Number of nodes successfully deleted
* @throws Exception IllegalArgumentException: Node is not empty
* @throws Exception exception during operation
*
*/
long delete(String key) throws Exception;

/**
* Whether the node exists
* <p>
* 節點是否存在
*
* @param key node name 節點名稱
* @return true: exists 存在
* @throws Exception exception during operation
*/
boolean exists(String key) throws Exception;

/**
* Monitor changes to all child nodes (without parent nodes), continuous monitoring
* <p>
* 監視所有子節點的變化（不包含父節點），持續監聽
*
* @param keyName of parent node 父節點名稱
* @param consumer Callback when child nodes change 子節點變化時的回撥
* @throws Exception exception during operation
*/
void watchChildren(String key, BiConsumer<Event, Node> consumer) throws Exception;

/**
* Monitor the changes of all child nodes (excluding the parent node),
* and judge whether it needs to exit the monitoring according to the exit signal.
* <p>
* 監視所有子節點的變化（不包含父節點），根據退出訊號判斷是否需要退出監聽
*
* @param keyName of parent node 父節點名稱
* @param exitSignSupplier Provide an exit signal, otherwise it will always listen 提供退出訊號，否則一直監聽
* @param consumer子節點變化時的回撥
* @throws Exception exception during operation
*/
void watchChildren(String key, Supplier<Boolean> exitSignSupplier, BiConsumer<Event, Node> consumer) throws Exception;

/**
* Used to create a parent (prefix) and a persistent node
* <p>
* 用於建立父節點（prefix）, 且為持久節點
*
* @param parentNode Name of parent node 父節點名稱
* @throws Exception exception during operation
*/
void createParentNode(String parentNode) throws Exception;

public enum Event {
UPDATE,
DELETE,
UNRECOGNIZED
}

@Override
public void watchChildren(String key, Supplier<Boolean> exitSignSupplier, BiConsumer<Event, Node> consumer) throws InterruptedException {
ByteSequence storeKey =
Optional.ofNullable(key)
.map(ByteSequence::fromString)
.orElse(null);

try (Watch watch = client.getWatchClient();
Watch.Watcher watcher = watch.watch(storeKey,
WatchOption.newBuilder().withPrefix(storeKey).build())) {
while (!exitSignSupplier.get()) {
WatchResponse response = watcher.listen();
response.getEvents().forEach(watchEvent -> {
// 跳過根節點的變化
if (watchEvent.getKeyValue().getKey().equals(storeKey)) return;
Event event; // 此處的Event為上一節中定義的列舉型別
switch (watchEvent.getEventType()) {
case PUT:
event = Event.UPDATE;
break;
case DELETE:
event = Event.DELETE;
break;
default:
event = Event.UNRECOGNIZED;
}
KeyValue keyValue = watchEvent.getKeyValue();
Node info = kv2NodeInfo(keyValue);
consumer.accept(event, info);
});
}
}
}

private static final int EPHEMERAL_LEASE = 60; // seconds
private Client client;
private long leaseId;

private void initLease() throws ExecutionException, InterruptedException {
Lease lease = client.getLeaseClient();
LeaseGrantResponse response = lease.grant(EPHEMERAL_LEASE).get();
leaseId = response.getID();
lease.keepAlive(leaseId);
}

@Override
public long put(String key, String value, boolean ephemeral) throws ExecutionException, InterruptedException {
log.debug(String.format("put %s to %s", value, key));
KV kv = client.getKVClient();
ByteSequence storeKey =
Optional.ofNullable(key)
.map(ByteSequence::fromString)
.orElse(null);
ByteSequence storeValue =
Optional.ofNullable(value)
.map(ByteSequence::fromString)
.orElse(null);
PutOption.Builder builder = PutOption.newBuilder();
if (ephemeral) builder.withLeaseId(leaseId);

PutResponse response = kv.put(storeKey,
storeValue, builder.build()).get();
log.info(String.format("put key-value: key: %s, reversion: %s, has-prev: %s, ephemeral: %s",
key, response.getHeader().getRevision(), response.hasPrevKv(), ephemeral));
return response.getHeader().getRevision();
}

/**
* Watch change of node
*/
class SubWatcher implements Watcher {

private BiConsumer<cn.suniper.mesh.discovery.model.Event, Node> consumer;
private Supplier<Boolean> exitSignSupplier;
private volatile boolean stopWatch;

SubWatcher(BiConsumer<cn.suniper.mesh.discovery.model.Event, Node> consumer, Supplier<Boolean> exitSignSupplier) {
this.consumer = consumer;
this.exitSignSupplier = exitSignSupplier;
}

@Override
public void process(WatchedEvent event) {
if (exitSignSupplier.get()) {
log.info("sub-node: stop watch event: " + event);
return;
}

log.debug("sub-node: watch event: " + event);
cn.suniper.mesh.discovery.model.Event wrapEvent;
Node node;
switch (event.getType()) {

case NodeCreated:
case NodeDataChanged:
wrapEvent = cn.suniper.mesh.discovery.model.Event.UPDATE;
try {
node = get(event.getPath(), this);
log.debug(String.format("get node(%s) data: ", event.getPath()) + node);

} catch (Throwable throwable) {
log.warn("error occurred in watcher", throwable);
return;
}
break;
case NodeDeleted:
wrapEvent = cn.suniper.mesh.discovery.model.Event.DELETE;
node = new Node();
node.setKey(event.getPath());
break;
default:
wrapEvent = cn.suniper.mesh.discovery.model.Event.UNRECOGNIZED;
node = new Node();
node.setKey(event.getPath());
}

stopWatch = true;
consumer.accept(wrapEvent, node);
}

private SubWatcher activate() {
this.stopWatch = false;
return this;
}
}

class ChildrenWatcher implements Watcher {

private final Supplier<Boolean> DEFAULT_SUPPLIER = () -> false;

private ConcurrentHashMap<String, SubWatcher> childrenWatcher;
private BiConsumer<cn.suniper.mesh.discovery.model.Event, Node> consumer;
private String path;
private Supplier<Boolean> exitSignSupplier;

ChildrenWatcher(String path, BiConsumer<cn.suniper.mesh.discovery.model.Event, Node> consumer,
Supplier<Boolean> exitSignSupplier) {
this.consumer = consumer;
this.childrenWatcher = new ConcurrentHashMap<>();
this.path = path;
this.exitSignSupplier = exitSignSupplier == null ? DEFAULT_SUPPLIER : exitSignSupplier;
this.listAndWatch(false);
}

@Override
public void process(WatchedEvent event) {
log.debug("parent-node: watch event: " + event);

if (exitSignSupplier.get()) {
log.info("parent-node: stop watch event: " + event);
return;
}

switch (event.getType()) {
case NodeChildrenChanged:
this.listAndWatch(true);
return;
default:
log.debug("ignore event");
}

try {
List<String> res = listKeys(path, this);
log.debug(String.format("Sub nodes of %s: %s", path, res));
} catch (Exception e) {
log.warn("failed to keep watch node: " + path, e);
}

}

private void listAndWatch(boolean accept) {
try {
List<String> subList = listKeys(path, this);

log.debug(String.format("size of %s: %s", path, subList.size()));
for (String sub : subList) {
SubWatcher watcher = childrenWatcher.computeIfAbsent(sub, k -> {
log.debug("create new watcher for " + sub);
SubWatcher newWatcher = new SubWatcher(consumer, exitSignSupplier);
newWatcher.stopWatch = true;
return newWatcher;
});

if (!watcher.stopWatch) continue;

log.debug("activate watcher for " + sub);
watcher.activate();

if (accept) {
Node node = get(sub, watcher);
consumer.accept(cn.suniper.mesh.discovery.model.Event.UPDATE, node);
} else {
zooKeeper.exists(sub, watcher, null, null);
}
}

} catch (Exception e) {
log.warn("failed to list and watch node: " + path, e);
}
}
}

@Override
public long put(String key, String value, boolean ephemeral) throws Exception {
Stat stat = zooKeeper.exists(key, null);
if (stat != null) {
zooKeeper.setData(key, value.getBytes(), stat.getVersion());
} else {
CreateMode mode = ephemeral ? CreateMode.EPHEMERAL : CreateMode.PERSISTENT;
zooKeeper.create(key, value.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, mode);
}

stat = zooKeeper.exists(key, null);
log.info(String.format("current stat: %s", stat));
return stat.getMzxid();
}

@Override
public void createParentNode(String parentNode) throws Exception {
PathUtils.validatePath(parentNode);
if (parentNode.equals("/")) return;
File node = new File(parentNode);
String parent = node.getParent();
if (!exists(parent)) {
createParentNode(parent);
}
put(parentNode, DEFAULT_NODE_VALUE);
}

public class ProviderInfo {
// provider
private String name;
private String ip;
private int weight;
private int port;
private long version;
}

public class Application {
private List<String> registryUrlList; // 用作註冊中心的host:port列表
private ProviderInfo providerInfo;
private String serverGroup; // 所在的服務組
}

public void register(Application application) throws Exception {
String parentNode = String.join("/", Constants.STORE_ROOT, application.getServerGroup());
ProviderInfo providerInfo = Optional.ofNullable(application.getProviderInfo())
.orElse(new ProviderInfo());

if (providerInfo.getIp() == null) {
InetAddress address = HostUtil.getLocalIv4Address();
if (address != null) providerInfo.setIp(address.getHostAddress());
else throw new IllegalStateException("cannot get local host IP address");
}
if (providerInfo.getPort() == 0 || providerInfo.getName() == null) {
throw new IllegalArgumentException("please check you provider info");
}
String storeValue = String.format("%s/%s/%s",
providerInfo.getIp(),
providerInfo.getPort(),
providerInfo.getWeight());

String storeKey = String.join("/", parentNode, providerInfo.getName());

// check and create
store.createParentNode(parentNode);
long reversion = store.put(storeKey, storeValue, true);
log.info(String.format("registered server: `%s` in node: `%s`, reversion: %s",
storeKey, storeValue, reversion));
}

private class Holder implements BiConsumer<Event, Node> {
private Map<String, ProviderInfo> providerInfoMap; // 快取 node path 和 provider info的對應關係（這個Map必須是執行緒安全的）
private UpdateAction updateAction; // 執行服務列表更新動作的類，這裡知道它的行為即可
private final AtomicLong lastUpdated = new AtomicLong(System.currentTimeMillis()); // 記錄列表的最後更新時間

Holder(UpdateAction updateAction, Map<String, ProviderInfo> providerInfoMap) {
this.providerInfoMap = providerInfoMap;
this.updateAction = updateAction;
}

@Override
public void accept(Event event, Node node) {
// 將所有的事件分為 delete 和 update 兩類
String key = node.getKey();
switch (event) {
case DELETE:
// 子節點失效（被刪除）時，移除本地快取的Provider資訊
ProviderInfo removed = providerInfoMap.remove(key);
if (removed != null)
log.info(String.format("Service offline: %s - %s:%s", key, removed.getIp(), removed.getPort()));
else
log.info("Service offline: %s - no such provide cache");
update();
break;
case UPDATE:
// 子節點更新時（新增/更新屬性）
ProviderInfo oldInfo = providerInfoMap.computeIfAbsent(key, k -> new ProviderInfo());
MapperUtil.node2Provider(node, oldInfo); // 這個方法將node資訊轉為provider資訊，並更新舊的provider資訊（使用快取，減少物件建立的次數）
update();
break;
default:
log.info(String.format("unrecognized event: %s, key: %s", event, node.getKey()));

}
}

private void update() {
// 通知updateAction更新快取的服務列表
lastUpdated.set(System.currentTimeMillis());
updateAction.doUpdate();
}

}

List<RegisteredServer> obtainServerListByKvStore(Map<String, ProviderInfo> providerInfoMap) {
// providerInfoMap: 同上，快取 node path 和 provider info的對應關係（這個Map必須是執行緒安全的）
Set<String> newKeys = new HashSet<>();
Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey());

try {
List<Node> nodeInfoList = store.list(parentNode);
Stream<ProviderInfo> stream = nodeInfoList
.stream()
.peek(collectNewNodesToSet) // 將所有的node path新增到set中
.map(node -> providerMap.computeIfAbsent(
node.getKey(), k -> MapperUtil.node2Provider(node))); // 將節點轉換為ProviderInfo並put到providerMap

providerMap.keySet()
.parallelStream()
.filter(k -> !newKeys.contains(k)) // 過濾出providerMap中已經失效的provider資訊（store.list返回的列表中不再存在）
.forEach(k -> providerMap.remove(k)); // 移除這些已失效的provider資訊
return map2ServerList(stream); // 將剩下的資訊轉換為Server List

} catch (Exception e) {
log.warn(String.format("failed to obtain list of servers: %s", parentNode), e);
return Lists.newArrayList();
}
}

<dependency>
<groupId>com.netflix.ribbon</groupId>
<artifactId>ribbon-core</artifactId>
<version>${ribbon.version}</version>
</dependency>
<dependency>
<groupId>com.netflix.ribbon</groupId>
<artifactId>ribbon-loadbalancer</artifactId>
<version>${ribbon.version}</version>
</dependency>

public List<T> getInitialListOfServers();
public List<T> getUpdatedListOfServers();

public interface ServerListUpdater {

/**
* start the serverList updater with the given update action
* This call should be idempotent.
*
* @param updateAction
*/
void start(UpdateAction updateAction);

/**
* stop the serverList updater. This call should be idempotent
*/
void stop();

/**
* @return the last update timestamp as a {@link java.util.Date} string
*/
String getLastUpdate();

/**
* @return the number of ms that has elapsed since last update
*/
long getDurationSinceLastUpdateMs();

/**
* @return the number of update cycles missed, if valid
*/
int getNumberMissedCycles();

/**
* @return the number of threads used, if vaid
*/
int getCoreThreads();
}

 public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {
LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);

try {
return command.submit(
new ServerOperation<T>() {
@Override
public Observable<T> call(Server server) {
URI finalUri = reconstructURIWithServer(server, request.getUri());
S requestForServer = (S) request.replaceUri(finalUri);
try {
return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));
} 
catch (Exception e) {
return Observable.error(e);
}
}
})
.toBlocking()
.single();
} catch (Exception e) {
Throwable t = e.getCause();
if (t instanceof ClientException) {
throw (ClientException) t;
} else {
throw new ClientException(e);
}
} 
}