Ceph分布式存储实践应用之Ceph Swift生产实践运用

1. Ceph封装与自动化装配

创建ceph-starter自动化工程：

创建一个spring boot工程，作为一个公用组件。

pom文件依赖：


 
 
     org.springframework.boot
     spring-boot-actuator-autoconfigure
 
 
 
     com.ceph
     rados
     0.6.0
 

 
 
     com.ceph
     libcephfs
     0.80.5
 

 
 
     org.javaswift
     joss
     0.10.2


直接采用目前的最新版， 加入Ceph相关的三个依赖。



代码实现
封装Ceph操作接口， CephSwiftOperator类：
public class CephSwiftOperator {

    private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

    
    private String username;   
    
    private String password;

    
    private String authUrl;

    
    private String defaultContainerName;

    
    private Account account;

    
    private Container container;   

    public CephSwiftOperator(String username, String password, String authUrl, String defaultContainerName) {
 // 初始化配置信息
 this.username = username;
 this.password = password;
 this.authUrl = authUrl;
 this.defaultContainerName = defaultContainerName;
 init();   
    }

    
    public void init() {
 try {
     // Ceph用户认证配置
     AccountConfig config = new AccountConfig();
     config.setUsername(username);
     config.setPassword(password);
     config.setAuthUrl(authUrl);
     config.setAuthenticationMethod(AuthenticationMethod.BASIC);
     account = new AccountFactory(config).createAccount();
     // 获取容器
     Container newContainer = account.getContainer(defaultContainerName);
     if (!newContainer.exists()) {
  container = newContainer.create();
  log.info("account container create ==> " + defaultContainerName);
     } else {
  container = newContainer;
  log.info("account container exists!  ==> " + defaultContainerName);
     }
 }catch(Exception e) {
     // 做异常捕获, 避免服务不能正常启动
     log.error("Ceph连接初始化异常： " + e.getMessage());
 }
    }   

    
    public void createObject(String remoteName, String filepath) {
 StoredObject object = container.getObject(remoteName);
 object.uploadObject(new File(filepath));
    }

    
    public void createObject(String remoteName, byte[] inputStream) {
 StoredObject object = container.getObject(remoteName);
 object.uploadObject(inputStream);
    }

    
    public void  retrieveObject(String objectName,String outpath){
 StoredObject object = container.getObject(objectName);
 object.downloadObject(new File(outpath));
    }
   
    public InputStream retrieveObject(String objectName){
 StoredObject object = container.getObject(objectName);
 return object.downloadObjectAsInputStream();
    }
   
    
    public boolean deleteObject(String objectName){
 try {
     StoredObject object = container.getObject(objectName);
     object.delete();
     return !object.exists();
 }catch(Exception e) {
     log.error("Ceph删除文件失败: " + e.getMessage());
 }
 return false;
    }

    
    public List listContainer() {
 List list = new ArrayList();
 Collection containers = account.list();
 for (Container currentContainer : containers) {
     list.add(currentContainer.getName());
     System.out.println(currentContainer.getName());

 }
 return list;
    }   
}   


此封装接口将纳入Spring 容器管理， 即为单例， 构造函数会初始化Ceph认证连接等信息


将Account 与Container 设为成员变量， 便于复用， 减少开销。


初始化会默认一个容器名称， 一般每个服务设置一个容器名称， 如果业务功能比较庞杂， 可以每个业务模块设置一个容器。


Swift Api已经做了较完善的封装， 我们内部使用比较简单， 主要封装上传和下载接口， 为便于调用处理， 做了进一步封装。

AutoCephSwiftConfiguration自动化配置类：
@Configuration
@EnableAutoConfiguration
@ConditionalOnProperty(name = "ceph.authUrl")
public class AutoCephSwiftConfiguration {   
    @Value("${ceph.username}")
    private String username;
    @Value("${ceph.password}")
    private String password;
    @Value("${ceph.authUrl}")
    private String authUrl;
    @Value("${ceph.defaultContainerName}")
    private String defaultContainerName;  
    @Bean
    public CephSwiftOperator cephSwiftOperator() {
 return new CephSwiftOperator(username, password, authUrl, defaultContainerName);
    }

}


ConditionalOnProperty根据ceph.authUrl属性来决定是否加载配置，如果配置文件中没有设置Ceph相关属性， 即使maven中引用， 启动也不会报错。 该自动化配置， 负责初始化一个Ceph Swift 接口操作实例。


自动化配置：
要让自定义Ceph Starter真正生效， 必须遵循Spring boot 的SPI扩展机制， 在resources环境中,  meta-INF目录下， 创建spring.factories文件:
# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
com.ceph.starter.AutoCephSwiftConfiguration

指定我们上面所写的自动化配置类。

2. 创建用户管理工程

工程配置

application.yml
server:
  port: 10692
spring:
  application:
    name: user-manager
  # 模板配置
  thymeleaf:
    prefix: classpath:/templates/
    suffix: .html
    mode: HTML
    encoding: utf-8
    servlet:
      content-type: text/html

  # 文件上传大小限制
  servlet:
    multipart:
      max-file-size: 100MB
      max-request-size: 100MB

# ceph swift 认证信息配置
ceph:
  username: cephtester:subtester
  password: 654321
  authUrl: http://10.10.20.11:7480/auth/1.0
  defaultContainerName: user_datainfo



POM依赖配置：
   


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


    org.springframework.boot
    spring-boot-starter-thymeleaf
   


    com.itcast.ceph
    ceph-starter
    1.0-SNAPSHOT
   
   

加入Ceph 自动化封装组件配置依赖。

3. Ceph文件上传实现

实现文件上传接口：

public String uploadUserFile(MultipartFile file) throws Exception {

    // 获取唯一文件ID标识
    String remoteFileId = globalIDGenerator.nextStrId();

    // 上传文件至CEPH
    cephSwiftOperator.createObject(remoteFileId, file.getBytes());

    return remoteFileId;
}



Controller层实现：
在UserManagerController下面， 增加上传接口：

@PostMapping("/upload")
@ResponseBody
public String upload(@RequestParam("file") MultipartFile file) {
    String  result = null;
    try {
 // 通过Ceph Swift上传文件
 String userFileId = userManagerService.uploadUserFile(file);
 result = "上传的文件ID: " + userFileId;
    }catch(Exception e) {
 log.error(e.getMessage(), e);
 result = "出现异常：" + e.getMessage();
    }
    return result;
}   

通过Spring boot实现文件上传， 注意以表单form形式提交， 类型为multipart/form-data， 参数名为file， 非直接数据流方式上传， 不然服务不能正常识别接收文件。

4. Ceph文件下载实现
新增一个接口， 根据上传的文件ID标识下载文件。

Service层：
实现下载用户文件接口：

public InputStream downloadUserFile(String fileId) throws Exception {
    return cephSwiftOperator.retrieveObject(fileId);
}



Controller层：

@RequestMapping(value = "/download")
public String downloadFile(@NotBlank(message = "文件ID不能为空!") String filename, HttpServletResponse response){   
    String result = null;   
    // 文件流缓存
    BufferedInputStream bis = null;
    // 文件输出流
    OutputStream os = null;
    try {
 // 1. 从Ceph服务器上获取文件流
 InputStream inputStream = userManagerService.downloadUserFile(filename);
 // 2.设置强制下载, 不直接打开
 response.setContentType("application/x-msdownload");
 // 3. 设置下载的文件名称
 response.addHeader("Content-disposition", "attachment; fileName=" + filename);
 // 4. 输出文件流
 byte[] buffer = new byte[1024];
 bis = new BufferedInputStream(inputStream);
 os = response.getOutputStream();
 int i = bis.read(buffer);
 while(i != -1) {
     os.write(buffer, 0, i);
     i = bis.read(buffer);
 }
 os.flush();
 return null;
    }catch(Exception e) {
 log.error(e.getMessage(), e);
 result = "出现异常：" + e.getMessage();
    }finally {
 // 最后, 要记住关闭文件流
 if(bis != null ) {
     try {
  bis.close();
     } catch (IOException e) {
  log.error(e.getMessage(), e);
     }
 }
    }   
    return result;

}


5. 功能验证

访问上传页面
地址： http://127.0.0.1:10692/user/file



上传成功后， 会返回文件ID：



下载文件：
输入文件ID进行下载：

Ceph分布式存储实践应用之Ceph Swift生产实践运用

Java相关栏目本月热门文章