java實現華為雲對象存儲OBS的文件上傳

最近碰到公司的磁盤需要擴容，新購進的存儲為華為雲OBS，需要將公司服務器文件遷移至華為雲OBS，且不影響業務，本地程序不需要做太大改動。

實現思路：  用監聽模式對指定文件目錄執行監聽，檢測新增文件上傳至OBS，並返回下載地址。

實現過程如下：

1.pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.zwj</groupId>
   <artifactId>obs-upload</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <packaging>jar</packaging>
   <description>OBS文件上傳</description>

   <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>1.5.9.RELEASE</version>
      <relativePath/> <!-- lookup parent from repository -->
   </parent>
   <properties>
      <java.version>1.8</java.version>
      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
      <tomcat.version>7.0.63</tomcat.version>
   </properties>

   <dependencies>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
      <dependency>
         <groupId>org.apache.tomcat</groupId>
         <artifactId>tomcat-jdbc</artifactId>
      </dependency>

       <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.4</version>
        </dependency>
        <dependency>
            <groupId>commons-lang</groupId>
            <artifactId>commons-lang</artifactId>
            <version>2.4</version>
        </dependency>
       <dependency>
            <groupId>commons-net</groupId>
            <artifactId>commons-net</artifactId>
            <version>3.3</version>
        </dependency>


      <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>
         <version>4.12</version>
         <scope>test</scope>
      </dependency>

      <dependency>
         <groupId>org.projectlombok</groupId>
         <artifactId>lombok</artifactId>
         <version>1.16.8</version>
      </dependency>

   <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-api</artifactId>
      <version>1.6.1</version>
   </dependency>

   <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>log4j-over-slf4j</artifactId>
      <version>1.6.1</version>
   </dependency>

   <dependency>
      <groupId>com.jcraft</groupId>
      <artifactId>jsch</artifactId>
      <version>0.1.54</version>
   </dependency>

      <dependency>
         <groupId>com.huaweicloud</groupId>
         <artifactId>esdk-obs-java</artifactId>
         <version>3.19.7</version>
         <exclusions>
            <exclusion>
               <groupId>org.apache.logging.log4j</groupId>
               <artifactId>log4j-core</artifactId>
            </exclusion>
            <exclusion>
               <groupId>org.apache.logging.log4j</groupId>
               <artifactId>log4j-api</artifactId>
            </exclusion>
         </exclusions>
      </dependency>

      <dependency>
         <groupId>cn.hutool</groupId>
         <artifactId>hutool-core</artifactId>
         <version>5.2.0</version>
      </dependency>

  </dependencies>

   <build>
      <plugins>
         <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
         </plugin>

         <plugin>
            <!-- 打包插件assembly -->
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <!-- 版本號 -->
            <version>2.2.1</version>
            <!-- 打包描述文件，用於描述把哪些資源進行打包 -->
            <configuration>
               <descriptors>
                  <descriptor>src/main/resources/assembly.xml</descriptor>
               </descriptors>
            </configuration>
            <!-- assembly插件執行配置 -->
            <executions>
               <execution>
                  <!-- 執行ID -->
                  <id>make-assembly</id>
                  <!-- 綁定到package生命周期階段上，執行maven package打包是會啟用本打包配置 -->
                  <phase>package</phase>
                  <goals>
                     <!-- 由於maven整項目打包時，有可能會build編譯多次，single表示執行一次 -->
                     <goal>single</goal>
                  </goals>
               </execution>
            </executions>
         </plugin>
      </plugins>

      <resources>
         <resource>
            <directory>src/main/html</directory>
            <includes>
               <include>**/**</include>
            </includes>
         </resource>
         <resource>
            <directory>src/main/java</directory>
            <excludes>
               <exclude>&gt;**/*.java</exclude>
            </excludes>
         </resource>
         <resource>
            <directory>src/main/resources</directory>
            <includes>
               <include>**/**</include>
            </includes>
            <filtering>false</filtering>
         </resource>
      </resources>
   </build>
</project>
2.OBS工具類：

package com.obsupload.configur;

import com.obs.services.ObsClient;
import com.obs.services.exception.ObsException;
import com.obs.services.model.*;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author: zhengwj
 * @Description:
 * @Date: 2020/4/20 12:47
 * @Version: 1.0
 */
public class OBSHandler {

    private String accessKeyId;// 華為雲的 Access Key Id
    private String accessKeySecret;// 華為雲的 Access Key Secret
    private String endpoint; // 華為雲連接的地址節點

    private String obsBucketName; // 創建的桶的名稱
    private String url; // 訪問OBS文件的url

    private static ObsClient obsClient; // 進行操作的華為雲的客戶端組件


    /**
     * 創建華為雲OBS的本地控制器
     * @param accessKeyId
     * @param accessKeySecret
     * @param endpoint
     */
    public OBSHandler(String accessKeyId, String accessKeySecret, String endpoint) {
        this.accessKeyId = accessKeyId;
        this.accessKeySecret = accessKeySecret;
        this.endpoint = endpoint;
    }

    public OBSHandler(String accessKeyId, String accessKeySecret, String endpoint, String obsBucketName) {
        this.accessKeyId = accessKeyId;
        this.accessKeySecret = accessKeySecret;
        this.endpoint = endpoint;
        this.obsBucketName = obsBucketName;
    }

    /**
     * 設置OBS訪問的CDN路徑
     * @param url 
     */
    public void setUrlForCDN(String url) {
        this.url = url;
    }

    /**
     * 設置OBS操作的同桶名稱
     * @param obsBucketName
     */
    public void setObsBucketName(String obsBucketName) {
        this.obsBucketName = obsBucketName;
    }

    /**
     * 獲取華為雲提供的操作客戶端實體類
     * @return
     */
    public ObsClient getObsClient() {
        if(obsClient == null) {
            obsClient = new ObsClient(accessKeyId, accessKeySecret, endpoint);
        }
        return obsClient;
    }

    /**
     * 下載ObsObject
     * @param bucketName    操作的桶的名稱 例："wangmarket1232311"
     * @param filePath   需要下載的文件路徑。 例："site/a.txt"
     * @return  下載文件的字節數組
     * @throws IOException
     */
    public byte[] getFileByteArray(String bucketName, String filePath) throws IOException {
        ObsObject obsObject = getObsClient().getObject(bucketName, filePath);
        InputStream input = obsObject.getObjectContent();
        byte[] b = new byte[1024];
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        int len;
        while ((len = input.read(b)) != -1){
            bos.write(b, 0, len);
        }
        bos.close();
        input.close();
        return bos.toByteArray();
    }


    /**
     * 獲取指定路徑下的ObsObject數量
     * @param bucketName  操作的桶的名稱 例："wangmarket1232311"
     * @param filePath 需要檢索的文件夾路徑 例："site/"
     * @return  檢索搜文件下的ObsObject的數量
     */
    public Integer getFolderObjectsSize(String bucketName, String filePath) {
        ListObjectsRequest request = new ListObjectsRequest(bucketName);
        if(filePath != null && (!filePath.trim().equals(""))){
            request.setPrefix(filePath);
        }
        ObjectListing result = getObsClient().listObjects(request);
        return new Integer(result.getObjects().size());
    }

    /**
     * 獲取指定路徑下的ObsObject
     * @param bucketName 操作的桶的名稱 例："wangmarket1232311"
     * @param filePath 需要檢索的文件夾路徑
     * @return 路徑下的所有的ObsObject，包括子文件夾下的ObsObject
     */
    public List<ObsObject> getFolderObjects(String bucketName, String filePath) {
        List<ObsObject> list = new ArrayList<ObsObject>();
        ListObjectsRequest request = new ListObjectsRequest(bucketName);
        if(filePath != null && (!filePath.trim().equals(""))){
            request.setPrefix(filePath);
        }
        request.setMaxKeys(100);
        ObjectListing result;
        do{
            result = getObsClient().listObjects(request);
            for(ObsObject obsObject : result.getObjects()){
                list.add(obsObject);
            }
            request.setMarker(result.getNextMarker());
        }while(result.isTruncated());
        return list;
    }

    /**
     * 刪除對象
     * @param bucketName 操作的桶的名稱 例："wangmarket1232311"
     * @param fileName  需要刪除的對象全名 例："site/20190817/localFile.sh"
     * @return
     */
    public DeleteObjectResult deleteObject(String bucketName, String fileName) {
        return getObsClient().deleteObject(bucketName, fileName);
    }

    /**
     * 創建文件夾
     * @param bucketName 操作的桶的名稱 例："wangmarket1232311"
     * @param fileName  新建文件夾的路徑，總根路徑開始，請務必以"/"結尾。例："2019/0817/"
     * @return
     */
    public PutObjectResult mkdirFolder(String bucketName, String fileName) {
        return getObsClient().putObject(bucketName, fileName, new ByteArrayInputStream(new byte[0]));
    }

    /**
     * 通過流上傳字符串為文件
     * @param bucketName  操作的桶的名稱 例："wangmarket1232311"
     * @param fileName  上傳的路徑和文件名 例："site/2010/example.txt"
     * @param content  上傳的String字符
     * @param encode  進行轉換byte時使用的編碼格式 例："UTF-8"
     * @return
     * @throws ObsException
     * @throws UnsupportedEncodingException
     */
    public PutObjectResult putStringFile(String bucketName, String fileName, String content, String encode) throws ObsException, UnsupportedEncodingException {
        return getObsClient().putObject(bucketName, fileName, new ByteArrayInputStream(content.getBytes(encode)));
    }


    /**
     * 上傳文件本地文件
     * @param bucketName 操作的桶的名稱 例："wangmarket1232311"
     * @param fileName  上傳的路徑和文件名 例："site/2010/example.txt"
     * @param localFile  需要上傳的文件
     * @return
     */
    public PutObjectResult putLocalFile(String bucketName, String fileName, File localFile) {
        return getObsClient().putObject(bucketName, fileName, localFile);
    }


    /**
     * 上傳文件流
     * @param bucketName 操作的桶的名稱 例："wangmarket1232311"
     * @param fileName 上傳的路徑和文件名 例："site/2010/example.txt"
     * @param inputStream  上傳文件的輸入流
     * @return
     */
    public PutObjectResult putFileByStream(String bucketName, String fileName, InputStream inputStream) {
        return getObsClient().putObject(bucketName, fileName, inputStream);
    }


    /**
     * 通過流上傳文件並設置指定文件屬性
     * @param bucketName  操作的桶的名稱 例："wangmarket1232311"
     * @param fileName  上傳的路徑和文件名 例："site/2010/example.txt"
     * @param inputStream  上傳文件的輸入流
     * @param metaData  上傳文件的屬性
     * @return
     */
    public PutObjectResult putFilebyInstreamAndMeta(String bucketName, String fileName, InputStream inputStream, ObjectMetadata metaData) {
        return getObsClient().putObject(bucketName, fileName, inputStream, metaData);
    }


    /**
     * OBS內對象復制
     * @param sourceBucketName   源文件的桶名稱 例："wangmarket1232311"
     * @param sourcePath  源文件的路徑和文件名 例："site/2010/example.txt"
     * @param destBucketName  目標文件的桶名稱 例："swangmarket34578345"
     * @param destPath 目標文件的路徑和文件名 例："site/2010/example_bak.txt"
     */
    public void copyObject(String sourceBucketName, String sourcePath,String destBucketName, String destPath) {
        getObsClient().copyObject(sourceBucketName, sourcePath, destBucketName, destPath);
    }


    /**
     * 獲得原生OBSBucket的訪問前綴
     * @return  桶原生的訪問前綴，即不經過CDN加速的訪問路徑
     */
    public String getOriginalUrlForOBS() {
        return "//" + obsBucketName + "." + endpoint.substring(8, endpoint.length()) + "/";
    }


    /**
     * 通過bucket的名字和連接點信息獲取bucket訪問的url
     * @param bucketName  桶的名稱 例："wangmarket21345665"
     * @param endpoint 連接點的名稱 例："obs.cn-north-1"
     * @return 根據信息獲得桶的訪問路徑 例："//wangmarket21345665.obs.cn-north-1.myhuaweicloud.com/"
     */
    public String getUrlByBucketName(String bucketName, String endpoint) {
        String url = null;
        if (url == null || url.length() == 0) {
            url = "//" + bucketName + "." +  endpoint + ".myhuaweicloud.com" + "/";
        }
        return url;
    }


    /**
     * 創建華為雲ObsBucket，默認設置為標准存儲，桶訪問權限為公共讀私有寫，同策略為所有用戶可讀桶內對象和桶內對象版本信息
     * @param obsBucketName 創建桶的名稱
     * @return  新創建的桶的名字
     */
    public String createOBSBucket(String obsBucketName) {
        // 將桶的名字進行保存
        this.obsBucketName = obsBucketName;
        ObsBucket obsBucket = new ObsBucket();
        obsBucket.setBucketName(obsBucketName);
        // 設置桶訪問權限為公共讀，默認是私有讀寫
        obsBucket.setAcl(AccessControlList.REST_CANNED_PUBLIC_READ);
        // 設置桶的存儲類型為標准存儲
        obsBucket.setBucketStorageClass(StorageClassEnum.STANDARD);
        // 創建桶
        getObsClient().createBucket(obsBucket);
        //設置桶策略
        String json = "{"
                + "\"Statement\":["
                + "{"
                + "\"Sid\":\"為授權用戶創建OBS使用的桶策略\","
                + "\"Principal\":{\"ID\" : \"*\"},"
                + "\"Effect\":\"Allow\","
                + "\"Action\":[\"GetObject\",\"GetObjectVersion\"],"
                + "\"Resource\": [\"" + obsBucketName + "/*\"]"
                + "}"
                + "]}";
        getObsClient().setBucketPolicy(obsBucketName, json);
        return obsBucketName;
    }


    /**
     * 獲取當前的桶列表
     * @return 當前桶的列表信息
     */
    public List<S3Bucket> getBuckets() {
        return getObsClient().listBuckets();
    }


    /**
     * 關閉當前的使用的OBSClient
     */
    public void closeOBSClient() {
        if(getObsClient() != null){
            try {
                getObsClient().close();
            } catch (IOException e){
                e.printStackTrace();
            }
        }
    }

    /**
     * 返回當前的創建桶的名稱 例："wangmarket1232311"
     * @return 如果有桶，那么返回桶的名稱，如 "wangmarket1232311" ，如果沒有，則返回 null
     */
    public String getObsBucketName() {
        return this.obsBucketName;
    }


    /**
     *  返回當前的桶的訪問路徑 例：“ http://cdn.leimingyun.com/”
     * @return 若已經手動設置CDN路徑返回為CND路徑，反之則為OBS原始的訪問路徑
     */
    public String getUrl() {
        // 用戶沒有配置CDN，獲的桶的原生訪問路徑
        if(url == null) {
            url = getOriginalUrlForOBS();
        }
        return url;
    }


    /**
     * 為對象設置公共讀
     * @param objectKey
     */
    public HeaderResponse setObjectAclPubilcRead(String objectKey){

        return  obsClient.setObjectAcl(obsBucketName, objectKey, AccessControlList.REST_CANNED_PUBLIC_READ);

    }

    /**
     *  獲得下載路徑
     * @param objectKey
     * @return
     */
    public String signatureUrl(String objectKey){
        long expireSeconds = 3600L;
        Map<String, String> headers = new HashMap<String, String>();

        String contentType = "text/plain";
        headers.put("Content-Type", contentType);

        TemporarySignatureRequest request = new TemporarySignatureRequest(HttpMethodEnum.PUT, expireSeconds);
        request.setBucketName(obsBucketName);
        request.setObjectKey(objectKey);
        request.setHeaders(headers);

        TemporarySignatureResponse response = obsClient.createTemporarySignature(request);

        return response.getSignedUrl();
    }

}

3.OBS業務實現類：

package com.obsupload.service;

import com.obsupload.configur.OBSHandler;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.filefilter.HiddenFileFilter;
import org.apache.commons.io.filefilter.IOFileFilter;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhengwj
 * @Description:
 * @Date: 2020/4/20 12:45
 * @Version: 1.0
 */
@Component
public class HuaweiyunOBS {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private OBSHandler obsHandler;

    @Value("${huawei.obs.accessKeyId}")
    private String accessKeyId;        //華為雲的 Access Key Id
    @Value("${huawei.obs.accessKeySecret}")
    private String accessKeySecret;    //華為雲的 Access Key Secret
    @Value("${huawei.obs.obsEndpoint}")
    private String obsEndpoint;    //格式如 obs.cn-north-1.myhuaweicloud.com
    @Value("${huawei.obs.bucketName}")
    private String bucketName;        //obs桶名

    @Value("${huawei.obs.parentPath}")
    private String parentPath;         //監控路徑

    /**
     * 以下配置用於 補數據
     */
    @Value("${local.specifiedPaths:}")
    private String specifiedPaths;  //指定上傳路徑
    @Value("${local.vice.isopen:false}")
    private boolean isOpenVice;   //是否開啟副應用

    /**
     * 獲取連接
     * @return
     */
    public OBSHandler getObsHander() {
        if(obsHandler == null) {
            obsHandler = new OBSHandler(accessKeyId,accessKeySecret,obsEndpoint);
            // 如果設置過CDN的路徑測設置為CDN路徑，沒有設置則為桶原生的訪問路徑
            //obsHandler.setUrlForCDN(Global.get("ATTACHMENT_FILE_URL"));
            // 在數據庫中讀取進行操作的桶的明恆
            obsHandler.setObsBucketName(bucketName);
            // 對桶名稱進行當前類內緩存
            bucketName = obsHandler.getObsBucketName();
        }
        return obsHandler;
    }


    /**
     * 增量錄音文件上傳OBS
     * @param file
     */
    @Async
    public void excute(File file){
        if(file.isFile() && file.getName().endsWith(".mp3")){
            int index = file.getAbsolutePath().indexOf("monitor");
            String fileName =file.getAbsolutePath().substring(index).replaceAll("\\\\", "/");
            try{
                getObsHander().putLocalFile(bucketName, fileName, file);
                getObsHander().setObjectAclPubilcRead(fileName);
                String url =  getObsHander().signatureUrl(fileName);
                log.info(url);
            }catch (Exception e){
                log.error("上傳{}失敗:{}",fileName,e);
            }finally {
    //            getObsHander().closeOBSClient();
            }
        }
    }

    /**
     * 文件監控
     *          增量上傳
     */
    public void monitoring(){
        if(isOpenVice){
            log.info("開啟上傳指定路徑下文件");
            uploadSpecified();
        }
        log.info("開啟監控.....");
        // 輪詢間隔 5 秒
        long interval = TimeUnit.SECONDS.toMillis(1);
        // 創建過濾器
        IOFileFilter directories = FileFilterUtils.and(
                FileFilterUtils.directoryFileFilter(),
                HiddenFileFilter.VISIBLE);
        IOFileFilter files    =  FileFilterUtils.and(
                FileFilterUtils.fileFileFilter(),
                FileFilterUtils.suffixFileFilter(".mp3"));
        IOFileFilter filter = FileFilterUtils.or(directories, files);
        // 使用過濾器
        FileAlterationObserver observer = new FileAlterationObserver(new File(parentPath), filter);
        //不使用過濾器
        //FileAlterationObserver observer = new FileAlterationObserver(new File(rootDir));
        observer.addListener(new FileListener());
        //創建文件變化監聽器
        FileAlterationMonitor monitor = new FileAlterationMonitor(interval, observer);
        // 開始監控
        try {
            monitor.start();
        }catch (Exception e){
            log.error("執行出錯:{}",e);
        }

    }


    /**
     * 上傳指定文件夾下的文件
     *
     */
    public void uploadSpecified(){
        if(StringUtils.isEmpty(specifiedPaths)){
            return;
        }
        String[] paths = specifiedPaths.split(",");
        for(String specifiedPath : paths){
            File specifiedFile = new File(specifiedPath);
            if(specifiedFile.isDirectory()){
                File[] files = specifiedFile.listFiles();
                for (File file : files){
                    excute(file);
                }
            }
        }
    }



    /**
     * 文件夾監聽器
     */
    class FileListener extends FileAlterationListenerAdaptor {

        private final Logger log = LoggerFactory.getLogger(getClass());

        /**
         * 文件創建執行
         */
        public void onFileCreate(File file) {
            log.info("[新建]:" + file.getAbsolutePath());
            excute(file);
        }

        /**
         * 文件刪除
         */
        public void onFileDelete(File file) {
        //    log.info("[刪除]:" + file.getAbsolutePath());
        }

        /**
         * 目錄創建
         */
        public void onDirectoryCreate(File directory) {
            log.info("[新建]:" + directory.getAbsolutePath());
        }

        public void onStart(FileAlterationObserver observer) {
            // TODO Auto-generated method stub
            super.onStart(observer);
        }
        public void onStop(FileAlterationObserver observer) {
            // TODO Auto-generated method stub
            super.onStop(observer);
        }

    }

}
4.application.properties配置

#華為雲的 Access Key Id
huawei.obs.accessKeyId=xxxxx
#華為雲的 Access Key Secret
huawei.obs.accessKeySecret=xxxxxx
#華為雲連接的地址節點
huawei.obs.obsEndpoint=xxxx
#桶的名稱
huawei.obs.bucketName=xxxx

huawei.obs.parentPath=/files/monitor/
#huawei.obs.parentPath=D://files/monitor/

#上傳指定路徑下的文件  用於監控主程序掛掉之后補數據
#local.vice.isopen=false
#local.specifiedPaths=/files/monitor/20200428/,/files/monitor/20200427/


大致實現如上所示，完成代碼請查看github：https://github.com/wojozer/obs-upload
歡迎留言交流分享！