Lanson

V1

2022/05/29阅读:17主题:丘比特忙

Ceph Swift 实践运用(四)

Ceph Swift 实践运用

一、Ceph封装与自动化装配

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

2、pom文件依赖:

    <dependencies>
        <!-- Spring Boot 自定义启动器的依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-actuator-autoconfigure</artifactId>
        </dependency>

        <!-- ceph 依赖 -->
        <dependency>
            <groupId>com.ceph</groupId>
            <artifactId>rados</artifactId>
            <version>0.6.0</version>
        </dependency>


        <!-- ceph fs 操作依赖 -->
        <dependency>
            <groupId>com.ceph</groupId>
            <artifactId>libcephfs</artifactId>
            <version>0.80.5</version>
        </dependency>


        <!-- ceph swift 依赖 -->
        <dependency>
            <groupId>org.javaswift</groupId>
            <artifactId>joss</artifactId>
            <version>0.10.2</version>
        </dependency>

    </dependencies>

直接采用目前的最新版, 加入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<Container> 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
  password654321
  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;

    }

五、功能验证

1、访问上传页面

地址: http://127.0.0.1:10692/user/file

2、上传成功后, 会返回文件ID:

3、下载文件:

输入文件 ID 进行下载:

分类:

后端

标签:

云计算

作者介绍

Lanson
V1

CSDN大数据领域博客专家