Ceph Swift 实践运用（四）

科技 07-27 来源： Lansonli

Ceph Swift 实践运用

一、Ceph 封装与自动化装配

1、创建 ceph-starter 自动化工程：

2、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 相关的三个依赖。

3、代码实现

封装 Ceph 操作接口， CephSwiftOperator 类：

package cn.it.ceph.starter;


import org.javaswift.joss.client.factory.AccountConfig;
import org.javaswift.joss.client.factory.AccountFactory;
import org.javaswift.joss.client.factory.AuthenticationMethod;
import org.javaswift.joss.model.Account;
import org.javaswift.joss.model.Container;
import org.javaswift.joss.model.StoredObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.File;
import java.io.InputStream;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;


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;


    /**
     * Ceph账户对象
     */
    private Account account;


    /**
     * Ceph容器对象
     */
    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());
        }
    }




    /**
     * 上传对象
     * @param remoteName
     * @param filepath
     */
    public void createObject(String remoteName, String filepath) {
        StoredObject object = container.getObject(remoteName);
        object.uploadObject(new File(filepath));
    }


    /**
     * 上传文件对象（字节数组形式）
     * @param remoteName
     * @param inputStream
     */
    public void createObject(String remoteName, byte[] inputStream) {
        StoredObject object = container.getObject(remoteName);
        object.uploadObject(inputStream);
    }


    /**
     * 获取指定对象
     * @param containerName
     * @param objectName
     * @param outpath
     */
    public void  retrieveObject(String objectName,String outpath){
        StoredObject object = container.getObject(objectName);
        object.downloadObject(new File(outpath));
    }


    /**
     * 下载文件， 转为文件流形式
     * @param objectName
     * @return
     */
    public InputStream retrieveObject(String objectName){
        StoredObject object = container.getObject(objectName);
        return object.downloadObjectAsInputStream();
    }




    /**
     * 删除指定文件对象
     * @param containerName
     * @param objectName
     * @return
     */
    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;
    }


    /**
     * 获取所有容器
     * @return
     */
    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;
    }


}

复制代码

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

4、自动化配置：

要让自定义 Ceph Starter 真正生效，必须遵循 Spring boot 的 SPI 扩展机制，在 resources 环境中, META-INF 目录下，创建 spring.factories 文件:

# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
cn.it.ceph.starter.AutoCephSwiftConfiguration

复制代码

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

二、创建用户管理工程

1、工程结构：

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://192.168.88.161:7480/auth/1.0
  defaultContainerName: user_datainfo

复制代码

三、Ceph 文件上传实现

1、实现文件上传接口：

   /*** 上传用户文件 * @return */
    public String uploadUserFile(MultipartFile file) throws Exception {
        // 获取唯一文件ID标识
        String remoteFileId = globalIDGenerator.nextStrId();
        // 上传文件至CEPH
        cephSwiftOperator.createObject(remoteFileId, file.getBytes());
        return remoteFileId;
    }

复制代码

2、Controller 层实现：

在 UserManagerController 下面，增加上传接口：

    /**
     * 用户文件上传
     * @param file
     * @return
     */
    @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) {
            e.printStackTrace();
            result = "出现异常：" + e.getMessage();
        }
        return result;
    }

复制代码

四、Ceph 文件下载实现

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

1、Service 层：

实现下载用户文件接口：

    /*** 下载用户文件 * @param fileId * @return * @throws Exception */ 
    public InputStream downloadUserFile(String fileId) throws Exception { 
        return cephSwiftOperator.retrieveObject(fileId); 
    }

复制代码

2、Controller 层：

    /**
     * 根据文件ID下载用户文件信息
     * @param filename
     * @return
     */
    @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) {
            e.printStackTrace();
            result = "出现异常：" + e.getMessage();
        }finally {
            // 最后, 要记住关闭文件流
            if(bis != null ) {
                try {
                    bis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


        return result;


    }

复制代码