Ceph分布式存储实践应用之集群测试验证（Rados运用）

monIp为ceph client节点连接地址与端口， 支持多个；

userKey为密钥， 对应ceph.client.admin.keyring中的key值。

启动接收两个参数， 一个是标识上传或下载， 另一个是标识文件名称。

CephOperator操作类的实现：

public class CephOperator {

    private CephMount mount;
    private String username;
    private String monIp;
    private String userKey;

    public CephOperator(String username, String monIp, String userKey, String mountPath) {
 this.username = username;
 this.monIp = monIp;
 this.userKey = userKey;
 this.mount = new CephMount(username);
 this.mount.conf_set("mon_host", monIp);
 mount.conf_set("key", userKey);
 mount.mount(mountPath);
    }

    //查看目录列表
    public void listDir(String path) throws IOException {
 String[] dirs = mount.listdir(path);
 System.out.println("contents of the dir: " + Arrays.asList(dirs));
    }

    //新建目录
    public void mkDir(String path) throws IOException {
 mount.mkdirs(path, 0755);//0表示十进制
    }

    //删除目录
    public void delDir(String path) throws IOException {
 mount.rmdir(path);
    }

    //重命名目录or文件
    public void renameDir(String oldName, String newName) throws IOException {
 mount.rename(oldName, newName);
    }

    //删除文件
    public void delFile(String path) throws IOException {
 mount.unlink(path);
    }

    
    public Boolean uploadFileByPath(String filePath, String fileName) {

 // 如果mount操作单元为空则直接返回
 if (this.mount == null) {
     return null;
 }
 // 文件描述信息定义
 char pathChar = File.separatorChar;
 String fileFullName = "";
 Long fileLength = 0l;
 Long uploadedLength = 0l;
 File file = null;

 // 定义文件流
 FileInputStream fis = null;

 // 获取本地文件信息
 fileFullName = filePath + pathChar + fileName;
 file = new File(fileFullName);
 if (!file.exists()) {
     return false;
 }
 fileLength = file.length();

 // 获取本地文件流
 try {
     fis = new FileInputStream(file);
 } catch (FileNotFoundException e) {
     e.printStackTrace();
 }

 // 判断文件是否已经存在
 String[] dirList = null;
 Boolean fileExist = false;
 try {
     dirList = this.mount.listdir("/");
     for (String fileInfo : dirList) {
  if (fileInfo.equals(fileName)) {
      fileExist = true;
  }
     }
 } catch (FileNotFoundException e) {
     e.printStackTrace();
 }

 if (!fileExist) {
     try {
  // 创建文件并设置为写入模式
  this.mount.open(fileName, CephMount.O_CREAT, 0);
  int fd = this.mount.open(fileName, CephMount.O_RDWR, 0);

  // 开始文件传输
  int length = 0;
  byte[] bytes = new byte[1024];
  while ((length = fis.read(bytes, 0, bytes.length)) != -1) {
      // 上传写入数据
      this.mount.write(fd, bytes, length, uploadedLength); 

      // 更新上传进度
      uploadedLength += length;

      // 输出上传百分比
      float rate = (float) uploadedLength * 100 / (float) fileLength;
      String ratevalue = (int) rate + "%";
      System.out.println(ratevalue);

      // 上传完成后退出
      if (uploadedLength == fileLength) {
   break;
      }
  }
  System.out.println("文件传输成功！");

  // 设置文件权限
  this.mount.fchmod(fd, 0666);

  // 关闭操作
  this.mount.close(fd);
  if (fis != null) {
      fis.close();
  }
  return true;
     } catch (Exception e) {
  e.printStackTrace();
     }
 } else if (fileExist) {
     try {
  // 获取文件大小
  CephStat stat = new CephStat();
  this.mount.stat(fileName, stat);
  long lastLen= stat.size;
  int fd = this.mount.open(fileName, CephMount.O_RDWR, 0);

  // 开始文件传输
  int length = 0;
  byte[] bytes = new byte[1024];
  long uploadActLen= 0;
  while ((length = fis.read(bytes, 0, bytes.length)) != -1) {
      // 更新写入
      this.mount.write(fd, bytes, length, lastLen);

      // 更新文件大小
      uploadActLen += length;
      // 更新文件上传百分比
      float rate = (float) uploadActLen * 100 / (float) fileLength;
      String ratevalue = (int) rate + "%";
      System.out.println(ratevalue);

      // complete flag
      if (uploadActLen == fileLength) {
   break;
      }
  }
  // 多次上传会进行追加
  System.out.println("追加文件传输成功！");

  // 设置文件权限
  this.mount.fchmod(fd, 0666);

  // 关闭操作
  this.mount.close(fd);
  if (fis != null) {
      fis.close();
  }
  return true;
     } catch (Exception e) {
  e.printStackTrace();
     }
 } else {
     try {
  if (fis != null) {
      fis.close();
  }
     } catch (Exception e) {
  e.printStackTrace();
     }
     return false;
 }

 return false;
    }

    // 设置当前的工作上传目录
    public void setWorkDir(String path) throws IOException {
 mount.chdir(path);
    }
 
    //外部获取mount
    public CephMount getMount() {
 return this.mount;
    }

    // 取消文件挂载
    public void umount() {
 mount.unmount();
    }
   
    
    public Boolean downloadFileByPath(String filePath, String fileName) {

 // 如果mount操作单元为空则直接返回
 if (this.mount == null) {
     return null;
 }

 // 文件描述信息定义
 char pathChar = File.separatorChar;
 String fileFullName = "";
 Long fileLength = 0l;
 Long downloadedLength = 0l;
 File file = null;

 // 定义文件流
 FileOutputStream fos = null;
 RandomAccessFile raf = null;

 // 创建新的文件
 fileFullName = filePath + pathChar + fileName;
 file = new File(fileFullName);

 // 获取 cephfs 的文件大小
 try {
     CephStat stat = new CephStat();
     this.mount.stat(fileName, stat);
     fileLength = stat.size;
 } catch (Exception e) {
     e.printStackTrace();
 }

 if (fileLength != 0) {
     if (!file.exists()) {
  // 下载文件
  int length = 10240;
  byte[] bytes = new byte[length];
  try {
      int fd = this.mount.open(fileName, CephMount.O_RDONLY, 0);
      fos = new FileOutputStream(file);
      float rate = 0;
      String ratevalue = "";
      while ((fileLength - downloadedLength) >= length && (this.mount.read(fd, bytes, (long) length, downloadedLength)) != -1) {
   fos.write(bytes, 0, length);
   fos.flush();
   downloadedLength += (long) length;

   // 输出进度百分比
   rate = (float) downloadedLength * 100 / (float) fileLength;
   ratevalue = (int) rate + "%";
   System.out.println(ratevalue);

   if (downloadedLength == fileLength) {
break;
   }
      }
      if (downloadedLength != fileLength) {
   this.mount.read(fd, bytes, fileLength - downloadedLength, downloadedLength);
   fos.write(bytes, 0, (int) (fileLength - downloadedLength));
   fos.flush();
   downloadedLength = fileLength;

   // 输出进度百分比
   rate = (float) downloadedLength * 100 / (float) fileLength;
   ratevalue = (int) rate + "%";
   System.out.println(ratevalue);
      }

      System.out.println("Download Success!");
      fos.close();
      this.mount.close(fd);
      return true;
  } catch (Exception e) {
      e.printStackTrace();
  }
     } else if (file.exists()) {
  // 下载文件
  int length = 10240;
  byte[] bytes = new byte[length];
  Long filePoint = file.length();
  try {
      int fd = this.mount.open(fileName, CephMount.O_RDONLY, 0);
      raf = new RandomAccessFile(file, "rw");
      raf.seek(filePoint);
      downloadedLength = filePoint;
      float rate = 0;
      String ratevalue = "";
      while ((fileLength - downloadedLength) >= length && (this.mount.read(fd, bytes, (long) length, downloadedLength)) != -1) {
   raf.write(bytes, 0, length);
   downloadedLength += (long) length;

   // 输出进度百分比
   rate = (float) downloadedLength * 100 / (float) fileLength;
   ratevalue = (int) rate + "%";
   System.out.println(ratevalue);

   if (downloadedLength == fileLength) {
break;
   }
      }
      if (downloadedLength != fileLength) {
   this.mount.read(fd, bytes, fileLength - downloadedLength, downloadedLength);
   raf.write(bytes, 0, (int) (fileLength - downloadedLength));
   downloadedLength = fileLength;

   // 输出进度百分比
   rate = (float) downloadedLength * 100 / (float) fileLength;
   ratevalue = (int) rate + "%";
   System.out.println(ratevalue);
      }
					// 如果下载中断, 会从上一次下载结束位置进行上传
      System.out.println("Cut Point Download Success!");
      raf.close();
      this.mount.close(fd);
      return true;
  } catch (Exception e) {
      e.printStackTrace();
  }
     } else {
  return false;
     }
 }else {
     System.out.println(" the file is empty!");
 } 
 return true;   
    }   
}

    
    
 org.springframework.boot
 spring-boot-starter-web
    

    
    
 com.ceph
 rados
 0.6.0
    
	
    
 com.ceph
 libcephfs
 0.80.5
    




    
 
     org.springframework.boot
     spring-boot-maven-plugin