搜索

Zookeeper 典型应用场景场景

程序IT圈

共 16246字，需浏览 33分钟

 · 2021-10-29

之前自己写了一些关于Zookeeper的基础知识，Zookeeper作为一种协调分布式应用高性能的调度服务，实际的应用场景也非常的广泛，这里主要通过几个例子来具体的说明Zookeeper在特定场景下的使用方式（下面的这些功能估计consul和etcd也能实现，以后学到了再说吧）。


2.1.一致性配置管理

我们在开发的时候，有时候需要获取一些公共的配置，比如数据库连接信息等，并且偶然可能需要更新配置。如果我们的服务器有N多台的话，那修改起来会特别的麻烦，并且还需要重新启动。这里Zookeeper就可以很方便的实现类似的功能。

2.1.1.思路


1. 将公共的配置存放在Zookeeper的节点中

2. 应用程序可以连接到Zookeeper中并对Zookeeper中配置节点进行读取或者修改（对于写操作可以进行权限验证设置），下面是具体的流程图：


2.1.2.事例

数据库配置信息一致性的维护
配置类：
public class CommonConfig implements Serializable{
  // 数据库连接配置
  private String dbUrl;
  private String username;
  private String password;
  private String driverClass;
  
  public CommonConfig() {}
  
  public CommonConfig(String dbUrl, String username, String password, String driverClass) {
    super();
    this.dbUrl = dbUrl;
    this.username = username;
    this.password = password;
    this.driverClass = driverClass;
  }
 
  public String getDbUrl() {
    return dbUrl;
  }
 
  public void setDbUrl(String dbUrl) {
    this.dbUrl = dbUrl;
  }
 
  public String getUsername() {
    return username;
  }
 
  public void setUsername(String username) {
    this.username = username;
  }
 
  public String getPassword() {
    return password;
  }
 
  public void setPassword(String password) {
    this.password = password;
  }
 
  public String getDriverClass() {
    return driverClass;
  }
 
  public void setDriverClass(String driverClass) {
    this.driverClass = driverClass;
  }
  
  @Override
  public String toString() {
    return "CommonConfig:{dbUrl:" + this.dbUrl +
        ", username:" + this.username + 
        ", password:" + this.password + 
        ", driverClass:" + this.driverClass + "}";
  }
}
配置管理中心
1. 获取本地配置信息
2. 修改配置，并同步
同步配置信息到Zookeeper服务器
public class ZkConfigMng {
  private String nodePath = "/commConfig";
  private CommonConfig commonConfig;
  private ZkClient zkClient;
  
  public CommonConfig initConfig(CommonConfig commonConfig) {
    if(commonConfig == null) {
      this.commonConfig = new CommonConfig("jdbc:mysql://127.0.0.1:3306/mydata?useUnicode=true&characterEncoding=utf-8",
          "root", "root", "com.mysql.jdbc.Driver"); 
    } else {
      this.commonConfig = commonConfig;
    }
    return this.commonConfig;
  }
  
  /**
   * 更新配置
   * 
   * @param commonConfig
   * @return
   */
  public CommonConfig update(CommonConfig commonConfig) {
    if(commonConfig != null) {
      this.commonConfig = commonConfig;
    }
    syncConfigToZookeeper();
    return this.commonConfig;
  }
  
  public void syncConfigToZookeeper() {
    if(zkClient == null) {
      zkClient = new ZkClient("127.0.0.1:2181");
    }
    if(!zkClient.exists(nodePath)) {
      zkClient.createPersistent(nodePath);
    }
    zkClient.writeData(nodePath, commonConfig);
  }
}
以上是提供者，下面我们需要一个客户端获取这些配置
public class ZkConfigClient implements Runnable {
  
  private String nodePath = "/commConfig";
  
  private CommonConfig commonConfig;
 
  @Override
  public void run() {
    ZkClient zkClient = new ZkClient(new ZkConnection("127.0.0.1:2181", 5000));
    while (!zkClient.exists(nodePath)) {
      System.out.println("配置节点不存在!");
      try {
        TimeUnit.SECONDS.sleep(1);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    // 获取节点
    commonConfig = (CommonConfig)zkClient.readData(nodePath);
    System.out.println(commonConfig.toString());
    zkClient.subscribeDataChanges(nodePath, new IZkDataListener() {
      
      @Override
      public void handleDataDeleted(String dataPath) throws Exception {
        if(dataPath.equals(nodePath)) {
          System.out.println("节点：" + dataPath + "被删除了！");
        }
      }
      
      @Override
      public void handleDataChange(String dataPath, Object data) throws Exception {
        if(dataPath.equals(nodePath)) {
          System.out.println("节点：" + dataPath + ", 数据：" + data + " - 更新");
          commonConfig = (CommonConfig) data;
        }
      }
    });
  }
 
}
下面启动Main函数

配置管理服务启动

public static void main(String[] args) throws InterruptedException {
    SpringApplication.run(ZookeeperApiDemoApplication.class, args);
    
    ZkConfigMng zkConfigMng = new ZkConfigMng();
    zkConfigMng.initConfig(null);
    zkConfigMng.syncConfigToZookeeper();
    TimeUnit.SECONDS.sleep(10);
    
    // 修改值
    zkConfigMng.update(new CommonConfig("jdbc:mysql://192.168.1.122:3306/mydata?useUnicode=true&characterEncoding=utf-8",
        "root", "wxh", "com.mysql.jdbc.Driver"));
  }
}
客户端启动：
public static void main(String[] args) throws InterruptedException {
    SpringApplication.run(ZookeeperApiDemoApplication.class, args);
 
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    // 模拟多个客户端获取配置
    executorService.submit(new ZkConfigClient());
    executorService.submit(new ZkConfigClient());
    executorService.submit(new ZkConfigClient());
  }
}
2.2.分布式锁
在我们日常的开发中，如果是单个进程中对共享资源的访问，我们只需要用synchronized或者lock就能实现互斥操作。但是对于跨进程、跨主机、跨网络的共享资源似乎就无能为力了。
2.1.1.思路
1. 首先zookeeper中我们可以创建一个/distributed_lock持久化节点
2. 然后再在/distributed_lock节点下创建自己的临时顺序节点，比如：/distributed_lock/task_00000000008
3. 获取所有的/distributed_lock下的所有子节点，并排序
4. 判读自己创建的节点是否最小值（第一位）
5. 如果是，则获取得到锁，执行自己的业务逻辑，最后删除这个临时节点。
6. 如果不是最小值，则需要监听自己创建节点前一位节点的数据变化，并阻塞。
7. 当前一位节点被删除时，我们需要通过递归来判断自己创建的节点是否在是最小的，如果是则执行5）；如果不是则执行6）（就是递归循环的判断）
下面是具体的流程图：

2.1.3.事例
public class DistributedLock {
  
  // 常亮
  static class Constant {
    private static final int SESSION_TIMEOUT = 10000;
    private static final String CONNECTION_STRING = "127.0.0.1:2181";
    private static final String LOCK_NODE = "/distributed_lock";
    private static final String CHILDREN_NODE = "/task_";
  }
  
  private ZkClient zkClient;
  
  public DistributedLock() {
    // 连接到Zookeeper
    zkClient = new ZkClient(new ZkConnection(Constant.CONNECTION_STRING));
    if(!zkClient.exists(Constant.LOCK_NODE)) {
      zkClient.create(Constant.LOCK_NODE, "分布式锁节点", CreateMode.PERSISTENT);
    }
  }
  
  public String getLock() {
    try {
      // 1。在Zookeeper指定节点下创建临时顺序节点
      String lockName = zkClient.createEphemeralSequential(Constant.LOCK_NODE + Constant.CHILDREN_NODE, "");
      // 尝试获取锁
      acquireLock(lockName);
      return lockName;
    } catch(Exception e) {
      e.printStackTrace();
    }
    
    return null;
  }
  
  /**
   * 获取锁
   * @throws InterruptedException 
   */
  public Boolean acquireLock(String lockName) throws InterruptedException {
    // 2.获取lock节点下的所有子节点
    List childrenList = zkClient.getChildren(Constant.LOCK_NODE);
    // 3.对子节点进行排序，获取最小值
    Collections.sort(childrenList, new Comparator() {
      @Override
      public int compare(String o1, String o2) {
        return Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
      }
      
    });
    // 4.判断当前创建的节点是否在第一位
    int lockPostion = childrenList.indexOf(lockName.split("/")[lockName.split("/").length - 1]);
    if(lockPostion < 0) {
      // 不存在该节点
      throw new ZkNodeExistsException("不存在的节点：" + lockName);
    } else if (lockPostion == 0) {
      // 获取到锁
      System.out.println("获取到锁：" + lockName);
      return true;
    } else if (lockPostion > 0) {
      // 未获取到锁，阻塞
      System.out.println("...... 未获取到锁，阻塞等待 。。。。。。");
      // 5.如果未获取得到锁，监听当前创建的节点前一位的节点
      final CountDownLatch latch = new CountDownLatch(1);
      IZkDataListener listener = new IZkDataListener() {
        
        @Override
        public void handleDataDeleted(String dataPath) throws Exception {
          // 6.前一个节点被删除,当不保证轮到自己
          System.out.println("。。。。。。前一个节点被删除 。。。。。。");
          acquireLock(lockName);
          latch.countDown();
        }
        
        @Override
        public void handleDataChange(String dataPath, Object data) throws Exception {
          // 不用理会
        }
      };
      try {
        zkClient.subscribeDataChanges(Constant.LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
        latch.await();
      } finally {
        zkClient.unsubscribeDataChanges(Constant.LOCK_NODE + "/" + childrenList.get(lockPostion - 1), listener);
      }
    }
    return false;
  }
  
  /**
   * 释放锁（删除节点）
   * 
   * @param lockName
   */
  public void releaseLock(String lockName) {
    zkClient.delete(lockName);
  }
  
  public void closeZkClient() {
    zkClient.close();
  }
}
 
@SpringBootApplication
public class ZookeeperDemoApplication {
 
  public static void main(String[] args) throws InterruptedException {
    SpringApplication.run(ZookeeperDemoApplication.class, args);
    
    DistributedLock lock = new DistributedLock();
    String lockName = lock.getLock();
    /** 
     * 执行我们的业务逻辑
     */
    if(lockName != null) {
      lock.releaseLock(lockName);
    }
    
    lock.closeZkClient();
  }
}
2.3.分布式队列
在日常使用中，特别是像生产者消费者模式中，经常会使用BlockingQueue来充当缓冲区的角色。但是在分布式系统中这种方式就不能使用BlockingQueue来实现了，但是Zookeeper可以实现。
2.1.1.思路

1. 首先利用Zookeeper中临时顺序节点的特点

2. 当生产者创建节点生产时，需要判断父节点下临时顺序子节点的个数，如果达到了上限，则阻塞等待；如果没有达到，就创建节点。

3. 当消费者获取节点时，如果父节点中不存在临时顺序子节点，则阻塞等待；如果有子节点，则获取执行自己的业务，执行完毕后删除该节点即可。

4. 获取时获取最小值，保证FIFO特性。

2.1.2.事例
这个是一个消费者对一个生产者，如果是多个消费者对多个生产者，对代码需要调整。
public interface AppConstant {
  static String ZK_CONNECT_STR = "127.0.0.1:2181";
  static String NODE_PATH = "/mailbox";
  static String CHILD_NODE_PATH = "/mail_";
  static int MAILBOX_SIZE = 10;
}
 
public class MailConsumer implements Runnable, AppConstant{
 
  private ZkClient zkClient;
  private Lock lock;
  private Condition condition;
  
  public MailConsumer() {
    lock = new ReentrantLock();
    condition = lock.newCondition();
    zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
    System.out.println("sucess connected to zookeeper server!");
    // 不存在就创建mailbox节点
    if(!zkClient.exists(NODE_PATH)) {
      zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
    }
  }
 
  @Override
  public void run() {
    IZkChildListener listener = new IZkChildListener() { 
      @Override
      public void handleChildChange(String parentPath, List currentChilds) throws Exception {
        System.out.println("Znode["+parentPath + "] size:" + currentChilds.size());
        // 还是要判断邮箱是否为空
        if(currentChilds.size() > 0) {
          // 唤醒等待的线程
          try {
            lock.lock();
            condition.signal();
          } catch (Exception e) {
            e.printStackTrace();
          } finally {
            lock.unlock();
          }
        }
      }
    };
    // 监视子节点的改变，不用放用while循环中，监听一次就行了，不需要重复绑定
    zkClient.subscribeChildChanges(NODE_PATH, listener);
    try {
      //循环随机发送邮件模拟真是情况
      while(true) {
        // 判断是否可以发送邮件
        checkMailReceive();
        // 接受邮件
        List mailList = zkClient.getChildren(NODE_PATH);
        // 如果mailsize==0,也没有关系；可以直接循环获取就行了
        if(mailList.size() > 0) {
          Collections.sort(mailList, new Comparator() {
            @Override
            public int compare(String o1, String o2) {
              return Integer.parseInt(o1.split("_")[1]) - Integer.parseInt(o2.split("_")[1]);
            }
          });
          // 模拟邮件处理(0-1S)
          TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
          zkClient.delete(NODE_PATH + "/" + mailList.get(0));
          System.out.println("mail has been received:" + NODE_PATH + "/" + mailList.get(0));
        }
      } 
    }catch (Exception e) {
      e.printStackTrace();
    } finally {
      zkClient.unsubscribeChildChanges(NODE_PATH, listener);
    }
  }
 
  private void checkMailReceive() {
    try {
      lock.lock();
      // 判断邮箱是为空
      List mailList = zkClient.getChildren(NODE_PATH);
      System.out.println("mailbox size: " + mailList.size());
      if(mailList.size() == 0) {
        // 邮箱为空，阻塞消费者，直到邮箱有邮件
        System.out.println("mailbox is empty, please wait 。。。");
        condition.await();
        // checkMailReceive();
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
  }
}
 
public class MailProducer implements Runnable, AppConstant{
  
  private ZkClient zkClient;
  private Lock lock;
  private Condition condition;
  
  /**
   * 初始化状态
   */
  public MailProducer() {
    lock = new ReentrantLock();
    condition = lock.newCondition();
    zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
    System.out.println("sucess connected to zookeeper server!");
    // 不存在就创建mailbox节点
    if(!zkClient.exists(NODE_PATH)) {
      zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
    }
  }
 
  @Override
  public void run() {
    IZkChildListener listener = new IZkChildListener() { 
      @Override
      public void handleChildChange(String parentPath, List currentChilds) throws Exception {
        System.out.println("Znode["+parentPath + "] size:" + currentChilds.size());
        // 还是要判断邮箱是否已满
        if(currentChilds.size() < MAILBOX_SIZE) {
          // 唤醒等待的线程
          try {
            lock.lock();
            condition.signal();
          } catch (Exception e) {
            e.printStackTrace();
          } finally {
            lock.unlock();
          }
        }
      }
    };
    // 监视子节点的改变，不用放用while循环中，监听一次就行了，不需要重复绑定
    zkClient.subscribeChildChanges(NODE_PATH, listener);
    try {
      //循环随机发送邮件模拟真是情况
      while(true) {
        // 判断是否可以发送邮件
        checkMailSend();
        // 发送邮件
        String cretePath = zkClient.createEphemeralSequential(NODE_PATH + CHILD_NODE_PATH, "your mail");
        System.out.println("your mail has been send:" + cretePath);
        // 模拟随机间隔的发送邮件(0-10S)
        TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
      } 
    }catch (Exception e) {
      e.printStackTrace();
    } finally {
      zkClient.unsubscribeChildChanges(NODE_PATH, listener);
    }
  }
  
  private void checkMailSend() {
    try {
      lock.lock();
      // 判断邮箱是否已满
      List mailList = zkClient.getChildren(NODE_PATH);
      System.out.println("mailbox size: " + mailList.size());
      if(mailList.size() >= MAILBOX_SIZE) {
        // 邮箱已满，阻塞生产者，直到邮箱有空间
        System.out.println("mailbox is full, please wait 。。。");
        condition.await();
        checkMailSend();
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
  }
}
2.4.均衡负载
首先我们需要简单的理解分布式和集群，通俗点说：分布式就是将一个系统拆分到多个独立运行的应用中（有可能在同一台主机也有可能在不同的主机上），集群就是将单个独立的应用复制多分放在不同的主机上来减轻服务器的压力。而Zookeeper不仅仅可以作为分布式集群的服务注册调度中心（例如dubbo），也可以实现集群的负载均衡。
2.4.1.思路
1. 首先我们要理解，如果是一个集群，那么他就会有多台主机。所以，他在Zookeeper中信息的存在应该是如下所示：

2. 如上的结构，当服务调用方调用服务时，就可以根据特定的均衡负载算法来实现对服务的调用（调用前需要监听/service/serviceXXX节点，以更新列表数据）
2.4.2.事例
/**
 * 服务提供者
 * 
 * @author Administrator
 *
 */
public class ServiceProvider {
  // 静态常量
  static String ZK_CONNECT_STR = "127.0.0.1:2181";
  static String NODE_PATH = "/service";
  static String SERIVCE_NAME = "/myService";
  
  private ZkClient zkClient;
  
  public ServiceProvider() {
    zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
    System.out.println("sucess connected to zookeeper server!");
    // 不存在就创建NODE_PATH节点
    if(!zkClient.exists(NODE_PATH)) {
      zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
    }
  }
  
  public void registryService(String localIp, Object obj) {
    if(!zkClient.exists(NODE_PATH + SERIVCE_NAME)) {
      zkClient.create(NODE_PATH + SERIVCE_NAME, "provider services list", CreateMode.PERSISTENT);
    }
    // 对自己的服务进行注册
    zkClient.createEphemeral(NODE_PATH + SERIVCE_NAME + "/" + localIp, obj);
    System.out.println("注册成功！[" + localIp + "]");
  } 
}
 
/**
 * 消费者，通过某种均衡负载算法选择某一个提供者
 * 
 * @author Administrator
 *
 */
public class ServiceConsumer {
  // 静态常量
  static String ZK_CONNECT_STR = "127.0.0.1:2181";
  static String NODE_PATH = "/service";
  static String SERIVCE_NAME = "/myService";
  
  private List serviceList = new ArrayList();
  
  private ZkClient zkClient;
  
  public ServiceConsumer() {
    zkClient = new ZkClient(new ZkConnection(ZK_CONNECT_STR));
    System.out.println("sucess connected to zookeeper server!");
    // 不存在就创建NODE_PATH节点
    if(!zkClient.exists(NODE_PATH)) {
      zkClient.create(NODE_PATH, "this is mailbox", CreateMode.PERSISTENT);
    }
  }
  
  /**
   * 订阅服务
   */
  public void subscribeSerivce() {
    serviceList = zkClient.getChildren(NODE_PATH + SERIVCE_NAME);
    zkClient.subscribeChildChanges(NODE_PATH + SERIVCE_NAME, new IZkChildListener() {
      @Override
      public void handleChildChange(String parentPath, List currentChilds) throws Exception {
        serviceList = currentChilds;
      }
    });
  }
  
  /**
   * 模拟调用服务
   */
  public void consume() {
    //负载均衡算法获取某台机器调用服务
    int index = new Random().nextInt(serviceList.size());
    System.out.println("调用[" + NODE_PATH + SERIVCE_NAME + "]服务：" + serviceList.get(index));
  }
}
3.总结
Zookeeper是一个功能非常强大的应用，除了上面几种应用外，还有命名服务、分布式协调通知等也是常用的场景。

【END】

欢迎关注Java公众号
浏览 15
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
【每周一课#06】MidJourney应用实战
#AI绘画#  #MJ#  #文生图#时间：4月24日周三 21:00课程大纲:1、关于AIGC：概念、发展历程、就业前景2、MJ基础认识：如何使用、底层逻辑、MJ与SD优缺点比较3、MJ基础功能介绍：任务指令、后缀参数、图生图、图生文、垫图、局部修改等4、MJ应用场景与变现方向
Python涨薪研究所
0
【每周一课#06】MidJourney 应用实战
#AI绘画#  #MJ#  #文生图#时间：4月24日周三 21:00课程大纲:1、关于AIGC：概念、发展历程、就业前景2、MJ基础认识：如何使用、底层逻辑、MJ与SD优缺点比较3、MJ基础功能介绍：任务指令、后缀参数、图生图、图生文、垫图、局部修改等4、MJ应用场景与变现方向
Python涨薪研究所
0
AI智能视觉检测技术在工业级测量领域的创新应用--AMB Tube-Q导管数字化测量系统
技术背景在深度学习算法出来之前，对于视觉算法来说，大致可以分为以下5个步骤：特征感知，图像预处理，特征提取，特征筛选，推理预测与识别。早期的机器学习中，占优势的统计机器学习群体中，对特征是不大关心的。深度学习是机器学习技术的一个方面，由人工神经网络提供支持。深度学习技术的工作原理是教机器通过实例学习
机器视觉
0
C#中using的使用的几种场景
    using在C#中随处可见，比如引入命名空间，这是也是每个C#的cs页面必须存在的，除了引入命名空间还有几种用法，下面将介绍这几种的使用。1、用于引入命名空间     这是最常见最基本的用法， 这样可以在程序中直
dotNET全栈开发
10
编辑选择：工业互联网标识解析应用优秀解决方案（四款）
随着工业互联网的快速发展，标识解析技术作为实现设备互联、数据互通及服务互操作的重要支撑，已然成为物理世界与数字世界之间的关键桥梁。它不仅推动着工业资源的高效配置和生产流程的持续优化，而且助力整个产业链的深度融合。但面对复杂且多变的工业环境，现有标识解析解决方案仍面临标准化、成熟度不一以及安全隐私方面
工业互联网世界
0
实战必备-数据清洗、数据存储、数据可视化-《AKShare-初阶-实战应用》
✔️课程介绍本课程主要给大家介绍 AKShare 的实战应用部分，主要包括对 AKShare 获取到的数据进行数据清洗，数据存储和数据可视化。课中会提供我们在进行数据处理中的最佳实践！课程目录✔️获取方式如下：  直接购买订阅： ¥39.8/年🔗扫码进入课程店铺：🔅懒得
数据科学实战
31
InfiniBand网络、HDR和IB在超算中的应用实践
InfiniBand（IB）是由InfiniBand贸易协会（IBTA）建立的先进计算机网络通信标准。它在高性能计算（HPC）中的广泛采用归功于它能够为网络传输提供卓越的吞吐量、带宽和低延迟。InfiniBand是计算系统内部和外部的关键数据连接。无论是通过直接链路还是通过网络交换机进行互连，Inf
架构师技术联盟
10
无任务学习及在机器人任务和运动规划中的应用
大数据文摘授权转载自中国人工智能学会作者：张宪琦，范晓鹏摘 要：本文提出了无任务学习的方法，阐述了其与现有方法（包括自监督学习、迁移学习、模仿学习、强化学习）的区别与联系；然后，介绍了无任务学习在机器人任务和运动规划领域的应用，并分析了无任务学习在该领域的优势和主要研究难点。最后，对无任务学习在机器
大数据文摘
0
进程监控神器Supervisor：让你的Python应用稳如老狗！
△△请给“Python猫”加星标 ，以免错过文章推送作者：Ais137原文地址：https://juejin.cn/post/7354406980784373798版权声明：本文经原作者授权 Python猫 转载，为便于阅读，标题&排版略作编辑。版权归原作者所有，如需转载，请先联系
Python猫
30
Nginx 的 5 大应用场景，太实用了！
来源：blog.csdn.net/vbirdbest/article/details/80913319👉 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利全栈前后端分离博客项目 2.0 版本
小哈学Java
10
点赞
评论
收藏
分享

手机扫一扫分享

举报