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 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; }
