上一篇文章简单介绍了ZooKeeper，讲了分布式中，每个微服务都会部署到多台服务器上，那服务之间的调用是怎么样的呢？如图：
1、集群A中的服务调用者如何发现集群B中的服务提供者呢？
2、集群A中的服务调用者如何选择集群B中的某一台服务提供者去调用呢？
3、集群B中某台机器下线，集群A怎么避免下次调用不在使用这台掉线的机器？
4、集群B提供的某个服务如何获知集群A中哪些机器正在消费该服务？
这篇文章写两个微服务，将两个服务部署到多台服务器中 ，通过将服务注册到ZooKeeper中，实现服务之间的调用。最终实现下面的ZooKeeper节点，然后通过服务节点下的地址，进行远程调用。
一、服务实现
一个获取订单的服务和顾客信息的服务，服务之间调用是通过订单服务查询此订单顾客的信息。 涉及的两个实体Order和Customer.
public class Custormer //顾客实体
{
public int Id { get; set; }
public string Name { get; set; }
public string Phone { get; set; }
}
public class Order //订单实体
{
public int Id { get; set; }
public int CustomerId { get; set; }
public string Goods { get; set; }
public string Address { get; set; } public Custormer Custormer;
}
订单实体中包含此订单顾客的引用。
创建一个订单微服务项目，实现获取订单列表的服务：
[Route("Order/GetOrders")]
public async Task<List<Order>> GetOrders()
{
List<Order> orders = new List<Order>();
Order order = null;
HttpClient client = new HttpClient();
for (var i = ; i < ; i++)
{
order = new Order();
order.Address = "浙江省杭州市拱墅区北部软件园" + i;
order.CustomerId = i;
order.Goods = "麻辣香锅" + i;
order.Id = i;
//这里需要调用获取顾客信息服务，获取顾客信息。这里先写null
order.Custormer = null;
orders.Add(order);
}
return orders;
}
创建一个顾客微服务项目，实现获取顾客信息的服务：
public class CustomerController : ControllerBase
{
[Route("Customer/GetCustomer")]
public Custormer GetCustomer(int Id)
{
return new Custormer() { Id=Id,Name="MicroHeart"+Id,Phone=""};
}
}
二、服务注册到ZooKeeper中
两个服务写完了，上篇讲的在服务启动的时候，需要将服务注册到ZooKeeper中，服务调用者启动的时候，将服务提供或者信息从注册中心下拉倒服务调用者本机缓存。当需要调用服务时，从本地缓存列表中找到服务提供者的地址列表，基于某种负载均衡策略(随机、轮询等)选择一台服务器发起远程调用。
在两个项目中的Startup构造函数中，调用下面方法，保证服务启动时就在ZooKeeper中注册服务。
public void InitZooKeeper()
{
var MyApp = "/MyApp";
//创建ZooKeeper 我就不在本地创建了 客户端和服务端都在本地的话，会造成误会
ZooKeeper zooKeeper = new ZooKeeper("118.24.96.212:2181", , new MyWatcher()); //创建 MyApp节点，数据为:MyAppData 权限控制为：开放 节点类型为：持久性节点
if (zooKeeper.existsAsync(MyApp) != null)
zooKeeper.createAsync(MyApp, Encoding.UTF8.GetBytes("MyAppData"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); //通过反射获取所有Controller下的方法，在获取方法上的Route特性，通过特性设置ZooKeeper节点。
Dictionary<string, List<string>> serviceAndApiPaths = new Dictionary<string, List<string>>();
var types = System.Reflection.Assembly.GetExecutingAssembly().GetTypes();
foreach (var type in types)
{
if (type.BaseType == typeof(ControllerBase))
{
var methods = type.GetMethods();
foreach (var method in methods)
{
foreach (var customAttribute in method.CustomAttributes)
{
if (customAttribute.AttributeType == typeof(RouteAttribute))
{
var serviceName = type.Name.Replace("Controller", "Services");
if (!serviceAndApiPaths.Keys.Contains(serviceName))
{
List<string> apiPaths = new List<string>();
//因为Route的值带"/" 会导致ZooKeeper认为是节点符号，所以要转换一下
apiPaths.Add(customAttribute.ConstructorArguments[].ToString().Replace("/","-"));
serviceAndApiPaths.Add(serviceName, apiPaths);
}
else
serviceAndApiPaths[serviceName].Add(customAttribute.ConstructorArguments[].ToString().Replace("/", "-"));
}
}
}
}
} //将这些接口列表 放到MyApp节点下
foreach(var item in serviceAndApiPaths)
{
//创建 服务节点，为持久性节点
if (zooKeeper.existsAsync($@"{MyApp}/{item.Key}") != null)
zooKeeper.createAsync($@"{MyApp}/{item.Key}", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
foreach (var apiPath in item.Value)
{
//创建 Api节点，为持久性节点
if (zooKeeper.existsAsync($@"{MyApp}/{item.Key}/{apiPath}") != null)
zooKeeper.createAsync($@"{MyApp}/{item.Key}/{apiPath}", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); //创建 Ip+port 节点，为临时性节点(由于我本地 不能通过我局域网Ip地址访问，所以我写死127.0.0.1) 写成临时节点 是因为
//当这个客户端与服务端断开时，对应的节点自动消失了。
//IPAddress[] IPList = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName()).AddressList;
//string currentIp = IPList.Where(ip=>ip.AddressFamily==System.Net.Sockets.AddressFamily.InterNetwork).Last().ToString();
string currentIp = "127.0.0.1";
if (zooKeeper.existsAsync($@"{MyApp}/{item.Key}/{apiPath}/{currentIp}:{Configuration["Port"]}") != null)
zooKeeper.createAsync($@"{MyApp}/{item.Key}/{apiPath}/{currentIp}:{Configuration["Port"]}", null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
}
}
}
这里简单介绍一下其中使用到的ZooKeeperAPI。
创建ZooKeeper的构造函数：ZooKeeper(string connectstring, int sessionTimeout, Watcher watcher, bool canBeReadOnly = false);
connectstring：ZooKeeper服务的地址和端口
sessionTimeout：连接超时时间，毫秒
watcher：观察者，相当于一个触发器，自己实现process方法
canBeReadOnly ：是否是只读权限
创建节点：Task<string> createAsync(string path, byte[] data, List<ACL> acl, CreateMode createMode);
path：节点路径 必须以“/”开头
data：节点的数据，数据大小不建议超过2M，数据格式为字节数组。
acl：权限相关
createMode：节点的类型(上篇文章讲到的四种类型 持久型节点、持久有序型节点、临时型节点、临时有序型节点)
获取子节点：Task<ChildrenResult> getChildrenAsync(string path, Watcher watcher);
path：节点路径 必须以“/”开头
watcher：：观察者，相当于一个触发器
上面的代码中服务的端口我没有写死，是通过获取appsettings.json文件中的Port参数值设置。配置文件和Program中的代码如下。我设置顾客服务端口为5000，订单服务端口为5100
{
"Logging": {
"LogLevel": {
"Default": "Warning"
}
},
"Port": "",
"AllowedHosts": "*"
}
public static void Main(string[] args)
{
//获取配置
var config = new ConfigurationBuilder()
//需要先设置路径 然后在路径中找到json文件
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile($"appsettings.json", true, true)
.Build(); //设置启动地址和端口号
CreateWebHostBuilder(args)
.UseUrls("http://127.0.0.1:" + config["Port"])
.UseConfiguration(config)
.Build()
.Run();
}
三、启动服务
这里介绍一个工具ZooInspector，下载地址，通过它可以很容易查看ZooKeeper里面的内容。
通过命令启动两个服务，通过ZooInspector看到ZooKeeper结构如下：
如果你关闭一个服务窗口，那对应的服务下面的IP列表就会消失，因为这个节点是临时节点。
现在我们已经实现了，服务的注册，现在可以回头来继续写刚才还没有完成的订单调用。需改获取订单列表里代码如下：
public async Task<List<Order>> GetOrders()
{
List<Order> orders = new List<Order>(); Order order = null;
HttpClient client = new HttpClient();
for (var i = ; i < ; i++)
{
order = new Order();
order.Address = "浙江省杭州市拱墅区北部软件园" + i;
order.CustomerId = i;
order.Goods = "麻辣香锅" + i;
order.Id = i;
//连接ZooKeeper
ZooKeeper zooKeeper = new ZooKeeper("118.24.96.212:2181", , new MyWatcher());
ChildrenResult childrenResult = null;
if (await zooKeeper.existsAsync("/MyApp/CustomerServices/Customer-GetCustomer") != null)
//获取所有顾客信息服务的地址
childrenResult = await zooKeeper.getChildrenAsync("/MyApp/CustomerServices/Customer-GetCustomer"); //生成一个随机数
Random random = new Random();
var num = random.Next(, childrenResult.Children.Count - ); //通过随机数 获取服务列表中随机的一个地址
var url = $@"http://{childrenResult.Children[num]}/Customer/GetCustomer?Id=" + order.CustomerId;
//调用顾客服务
var result = await client.GetAsync(url); Custormer custormer = JsonConvert.DeserializeObject<Custormer>(result.Content.ReadAsStringAsync().Result);
order.Custormer = custormer; orders.Add(order);
}
return orders;
}
不过刚才我们仅仅部署了服务到一台服务器中，现在我们改变端口配置，通过命令启动多个实例。如文章的第二个图，顾客服务配置了3台服务器(其实都在同一电脑)，订单服务也配置了3台服务器，当订单服务调用时，会从中随机选一台服务器，进行调用。
通过Postman调用接口，结果中返回了订单列表，且订单中包含顾客信息。
