CuratorFramework实现zookeeper文件夹与文件的上传下载

今天药忘吃喽~ 2022-09-29 06:09 204阅读 0赞

原文URL： http://blog.csdn.net/cjuexuan/article/details/49612153

# 需求 #

--------------------

近期有一个solr项目，用户可以将自定义的配置文件与文件夹上传和下载到zookeeper中，原来一种做法是用solr中提供的ZkCLI实现，不过那个比较重，而且定制化也一般，所以我打算自己封装一个轻量级的Util工具，原来打算用原生的实现，不过后面发现了CuratorFramework这个工具，发现挺好用的，不过；里面貌似对文件夹的上传支持的不是很好，所以打算在那个基础上实现轻量级封装

# 思路 #

对于文件夹的上传和下载其实最简单的思路就是递归   
对于上传，拿到一个[操作系统][Link 1]的路径，判断是否是文件，是文件那么就直接上传，退出程序，否则先在zk中创建一个空目录，再将操作系统路径下列出所有的子节点，对于子节点中的文件进行上传，对于子节点中的文件夹进行递归。   
对于下载，则是判断一个zkPath是否有Children，如果没有Children，则说明是最后一级，则直接下载，否则在操作系统本地创建一个文件夹，将zkPath的所有孩子节点依次判断是否有子节点，没有就直接下载，有就对该节点递归遍历。

# 具体代码 #

package com.linewell.solrdemo
    
    import java.io.{FileOutputStream, File, FileInputStream}
    
    import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
    import org.apache.curator.retry.ExponentialBackoffRetry
    import org.slf4j.LoggerFactory
    
    /** * 主要提供上传下载功能 * Created by ctao on 2015/11/2. */
    object ZkUtil{ 
      lazy val logger = LoggerFactory.getLogger(this.getClass)
      var zkClient: CuratorFramework = _
    
      /** * 初始化CuratorFramework */
      def init(): Unit = {
        val retryPolicy = new ExponentialBackoffRetry(com.linewell.solrdemo.BASE_SlEEP_TIMESMS,
          com.linewell.solrdemo.MAX_RETRIES)
        val builder = CuratorFrameworkFactory.builder().connectString(com.linewell.solrdemo.ZK_HOST).
          namespace(com.linewell.solrdemo.NAME_SPACE).retryPolicy(retryPolicy).
          connectionTimeoutMs(com.linewell.solrdemo.ZK_TIMEOUT)
        zkClient = builder.build()
        zkClient.start()
      }
    
    
      /** * 销毁CuratorFramework */
      def destroy(): Unit = {
        zkClient.close()
      }
    
      /** * * @param path 绝对路径 * @return 字节码数组 */
      private def file2Byte(path: String) = {
        val file = new File(path)
        if (file.isDirectory) {
          null
        } else if (file.isFile) {
          val is = new FileInputStream(file)
          Iterator continually is.read takeWhile (-1 !=) map (_.toByte) toArray
        } else {
          throw new Exception()
        }
      }
    
    
      /** * 上传方法 * @param confName 配置文件名 * @param osPath 系统路径 */
      def upload(confName: String, osPath: String): Unit = {
        if (new File(osPath).isFile) {
          val newConfName = s"/$confName"
          zkClient.create().forPath(newConfName, file2Byte(osPath))
          logger.info(s"upload $osPath,create node:$newConfName")
          return
        } else {
          val newConfName = s"/$confName"
          zkClient.create().forPath(newConfName)
          logger.info(s"create node:$newConfName")
          val files = new File(osPath).listFiles()
          files.foreach { file =>
            if (file.isDirectory) {
              val childName = s"$confName/${file.getName}"
              upload(childName, file.getAbsolutePath)
            } else {
              val childName = s"/$confName/${file.getName}"
              zkClient.create().forPath(childName, file2Byte(file.getAbsolutePath))
              logger.info(s"upload ${file.getAbsolutePath} create node:$childName")
            }
    
          }
        }
      }
    
      /** * 下载方法 * @param confName 配置文件名 * @param osPath 系统路径 */
      def download(confName: String, osPath: String): Unit = {
        if (zkClient.getChildren.forPath(confName).isEmpty) {
          val file = new File(osPath)
          val fos = new FileOutputStream(file)
          val bytes = zkClient.getData.forPath(confName)
          fos.write(bytes)
          fos.flush()
          fos.close()
          logger.info(s"download $confName,create file:$osPath")
        } else {
          new File(osPath).mkdir()
          val list = zkClient.getChildren.forPath(confName).toArray
          if (list.nonEmpty) {
            list.foreach { zkFile =>
              if (zkClient.getChildren.forPath(s"$confName/$zkFile").isEmpty) {
                val file = new File(s"$osPath/$zkFile")
                val fos = new FileOutputStream(file)
                val bytes = zkClient.getData.forPath(s"$confName/$zkFile")
                fos.write(bytes)
                fos.flush()
                fos.close()
                logger.info(s"download $confName/$zkFile,create file:$osPath/$zkFile")
              } else {
                download(s"$confName/$zkFile", s"$osPath/$zkFile")
              }
            }
          }
        }
    
      }
    
    
    }
    
    
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97
       98
       99
       100
       101
       102
       103
       104
       105
       106
       107
       108
       109
       110
       111
       112
       113
       114
       115
       116
       117
       118
       119
       120
       121
       122
       123
       124
       125
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97
       98
       99
       100
       101
       102
       103
       104
       105
       106
       107
       108
       109
       110
       111
       112
       113
       114
       115
       116
       117
       118
       119
       120
       121
       122
       123
       124
       125

包对象放置常量配置

package com.linewell
    
    /** * 配置内容 * Created by ctao on 2015/10/30. */
    package object solrdemo { 
      //zookeeper主机端口
      val ZK_HOST = "ucap:2181"
      //超时时间
      val ZK_TIMEOUT = 10000
      //zk命名空间
      val NAME_SPACE = "test"
      //初始化重试时间
      val BASE_SlEEP_TIMESMS = 3
      //重试次数
      val MAX_RETRIES = 10000
    }
    
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19

# 运行结果 #

zk截图   
![上传效果][20151103095421979]

下载本地截图   
![下载效果][20151103095450947]

[Link 1]: http://lib.csdn.net/base/operatingsystem
[20151103095421979]: /images/20220708/db3e1a0b86a14f6eb6ffb7c2b9b0b86b.png
[20151103095450947]: /images/20220708/98e14255b3914945bebb69717e9cbc26.png