AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Zookeeper（三）典型使用场景

2023-02-20 114
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
注册配置 MSE Nacos/ZooKeeper，118元/月
云原生网关 MSE Higress，422元/月
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
简介： Zookeeper（三）典型使用场景

分布式集群管理

使用Zookeeper实现分布式集群管理，功能如下：

  • 查看线上服务节点的资源使用情况
  • 服务离线通知
  • 服务资源超出阈值告警

架构设计：

monitor.png

具体实现：

  • 服务端收集服务状态信息发送给Zookeeper
  • 监控管理中心监听Zookeeper monitor节点下的子结点状态，做事件响应

代码：

服务端：使用agent技术监测服务端信息

  • pom
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <artifactId>my-zookeeper-demo</artifactId>
        <groupId>com.my</groupId>
        <version>1.0.0-SNAPSHOT</version>
    </parent>
    <artifactId>agent</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <packaging>jar</packaging>
    <inceptionYear>2020-Now</inceptionYear>
    <properties>
        <java.version>1.8</java.version>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
    </properties>
    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.8</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.11</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.9.3</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>2.2</version>
                <configuration>
                    <archive>
                        <manifestEntries>
                            <Project-name>monitor-collector</Project-name>
                            <Project-version>${project.version}</Project-version>
                            <Premain-Class>agent.MonitorAgent</Premain-Class>
                            <Can-Redefine-Classes>true</Can-Redefine-Classes>
                            <Can-Retransform-Classes>true</Can-Retransform-Classes>
                        </manifestEntries>
                    </archive>
                    <skip>true</skip>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.cbt.agent.Agent</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

dependency-reduced-pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <parent>
    <artifactId>my-zookeeper-demo</artifactId>
    <groupId>com.my</groupId>
    <version>1.0.0-SNAPSHOT</version>
  </parent>
  <modelVersion>4.0.0</modelVersion>
  <artifactId>agent</artifactId>
  <version>1.0.0-SNAPSHOT</version>
  <inceptionYear>2020-Now</inceptionYear>
  <build>
    <plugins>
      <plugin>
        <artifactId>maven-jar-plugin</artifactId>
        <version>2.2</version>
        <configuration>
          <archive>
            <manifestEntries>
              <Project-name>monitor-collector</Project-name>
              <Project-version>${project.version}</Project-version>
              <Premain-Class>agent.MonitorAgent</Premain-Class>
              <Can-Redefine-Classes>true</Can-Redefine-Classes>
              <Can-Retransform-Classes>true</Can-Retransform-Classes>
            </manifestEntries>
          </archive>
          <skip>true</skip>
        </configuration>
      </plugin>
      <plugin>
        <artifactId>maven-shade-plugin</artifactId>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <transformers>
                <transformer>
                  <mainClass>com.cbt.agent.Agent</mainClass>
                </transformer>
              </transformers>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
  <properties>
    <java.version>1.8</java.version>
    <maven.compiler.source>${java.version}</maven.compiler.source>
    <maven.compiler.target>${java.version}</maven.compiler.target>
  </properties>
</project>
  • domain
package agent.domain;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.experimental.Accessors;
/**
 * 操作系统信息
 *
 * @author Zijian Liao
 * @since 1.0.0
 */
@Data
@Accessors(chain = true)
@EqualsAndHashCode
public class OSInfo {
    public String ip;
    public Double cpu;
    public long usedMemorySize;
    public long usableMemorySize;
    public String pid;
    public long lastUpdateTime;
}


package agent.domain;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.lang.management.ThreadMXBean;
/**
 * 获取CPU相关信息
 */
public class CPUMonitorCalc {
    private static CPUMonitorCalc instance = new CPUMonitorCalc();
    private OperatingSystemMXBean osMxBean;
    private ThreadMXBean threadBean;
    private long preTime = System.nanoTime();
    private long preUsedTime = 0;
    private CPUMonitorCalc() {
        osMxBean = ManagementFactory.getOperatingSystemMXBean();
        threadBean = ManagementFactory.getThreadMXBean();
    }
    public static CPUMonitorCalc getInstance() {
        return instance;
    }
    public double getProcessCpu() {
        long totalTime = 0;
        for (long id : threadBean.getAllThreadIds()) {
            totalTime += threadBean.getThreadCpuTime(id);
        }
        long curtime = System.nanoTime();
        long usedTime = totalTime - preUsedTime;
        long totalPassedTime = curtime - preTime;
        preTime = curtime;
        preUsedTime = totalTime;
        return (((double) usedTime) / totalPassedTime / osMxBean.getAvailableProcessors()) * 100;
    }
}
  • agent
package agent;
import agent.domain.CPUMonitorCalc;
import agent.domain.OSInfo;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.I0Itec.zkclient.ZkClient;
import java.lang.instrument.Instrumentation;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryUsage;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;
/**
 * agent
 *
 * @author Zijian Liao
 * @since 1.0.0
 */
public class MonitorAgent {
    private static final String CLUSTER = "172.20.140.111:2181,172.20.140.220:2181,172.20.140.28:2181";
    private static final String ROOT_PATH = "/my-manager";
    private static final String SERVER_PATH = ROOT_PATH + "/server";
    public static void premain(String agentArgs, Instrumentation inst) {
        System.out.println("搜集服务信息插件启动中....");
        ZkClient zkClient = new ZkClient(CLUSTER);
        if (!zkClient.exists(ROOT_PATH)) {
            zkClient.createPersistent(ROOT_PATH);
        }
        //创建临时节点
        String nodePath = zkClient.createEphemeralSequential(SERVER_PATH, getOsInfo());
        //上报信息
        Thread thread = new Thread(()->{
            while(true){
                String osInfo = getOsInfo();
                zkClient.writeData(nodePath, osInfo);
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"zk-monitor");
        thread.setDaemon(true);
        thread.start();
        System.out.println("搜集服务信息插件已启动");
    }
    private static String getOsInfo() {
        OSInfo bean = new OSInfo();
        bean.lastUpdateTime = System.currentTimeMillis();
        bean.ip = getLocalIp();
        bean.cpu = CPUMonitorCalc.getInstance().getProcessCpu();
        MemoryUsage memoryUsag = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage();
        bean.usedMemorySize = memoryUsag.getUsed() / 1024 / 1024;
        bean.usableMemorySize = memoryUsag.getMax() / 1024 / 1024;
        bean.pid = ManagementFactory.getRuntimeMXBean().getName();
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.writeValueAsString(bean);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
    private static String getLocalIp() {
        InetAddress addr = null;
        try {
            addr = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
        return addr.getHostAddress();
    }
}

打包后运行服务端时，只要加上vm参数 -javaagent:D:\agent-1.0.0-SNAPSHOT.jar 即可挂载上监控插件

监控中心

@Controller
public class MonitorController implements InitializingBean {
    private static final String CLUSTER = "172.20.140.111:2181,172.20.140.220:2181,172.20.140.28:2181";
    private static final String ROOT_PATH = "/my-manager";
    private static final String SERVER_PATH = ROOT_PATH + "/server";
    ZkClient zkClient;
    @RequestMapping("/list")
    public String list(Model model){
        List<OSInfo> items = new ArrayList<>();
        List<String> children = zkClient.getChildren(ROOT_PATH);
        for (String child : children) {
            String path = ROOT_PATH + "/" + child;
            System.out.println(path);
            OSInfo osInfo = convert(zkClient.readData(path));
            System.out.println(osInfo);
            items.add(osInfo);
        }
        model.addAttribute("items", items);
        return "list";
    }
    ObjectMapper mapper = new ObjectMapper();
    private OSInfo convert(String json) {
        try {
            return mapper.readValue(json, OSInfo.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        zkClient = new ZkClient(CLUSTER);
    }
}

分布式注册中心

详见dubbo篇

分布式锁

使用Zookeeper实现一把分布式锁，具体功能

  • 同时只有一个服务能获取到锁
  • 未获取到服务的线程进入阻塞
  • 服务解锁后通知其他服务唤醒阻塞线程

具体实现：

  • 各个服务获取锁时在lock结点下创建临时节点
  • 获取到lock结点下所有子结点
  • 判断自己是不是最前的结点
  • 是则获取到锁，否则监听前一个结点，并阻塞自己
  • 服务解锁后，删除自己这个结点，Zookeeper通知下一个监听结点
  • 结点重新获取lock结点下所有子结点判断自己是不是最前的结点（检查一次，也可以不检查），唤醒阻塞线程，获取到了锁
  • 重复以上逻辑

代码：

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
public class ZookeeperLock {
    private static final String CLUSTER = "172.20.140.111:2181,172.20.140.220:2181,172.20.140.28:2181";
    private ZkClient zkClient;
    private static final String ROOT_PATH = "/my-lock";
    private static final Map<Lock,Thread> nodeMap = new HashMap<>();
    public ZookeeperLock() {
        zkClient = new ZkClient(CLUSTER);
        if (!zkClient.exists(ROOT_PATH)) {
            zkClient.createPersistent(ROOT_PATH);
        }
    }
    // 获取锁
    public Lock lock(String lockId, long timeout) {
        // 创建临时节点
        Lock lockNode = createLockNode(lockId);
        lockNode = tryActiveLock(lockNode);// 尝试激活锁
        if (!lockNode.isActive()) {
            nodeMap.put(lockNode,Thread.currentThread());
            System.out.println(Thread.currentThread().getName()+"park");
            LockSupport.parkNanos(timeout);
            System.out.println(Thread.currentThread().getName()+"unpark");
        }
        if (!lockNode.isActive()) {
            throw new RuntimeException(" lock  timeout");
        }
        return lockNode;
    }
    // 释放锁
    public void unlock(Lock lock) {
        if (lock.isActive()) {
            zkClient.delete(lock.getPath());
        }
    }
    // 尝试激活锁
    private Lock tryActiveLock(Lock lockNode) {
        // 获取根节点下面所有的子节点
        List<String> list = zkClient.getChildren(ROOT_PATH)
                .stream()
                .sorted()
                .map(p -> ROOT_PATH + "/" + p)
                .collect(Collectors.toList());      // 判断当前是否为最小节点
        String firstNodePath = list.get(0);
        // 最小节点是不是当前节点
        if (firstNodePath.equals(lockNode.getPath())) {
            lockNode.setActive(true);
        } else {
            String upNodePath = list.get(list.indexOf(lockNode.getPath()) - 1);
            zkClient.subscribeDataChanges(upNodePath, new IZkDataListener() {
                @Override
                public void handleDataChange(String dataPath, Object data) throws Exception {
                }
                @Override
                public void handleDataDeleted(String dataPath) throws Exception {
                    System.out.println("节点删除:" + dataPath);
                    Lock lock = tryActiveLock(lockNode);
                    if (lock.isActive()) {
                        LockSupport.unpark(nodeMap.get(lockNode)); // 释放了
                    }
                    zkClient.unsubscribeDataChanges(upNodePath, this);
                }
            });
        }
        return lockNode;
    }
    public Lock createLockNode(String lockId) {
        String nodePath = zkClient.createEphemeralSequential(ROOT_PATH + "/" + lockId, "w");
        return new Lock(lockId, nodePath);
    }
    public static class Lock {
        private String lockId;
        private String path;
        private boolean active;
        public Lock(String lockId, String path) {
            this.lockId = lockId;
            this.path = path;
        }
        public String getPath() {
            return path;
        }
        public void setPath(String path) {
            this.path = path;
        }
        public boolean isActive() {
            return active;
        }
        public void setActive(boolean active) {
            this.active = active;
        }
    }
    public static void main(String[] args) {
        ZookeeperLock lock = new ZookeeperLock();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        for(int i =0 ;i<10;i++){
            new Thread(()->{
                try {
                    countDownLatch.await();
                    System.out.println(Thread.currentThread().getName() + "准备获取锁");
                    Lock lockNode = lock.lock("lockKey", TimeUnit.SECONDS.toNanos(10));
                    System.out.println(Thread.currentThread().getName() + "获取到锁");
                    Thread.sleep(5000);
                    lock.unlock(lockNode);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },"Thread-"+i).start();
        }
        countDownLatch.countDown();
    }
}
文章标签：
微服务引擎
应用服务中间件
监控
Dubbo
关键词：
微服务引擎使用场景
微服务引擎典型
微服务引擎典型使用场景
相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作，您将了解并实现在线业务的微服务全链路灰度能力。
阿紫1
目录
相关文章
香吧香
|
5月前
|
消息中间件 存储 负载均衡
zookeeper 特点、使用场景及安装，配置文件解析
zookeeper 特点、使用场景及安装，配置文件解析
香吧香
78 2
赵广陆
|
分布式计算 Dubbo Hadoop
Zookeeper都有哪些使用场景？
Zookeeper都有哪些使用场景？
赵广陆
94 0
wljslmz
|
存储 算法 调度
Zookeeper是什么，它有什么特性与使用场景？
Zookeeper是什么，它有什么特性与使用场景？
wljslmz
1134 0
小小工匠
|
Java Apache
Apache ZooKeeper - Leader Election使用场景
Apache ZooKeeper - Leader Election使用场景
小小工匠
88 0
rr3paqhvizt6a
|
存储 安全 NoSQL
面试官：Zookeeper是什么，它有什么特性与使用场景？
本文介绍什么是Zookeeper以及它的特性与使用场景。
rr3paqhvizt6a
162 0
面试官：Zookeeper是什么，它有什么特性与使用场景？
图南(母晶晶)
|
消息中间件 分布式计算 资源调度
分布式的几件小事（九）zookeeper都有哪些使用场景
zookeeper都有哪些使用场景
图南(母晶晶)
86 0
huc_逆天
|
消息中间件 存储 设计模式
Zookeeper从入门到放弃之Zookeeper典型应用场景
Zookeeper从入门到放弃之Zookeeper典型应用场景
huc_逆天
195 0
Zookeeper从入门到放弃之Zookeeper典型应用场景
看山灬
|
SQL 存储 分布式计算
ZooKeeper介绍及典型使用场景
ZooKeeper（动物园管理员），顾名思义，是用来管理Hadoop（大象）、Hive（蜜蜂）、Pig（小猪）的管理员，同时Apache HBase、Apache Solr、LinkedIn Sensei等众多项目中都采用了ZooKeeper。
看山灬
185 0
徐雷frank
|
分布式计算 算法 网络协议
ZooKeeper分布式架构实战系列(01)：ZooKeeper概念、功能、架构、与使用场景和面试题
ZooKeeper是分布式协调服务，开源分布式架构领域不可或缺的技术，使用非常的广泛，比如常见的Hadoop大数据架构，阿里巴巴的Dubbo分布式架构，Spring Cloud微服务架构等都有涉及。BAT名企面试中也是经常作为考点。
徐雷frank
2530 0
1176112968452250
|
3月前
|
安全 应用服务中间件 API
微服务分布式系统架构之zookeeper与dubbo-2
微服务分布式系统架构之zookeeper与dubbo-2
1176112968452250
74 1

热门文章

最新文章

  • 1
    Dubbo的Zookeeper单机配置和Zookeeper集群配置
  • 2
    Spring Cloud Zookeeper 中心化配置文件
  • 3
    ZooKeeper 未授权访问漏洞
  • 4
    zookeeper集群搭建
  • 5
    阿里云微服务引擎 MSE 12 月产品动态
  • 6
    Dubbo与Zookeeper、SpringMVC整合和使用（负载均衡、容错）（转）
  • 7
    阿里云微服务引擎 MSE 2023 年 10 月产品动态
  • 8
    ZooKeeper 避坑实践：由于jute.maxbuffer 设置问题导致的集群不可用
  • 9
    zookeeper源码分析之leader选举
  • 10
    【ZooKeeper Notes 7】使用super身份对有权限的节点进行操作
  • 1
    低成本、高稳定性 ｜满帮集团 Eureka 和 ZooKeeper 的上云实践
    68053
  • 2
    低成本、高稳定性 | 满帮集团 Eureka 和 ZooKeeper 的上云实践
    57155
  • 3
    阿里云微服务引擎及 API 网关 2024 年 4 月产品动态
    384
  • 4
    阿里云微服务引擎 MSE 及 API 网关 2024 年 04 月产品动态
    260
  • 5
    深入理解ZooKeeper：分布式协调服务的核心与实践
    312
  • 6
    ZooKeeper数据模型你懂吗？
    96
  • 7
    【中间件】zookeeper的实现原理
    94
  • 8
    R语言参数自抽样法Bootstrap：估计MSE、经验功效、杰克刀Jackknife、非参数自抽样法可视化自测题
    64
  • 9
    Zookeeper笔记
    88
  • 10
    Kafka【部署 03】Zookeeper与Kafka自动部署脚本
    99

    • 相关课程

    更多
  • 基于MSE的大促场景流量入口防护最佳实践
  • Dubbo + ZooKeeper 的服务发现最佳实践
  • MSE微服务测试最佳实践 - 自动化回归
  • 基于Zookeeper、Dubbo构建互联网分布式基础架构
  • 大数据ZooKeeper快速入门
  • 分布式协调系统 Zookeeper 快速入门

    • 相关电子书

    更多
  • 《MSE 微服务网关》
  • 微服务引擎 MSE 治理中心重磅发布
  • 阿里云微服务引擎 MSE 2.0 线上发布

    • 相关实验场景

    更多
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 基于MSE实现微服务的全链路灰度
  • 基于MSE实现K8s集群内多服务的灰度发布
  • RocketMQ的常规运维实践应用
    • 下一篇
    使用阿里云接口（API）进行身份证实名认证