Learn and share.
引言：为什么需要配置中心？
在微服务架构中，配置分散在数十甚至上百个服务实例中，传统本地配置文件管理面临配置漂移、环境不一致、敏感信息泄露等挑战。配置中心作为基础设施关键组件，核心解决：
1、集中管理：统一管控所有服务配置
2、动态生效：配置变更无需重启服务
3、环境隔离：开发、测试、生产环境完全隔离
4、安全合规：敏感信息加密存储与访问审计
5、高可用性：避免配置服务成为单点故障
在云原生时代，分布式系统的稳定运行离不开一个可靠的“数据中枢”——它需要存储集群配置、服务状态、元数据等关键信息，还要保证多节点间的数据一致、服务不中断。而etcd，正是这样一个被Kubernetes等核心云原生组件“依赖”的分布式键值存储系统，其核心定位清晰明确：作为分布式键值存储系统（Distributed Key-Value Store），它是Kubernetes的事实标准配置中心（Control Plane 数据存储），且基于Raft共识算法实现强一致性，成为支撑云原生生态的核心基石。它就像分布式系统的“大脑”，默默支撑着整个集群的协调与运转，却常常被隐藏在底层细节之后。
在分布式系统的世界里，有一个“隐形协调者”始终在默默发力——它就是ZooKeeper。无论是Hadoop、Kafka等大数据框架，还是Dubbo等微服务架构，都离不开它的支撑。很多开发者只知道它能实现分布式锁、服务注册，但很少深入了解其背后的设计逻辑：它的核心功能到底有哪些？独特特性是什么？又靠哪些架构和算法，实现了高可用、强一致性的承诺？今天这篇博客，就带你从零到一吃透ZooKeeper的核心逻辑。
在大数据时代，企业每天要处理数以亿计的实时数据，比如用户点击、传感器信号、交易记录等，传统消息系统在吞吐量、延迟、可靠性上逐渐力不从心。而Kafka作为一款开源分布式事件流处理平台，凭借“高吞吐、低延迟、可扩展、强可靠”的特性，成为全球超80%大数据场景的首选工具，更是大数据生态中日志收集、流式计算、数据同步的核心组件。
在我看来，中台并不是整个企业的数据集合，也不是什么具体的技术，甚至也不应该是一个单纯的IT概念。
中台本身应该是一家企业的一种组织架构方式及业务组织形态。在这种组织形态中，借鉴了IT技术中代码或模块复用的概念，实现了各个维度的能力高度整合与复用，而IT只是其中重要的一环，是一种重要的保障手段。
要建立中台，首先要改变的是企业组织架构，以及业务组织形态。干掉各个山头，从整体战略战术上考虑资源投放方式。保证各个业务环节的互联互通，加强能力复用，避免重复建设，实现企业高效运转。这无疑是一个一把手工程。
中台的最终目标，就是业务封装能力。通过将各领域下的通用功能进行封装，可以快速支持新的业务领域和业务模式。而投射到技术上，就是不要重新造轮子，通过复用大量已有服务，快速搭建系统，对新业务进行支持。
而我们说的建中台，最终都应该是业务中台。其他的各种技术类中台，都属于平台的范畴，服务于业务中台。
最近花了一些时间，看了一下OpenResty的东西，感觉用来做API网关的确很赞：
https://github.com/neohope/NeoDemosOpenResty
官网：
https://openresty.org/cn/
随着需要管理服务的增多，如何编排服务，成了一个很迫切的问题。本文就介绍几种常见的微服务编排方式：
1、Orchestration
这种方式，和BPM、ESB的思想很相似，实现方案多是同步的。
首先要有一个流程控制服务，该服务接收请求，依照业务逻辑规则，依次调用各个微服务，并最终完成处理逻辑。
这种方法的好处是，流程控制服务时时刻刻都知道每一笔业务究竟进行到了什么地步，监控业务成了相对简单的事情。
这种方法的坏处是，流程控制服务很容易控制了太多的业务逻辑，耦合度过高，变得臃肿，而各个微服务退化为单纯的增删改查，容易失去自身价值。
为了便于理解，您可以把控制服务看作BPM、ESB引擎，微服务为BPM、ESB的各种组件。
2、Choreography
这种方式，可以看作一种消息驱动模式，或者说是订阅发布模式，实现方案多是异步的。
每笔业务到来后，各个监听改事件的服务，会主动获取消息，处理，并可以按需发布自己的消息。
这种方法的好处是，耦合度低，每个服务都可以各司其职。
这种方法的坏处是，业务流程是通过订阅的方式来体现的，很难直接监控每笔业务的处理，因此需要增加相应的监控系统，来保证业务顺畅进行。
为了便于理解，您可以把不同队列看作不同种类的消息，微服务看作消息处理函数。
3、API网关
API网关，可以看作一种简单的接口聚合/拆分的方式。
每笔业务到来后，先到达网关，网关调用各微服务，并最终聚合/拆分需反馈的结果。
这种方法的好处是，对外接口相对稳定，可以利用LAN的带宽，弥补因特网的不足。
这种方法的坏处是，只适合业务逻辑较为简单的场景，业务逻辑过于复杂时，网关接口耦合度及复杂度会急剧升高，变得臃肿。
其实就是一个适配网关，比如对于Web端，可以一个页面同时发起几十个请求，而对于移动端，最好是一个页面就几个请求
。而采用API网关，后面的微服务可以是相同的。
以在同一台机器上的三个节点的集群为例：
1、在每个节点的zoo.cfg增加下面的配置（只给出了变动的部分）
dataDir=D:/Publish/ZooKeeper/node01 clientPort=2181 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
dataDir=D:/Publish/ZooKeeper/node02 clientPort=2182 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
dataDir=D:/Publish/ZooKeeper/node03 clientPort=2183 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
2、在每个dataDir增加一个myid文件，内容分别为1，2，3
3、现在可以启动哦
4、如果是在不同的服务器上，则dataDir、clientPort及2888:3888都不需要变动，localhost换成对应的计算机名称或ip即可。我这里是在一台电脑上运行的，所以要避免路径及端口冲突。
Queue实现了生产者——消费者模式。
1、QueueTest.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
/**
* Created by Hansen
*/
public class QueueTest implements Watcher {
static ZooKeeper zk = null;
static Object mutex;
private String root;
/**
* 构造函数
* @param hostPort
* @param name
*/
QueueTest(String hostPort, String name) {
this.root = name;
//创建连接
if (zk == null) {
try {
System.out.println("Starting ZK:");
zk = new ZooKeeper(hostPort, 30000, this);
mutex = new Object();
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
// 创建root节点
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
}
/**
* exists回调函数
* @param event 发生的事件
* @see org.apache.zookeeper.Watcher
*/
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
/**
* 添加任务队列
* @param i
* @return
*/
boolean produce(int i) throws KeeperException, InterruptedException {
String s = "element"+i;
zk.create(root + "/element", s.getBytes(Charset.forName("UTF-8")), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}
/**
* 从任务队列获取任务
* @return
* @throws KeeperException
* @throws InterruptedException
*/
int consume() throws KeeperException, InterruptedException {
Stat stat = null;
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() == 0) {
System.out.println("Going to wait");
mutex.wait();
} else {
//首先进行排序，找到id最小的任务编号
Integer min = Integer.MAX_VALUE;
for (String s : list) {
Integer tempValue = new Integer(s.substring(7));
if (tempValue < min) min = tempValue;
}
//从节点获取任务，处理，并删除节点
System.out.println("Processing task: " + root + "/element" + padLeft(min));
byte[] buff = zk.getData(root + "/element" + padLeft(min), false, stat);
System.out.println("The value in task is: " + new String(buff));
zk.delete(root + "/element" + padLeft(min), -1);
return min;
}
}
}
}
/**
* 格式化数字字符串
* @param num
*/
public static String padLeft(int num) {
return String.format("%010d", num);
}
/**
* 入口函数
* @param args
*/
public static void main(String args[]) {
String hostPort = "localhost:2181";
String root = "/neohope/queue";
int max = 10;
QueueTest q = new QueueTest(hostPort, root);
for (int i = 0; i < max; i++) {
try {
q.produce(i);
} catch (KeeperException e) {
} catch (InterruptedException e) {
}
}
for (int i = 0; i < max; i++) {
try {
int r = q.consume();
System.out.println("Item: " + r);
} catch (KeeperException ex) {
ex.printStackTrace();
break;
} catch (InterruptedException ex) {
ex.printStackTrace();
break;
}
}
}
}
2、尝试运行一下。
Barrier主要用于ZooKeeper中的同步。
1、BarrierTest.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
/**
* Created by Hansen
*/
public class BarrierTest implements Watcher, Runnable {
static ZooKeeper zk = null;
static Object mutex;
String root;
int size;
String name;
/**
* 构造函数
*
* @param hostPort
* @param root
* @param name
* @param size
*/
BarrierTest(String hostPort, String root, String name, int size) {
this.root = root;
this.name = name;
this.size = size;
//创建连接
if (zk == null) {
try {
System.out.println("Begin Starting ZK:");
zk = new ZooKeeper(hostPort, 30000, this);
mutex = new Object();
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
}
// 创建barrier节点
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
/**
* exists回调函数
* @param event 发生的事件
* @see org.apache.zookeeper.Watcher
*/
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
/**
* 新建节点，并等待其他节点被新建
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean enter() throws KeeperException, InterruptedException{
zk.create(root + "/" + name, "Hi".getBytes(Charset.forName("UTF-8")), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
System.out.println("Begin enter barier:" + name);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() < size) {
mutex.wait();
} else {
System.out.println("Finished enter barier:" + name);
return true;
}
}
}
}
/**
* 新建节点，并等待其他节点被新建
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean doSomeThing()
{
System.out.println("Begin doSomeThing:" + name);
//do your job here
System.out.println("Finished doSomeThing:" + name);
return true;
}
/**
* 删除自己的节点，并等待其他节点被删除
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean leave() throws KeeperException, InterruptedException{
zk.delete(root + "/" + name, -1);
System.out.println("Begin leave barier:" + name);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() > 0) {
mutex.wait();
} else {
System.out.println("Finished leave barier:" + name);
return true;
}
}
}
}
/**
* 线程函数，等待DataMonitor退出
* @see java.lang.Runnable
*/
@Override
public void run() {
//进入barrier
try {
boolean flag = this.enter();
if (!flag) System.out.println("Error when entering the barrier");
} catch (KeeperException ex) {
ex.printStackTrace();
} catch (InterruptedException ex) {
ex.printStackTrace();
}
//处理同步业务
try {
doSomeThing();
Thread.sleep(1000);
} catch (InterruptedException e) {
}
//离开barrier
try {
this.leave();
} catch (KeeperException ex) {
ex.printStackTrace();
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
/**
* 入口函数
* @param args
*/
public static void main(String args[]) throws IOException {
String hostPort = "localhost:2181";
String root = "/neohope/barrier";
try {
new Thread(new BarrierTest("127.0.0.1:2181", root,"001", 1)).start();
new Thread(new BarrierTest("127.0.0.1:2181", root,"002", 2)).start();
new Thread(new BarrierTest("127.0.0.1:2181", root,"003", 3)).start();
} catch (Exception e) {
e.printStackTrace();
}
System.in.read();
}
}
2、运行结果（由于Finished enter barier时，第一次同步已经结束了，所以是与Begin doSomeThing混在一起的）
Begin enter barier:001 Begin enter barier:003 Begin enter barier:002 Finished enter barier:001 Begin doSomeThing:001 Finished doSomeThing:001 Finished enter barier:002 Begin doSomeThing:002 Finished doSomeThing:002 Finished enter barier:003 Begin doSomeThing:003 Finished doSomeThing:003 Begin leave barier:002 Begin leave barier:001 Begin leave barier:003 Finished leave barier:002 Finished leave barier:003 Finished leave barier:001