聊聊PowerJob的AliOssService

序

本文主要研究一下PowerJob的AliOssService
DFsService

tech/powerjob/server/extension/dfs/DFsService.java

1. public interface DFsService {
3.     /**
4.      * 存储文件
5.      * @param storeRequest 存储请求
6.      * @throws IOException 异常
7.      */
8.     void store(StoreRequest storeRequest) throws IOException;
10.     /**
11.      * 下载文件
12.      * @param downloadRequest 文件下载请求
13.      * @throws IOException 异常
14.      */
15.     void download(DownloadRequest downloadRequest) throws IOException;
17.     /**
18.      * 获取文件元信息
19.      * @param fileLocation 文件位置
20.      * @return 存在则返回文件元信息
21.      * @throws IOException 异常
22.      */
23.     Optional<FileMeta> fetchFileMeta(FileLocation fileLocation) throws IOException;
25.     /**
26.      * 清理 powerjob 认为“过期”的文件
27.      * 部分存储系统自带生命周期管理（如阿里云OSS，则不需要单独实现该方法）
28.      * @param bucket bucket
29.      * @param days 天数，需要清理超过 X 天的文件
30.      */
31.     default void cleanExpiredFiles(String bucket, int days) {
32.     }
33. }

复制代码

  DFsService接口定义了store、download、fetchFileMeta、cleanExpiredFiles方法
  AbstractDFsService

tech/powerjob/server/persistence/storage/AbstractDFsService.java

1. @Slf4j
2. public abstract class AbstractDFsService implements DFsService, ApplicationContextAware, DisposableBean {
4.     protected ApplicationContext applicationContext;
6.     public AbstractDFsService() {
7.         log.info("[DFsService] invoke [{}]'s constructor", this.getClass().getName());
8.     }
10.     abstract protected void init(ApplicationContext applicationContext);
12.     protected static final String PROPERTY_KEY = "oms.storage.dfs";
14.     protected static String fetchProperty(Environment environment, String dfsType, String key) {
15.         String pKey = String.format("%s.%s.%s", PROPERTY_KEY, dfsType, key);
16.         return environment.getProperty(pKey);
17.     }
19.     @Override
20.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
21.         this.applicationContext = applicationContext;
22.         log.info("[DFsService] invoke [{}]'s setApplicationContext", this.getClass().getName());
23.         init(applicationContext);
24.     }
25. }

复制代码

  AbstractDFsService声明实现DFsService、ApplicationContextAware、DisposableBean接口，它在setApplicationContext方法执行了init
  AliOssService

tech/powerjob/server/persistence/storage/impl/AliOssService.java

1. @Slf4j
2. @Priority(value = Integer.MAX_VALUE - 1)
3. @Conditional(AliOssService.AliOssCondition.class)
4. public class AliOssService extends AbstractDFsService {
6.     private static final String TYPE_ALI_OSS = "alioss";
8.     private static final String KEY_ENDPOINT = "endpoint";
9.     private static final String KEY_BUCKET = "bucket";
10.     private static final String KEY_CREDENTIAL_TYPE = "credential_type";
11.     private static final String KEY_AK = "ak";
12.     private static final String KEY_SK = "sk";
13.     private static final String KEY_TOKEN = "token";
15.     private OSS oss;
16.     private String bucket;
18.     private static final int DOWNLOAD_PART_SIZE = 10240;
20.     private static final String NO_SUCH_KEY = "NoSuchKey";
22.     //......
24.     void initOssClient(String endpoint, String bucket, String mode, String ak, String sk, String token) throws Exception {
26.         log.info("[AliOssService] init OSS by config: endpoint={},bucket={},credentialType={},ak={},sk={},token={}", endpoint, bucket, mode, ak, sk, token);
28.         if (StringUtils.isEmpty(bucket)) {
29.             throw new IllegalArgumentException("'oms.storage.dfs.alioss.bucket' can't be empty, please creat a bucket in aliyun oss console then config it to powerjob");
30.         }
32.         this.bucket = bucket;
34.         CredentialsProvider credentialsProvider;
35.         CredentialType credentialType = CredentialType.parse(mode);
36.         switch (credentialType) {
37.             case PWD:
38.                 credentialsProvider = new DefaultCredentialProvider(ak, sk, token);
39.                 break;
40.             case SYSTEM_PROPERTY:
41.                 credentialsProvider = CredentialsProviderFactory.newSystemPropertiesCredentialsProvider();
42.                 break;
43.             default:
44.                 credentialsProvider = CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider();
45.         }
47.         this.oss = new OSSClientBuilder().build(endpoint, credentialsProvider);
48.         log.info("[AliOssService] initialize successfully, THIS_WILL_BE_THE_STORAGE_LAYER.");
49.     }
51.     //......   
52. }   

复制代码

  AliOssService继承了AbstractDFsService
  store

1.     @Override
2.     public void store(StoreRequest storeRequest) throws IOException {
4.         ObjectMetadata objectMetadata = new ObjectMetadata();
6.         PutObjectRequest putObjectRequest = new PutObjectRequest(bucket, parseFileName(storeRequest.getFileLocation()), storeRequest.getLocalFile(), objectMetadata);
7.         oss.putObject(putObjectRequest);
8.     }

复制代码

  store方法创建PutObjectRequest，使用oss.putObject举行上传
  download

1.     @Override
2.     public void download(DownloadRequest downloadRequest) throws IOException {
4.         FileLocation dfl = downloadRequest.getFileLocation();
5.         DownloadFileRequest downloadFileRequest = new DownloadFileRequest(bucket, parseFileName(dfl), downloadRequest.getTarget().getAbsolutePath(), DOWNLOAD_PART_SIZE);
6.         try {
7.             FileUtils.forceMkdirParent(downloadRequest.getTarget());
8.             oss.downloadFile(downloadFileRequest);
9.         } catch (Throwable t) {
10.             ExceptionUtils.rethrow(t);
11.         }
12.     }

复制代码

  download方法则根据DownloadRequest指定的FileLocation创建DownloadFileRequest，然后通过oss.downloadFile(downloadFileRequest)举行下载
  fetchFileMeta

1.     @Override
2.     public Optional<FileMeta> fetchFileMeta(FileLocation fileLocation) throws IOException {
3.         try {
4.             ObjectMetadata objectMetadata = oss.getObjectMetadata(bucket, parseFileName(fileLocation));
5.             return Optional.ofNullable(objectMetadata).map(ossM -> {
7.                 Map<String, Object> metaInfo = Maps.newHashMap();
8.                 metaInfo.putAll(ossM.getRawMetadata());
9.                 if (ossM.getUserMetadata() != null) {
10.                     metaInfo.putAll(ossM.getUserMetadata());
11.                 }
13.                 return new FileMeta()
14.                         .setLastModifiedTime(ossM.getLastModified())
15.                         .setLength(ossM.getContentLength())
16.                         .setMetaInfo(metaInfo);
17.             });
18.         } catch (OSSException oe) {
19.             String errorCode = oe.getErrorCode();
20.             if (NO_SUCH_KEY.equalsIgnoreCase(errorCode)) {
21.                 return Optional.empty();
22.             }
23.             ExceptionUtils.rethrow(oe);
24.         }
25.         return Optional.empty();
26.     }

复制代码

  fetchFileMeta方法通过oss.getObjectMetadata获取ObjectMetadata
  cleanExpiredFiles

1.     @Override
2.     public void cleanExpiredFiles(String bucket, int days) {
3.         /*
4.         阿里云 OSS 自带生命周期管理，请参考文档进行配置，代码层面不进行实现（浪费服务器资源）https://help.aliyun.com/zh/oss/user-guide/overview-54
5.         阿里云 OSS 自带生命周期管理，请参考文档进行配置，代码层面不进行实现（浪费服务器资源）https://help.aliyun.com/zh/oss/user-guide/overview-54
6.         阿里云 OSS 自带生命周期管理，请参考文档进行配置，代码层面不进行实现（浪费服务器资源）https://help.aliyun.com/zh/oss/user-guide/overview-54
7.          */
8.     }

复制代码

  cleanExpiredFiles则是空操纵
  init

1.     protected void init(ApplicationContext applicationContext) {
2.         Environment environment = applicationContext.getEnvironment();
4.         String endpoint = fetchProperty(environment, TYPE_ALI_OSS, KEY_ENDPOINT);
5.         String bkt = fetchProperty(environment, TYPE_ALI_OSS, KEY_BUCKET);
6.         String ct = fetchProperty(environment, TYPE_ALI_OSS, KEY_CREDENTIAL_TYPE);
7.         String ak = fetchProperty(environment, TYPE_ALI_OSS, KEY_AK);
8.         String sk = fetchProperty(environment, TYPE_ALI_OSS, KEY_SK);
9.         String token = fetchProperty(environment, TYPE_ALI_OSS, KEY_TOKEN);
11.         try {
12.             initOssClient(endpoint, bkt, ct, ak, sk, token);
13.         } catch (Exception e) {
14.             ExceptionUtils.rethrow(e);
15.         }
16.     }

复制代码

  init则是通过environment获取相关属性，然后执行initOssClient
  小结

DFsService接口定义了store、download、fetchFileMeta、cleanExpiredFiles方法；AbstractDFsService声明实现DFsService、ApplicationContextAware、DisposableBean接口，它在setApplicationContext方法执行了init；AliOssService继承了AbstractDFsService，通过ossClient实现了store、download、fetchFileMeta方法。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。