Java 文件分块上传客户端和服务器端源代码

Java 文件分块上传客户端源代码  http://blog.csdn.net/defonds/article/details/8575893

MIME协议(中文版).doc

        本博客介绍如何进行文件的分块上传。本文侧重介绍客户端，服务器端请参考博客《Java 文件分块上传服务器端源代码》。建议读者朋友在阅读本文代码前先了解一下 MIME 协议。

        所谓分块上传并非把大文件进行物理分块，然后挨个上传，而是依次读取大文件的一部分文件流进行上传。分块，倒不如说分流比较切实。本文通过一个项目中的示例，说明使用 Apache 的 HttpComponents/HttpClient 对大文件进行分块上传的过程。示例使用的版本是 HttpComponents Client 4.2.1。
        本文仅以一小 demo 功能性地解释 HttpComponents/HttpClient 分块上传，没有考虑 I/O 关闭、多线程等资源因素，读者可以根据自己的项目酌情处理。
        本文核心思想及流程：以 100 MB 大小为例，大于 100 MB 的进行分块上传，否则整块上传。对于大于 100 MB 的文件，又以 100 MB 为单位进行分割，保证每次以不大于 100 MB 的大小进行上传。比如 304 MB 的一个文件会分为 100 MB、100 MB、100 MB、4 MB 等四块依次上传。第一次读取 0 字节开始的 100 MB 个字节，上传；第二次读取第 100 MB 字节开始的 100 MB 个字节，上传；第三次读取第 200 MB 字节开始的 100 MB 个字节，上传；第四次读取最后剩下的 4 MB 个字节进行上传。

        自定义的 ContentBody 源码如下，其中定义了流的读取和输出：

package com.defonds.rtupload.common.util.block;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.io.RandomAccessFile;

import org.apache.http.entity.mime.content.AbstractContentBody;

import com.defonds.rtupload.GlobalConstant;

public class BlockStreamBody extends AbstractContentBody {

    //给MultipartEntity看的2个参数
    private long blockSize = 0;//本次分块上传的大小
    private String fileName = null;//上传文件名
    //writeTo需要的3个参数
    private int blockNumber = 0, blockIndex = 0;//blockNumber分块数;blockIndex当前第几块
    private File targetFile = null;//要上传的文件

    private BlockStreamBody(String mimeType) {
        super(mimeType);
        // TODO Auto-generated constructor stub
    }

    /**
     * 自定义的ContentBody构造子
     * @param blockNumber分块数
     * @param blockIndex当前第几块
     * @param targetFile要上传的文件
     */
    public BlockStreamBody(int blockNumber, int blockIndex, File targetFile) {
        this("application/octet-stream");
        this.blockNumber = blockNumber;//blockNumber初始化
        this.blockIndex = blockIndex;//blockIndex初始化
        this.targetFile = targetFile;//targetFile初始化
        this.fileName = targetFile.getName();//fileName初始化
        //blockSize初始化
        if (blockIndex < blockNumber) {//不是最后一块，那就是固定大小了
            this.blockSize = GlobalConstant.CLOUD_API_LOGON_SIZE;
        } else {//最后一块
            this.blockSize = targetFile.length() - GlobalConstant.CLOUD_API_LOGON_SIZE * (blockNumber - 1);
        }
    }

    @Override
    public void writeTo(OutputStream out) throws IOException {
        byte b[] = new byte[1024];//暂存容器
        RandomAccessFile raf  = new RandomAccessFile(targetFile, "r");//负责读取数据
        if (blockIndex == 1) {//第一块
            int n = 0;
            long readLength = 0;//记录已读字节数
            while (readLength <= blockSize - 1024) {//大部分字节在这里读取
                n = raf.read(b, 0, 1024);
                readLength += 1024;
                out.write(b, 0, n);
            }
            if (readLength <= blockSize) {//余下的不足 1024 个字节在这里读取
                n = raf.read(b, 0, (int)(blockSize - readLength));
                out.write(b, 0, n);
            }
        } else if (blockIndex < blockNumber) {//既不是第一块，也不是最后一块
            raf.seek(GlobalConstant.CLOUD_API_LOGON_SIZE * (blockIndex - 1));//跳过前[块数*固定大小 ]个字节
            int n = 0;
            long readLength = 0;//记录已读字节数
            while (readLength <= blockSize - 1024) {//大部分字节在这里读取
                n = raf.read(b, 0, 1024);
                readLength += 1024;
                out.write(b, 0, n);
            }
            if (readLength <= blockSize) {//余下的不足 1024 个字节在这里读取
                n = raf.read(b, 0, (int)(blockSize - readLength));
                out.write(b, 0, n);
            }
        } else {//最后一块
            raf.seek(GlobalConstant.CLOUD_API_LOGON_SIZE * (blockIndex - 1));//跳过前[块数*固定大小 ]个字节
            int n = 0;
            while ((n = raf.read(b, 0, 1024)) != -1) {
                out.write(b, 0, n);
            }
        }

        //TODO 最后不要忘掉关闭out/raf
    }

    @Override
    public String getCharset() {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public String getTransferEncoding() {
        // TODO Auto-generated method stub
        return "binary";
    }

    @Override
    public String getFilename() {
        // TODO Auto-generated method stub
        return fileName;
    }

    @Override
    public long getContentLength() {
        // TODO Auto-generated method stub
        return blockSize;
    }

}

        在自定义的 HttpComponents/HttpClient 工具类 HttpClient4Util 里进行分块上传的封装：

public static String restPost(String serverURL, File targetFile,Map<String, String> mediaInfoMap){

    String content ="";
    try {
        DefaultHttpClient httpClient = new DefaultHttpClient();
        HttpPost post = new HttpPost(serverURL +"?");
        httpClient.getParams().setParameter("http.socket.timeout",60*60*1000);
        MultipartEntity mpEntity = new MultipartEntity();
        List<String> keys = new ArrayList<String>(mediaInfoMap.keySet());
        Collections.sort(keys, String.CASE_INSENSITIVE_ORDER);
        for (Iterator<String> iterator = keys.iterator(); iterator.hasNext();) {
            String key = iterator.next();
            if (StringUtils.isNotBlank(mediaInfoMap.get(key))) {
                mpEntity.addPart(key, new StringBody(mediaInfoMap.get(key)));
            }
        }

        if(targetFile!=null&&targetFile.exists()){
            ContentBody contentBody = new FileBody(targetFile);
            mpEntity.addPart("file", contentBody);
        }
        post.setEntity(mpEntity);


        HttpResponse response = httpClient.execute(post);
        content = EntityUtils.toString(response.getEntity());
        httpClient.getConnectionManager().shutdown();
    } catch (Exception e) {
        e.printStackTrace();
    }
    System.out.println("=====RequestUrl==========================
"
            +getRequestUrlStrRest(serverURL, mediaInfoMap).replaceAll("&fmt=json", ""));
    System.out.println("=====content==========================
"+content);
    return content.trim();
}

        其中 "file" 是分块上传服务器对分块文件参数定义的名字。细心的读者会发现，整块文件上传直接使用 Apache 官方的 InputStreamBody，而分块才使用自定义的 BlockStreamBody。

        最后调用 HttpClient4Util 进行上传：

public static Map<String, String> uploadToDrive(
        Map<String, String> params, String domain) {

    File targetFile = new File(params.get("filePath"));
    long targetFileSize = targetFile.length();
    int mBlockNumber = 0;
    if (targetFileSize < GlobalConstant.CLOUD_API_LOGON_SIZE) {
        mBlockNumber = 1;
    } else {
        mBlockNumber = (int) (targetFileSize / GlobalConstant.CLOUD_API_LOGON_SIZE);
        long someExtra = targetFileSize
                % GlobalConstant.CLOUD_API_LOGON_SIZE;
        if (someExtra > 0) {
            mBlockNumber++;
        }
    }
    params.put("blockNumber", Integer.toString(mBlockNumber));

    if (domain != null) {
        LOG.debug("Drive---domain=" + domain);
        LOG.debug("drive---url=" + "http://" + domain + "/sync"
                + GlobalConstant.CLOUD_API_PRE_UPLOAD_PATH);
    } else {
        LOG.debug("Drive---domain=null");
    }
    String responseBodyStr = HttpClient4Util.getRest("http://" + domain
            + "/sync" + GlobalConstant.CLOUD_API_PRE_UPLOAD_PATH, params);

    ObjectMapper mapper = new ObjectMapper();
    DrivePreInfo result;
    try {
        result = mapper.readValue(responseBodyStr, ArcDrivePreInfo.class);
    } catch (IOException e) {
        LOG.error("Drive.preUploadToArcDrive error.", e);
        throw new RtuploadException(GlobalConstant.ERROR_CODE_13001);// TODO
    }
    // JSONObject jsonObject = JSONObject.fromObject(responseBodyStr);
    if (Integer.valueOf(result.getRc()) == 0) {
        int uuid = result.getUuid();
        String upsServerUrl = result.getUploadServerUrl().replace("https",
                "http");
        if (uuid != -1) {
            upsServerUrl = upsServerUrl
                    + GlobalConstant.CLOUD_API_UPLOAD_PATH;
            params.put("uuid", String.valueOf(uuid));

            for (int i = 1; i <= mBlockNumber; i++) {
                params.put("blockIndex", "" + i);
                HttpClient4Util.restPostBlock(upsServerUrl, targetFile,
                        params);//
            }
        }
    } else {
        throw new RtuploadException(GlobalConstant.ERROR_CODE_13001);// TODO
    }
    return null;
}

        其中 params 这个 Map 里封装的是服务器分块上传所需要的一些参数，而上传块数也在这里进行确定。
        本文中的示例经本人测试能够上传大文件成功，诸如 *.mp4 的文件上传成功没有出现任何问题。如果读者朋友测试时遇到问题无法上传成功，请在博客后跟帖留言，大家共同交流下。本文示例肯定还存在很多不足之处，如果读者朋友发现还请留言指出，笔者先行谢过了。

 

Java 文件分块上传服务器端源代码 http://blog.csdn.net/defonds/article/details/8647129

 

 本博客将介绍如何进行文件的分块上传。如果读者还想了解文件的“分块”下载相关内容可以去参考博客《Java 服务器端支持断点续传的源代码【支持快车、迅雷】》。
        本文侧重介绍服务器端，客户端端请参考本篇博客的姊妹篇《Java 文件分块上传客户端源代码》，关于分块上传的思想及其流程，已在该博客中进行了详细说明，这里不再赘述。

        直接上代码。接收客户端 HTTP 分块上传请求的 Spring MVC 控制器源代码如下：

@Controller
public class UploadController extends BaseController {

    private static final Log log = LogFactory.getLog(UploadController.class);
    private UploadService uploadService;
    private AuthService authService;

    /**
     * 大文件分成小文件块上传,一次传递一块,最后一块上传成功后,将合并所有已经上传的块,保存到File Server
     * 上相应的位置,并返回已经成功上传的文件的详细属性. 当最后一块上传完毕，返回上传成功的信息。此时用getFileList查询该文件，
     * 该文件的uploadStatus为2。client请自行处理该状态下文件如何显示。(for UPS Server)
     *
     */
    @RequestMapping("/core/v1/file/upload")
    @ResponseBody
    public Object upload(HttpServletResponse response,
            @RequestParam(value = "client_id", required = false) String appkey,
            @RequestParam(value = "sig", required = false) String appsig,
            @RequestParam(value = "token", required = false) String token,
            @RequestParam(value = "uuid", required = false) String uuid,
            @RequestParam(value = "block", required = false) String blockIndex,
            @RequestParam(value = "file", required = false) MultipartFile multipartFile,
            @RequestParam Map<String, String> parameters) {

        checkEmpty(appkey, BaseException.ERROR_CODE_16002);
        checkEmpty(token, BaseException.ERROR_CODE_16007);
        checkEmpty(uuid, BaseException.ERROR_CODE_20016);
        checkEmpty(blockIndex, BaseException.ERROR_CODE_20006);
        checkEmpty(appsig, BaseException.ERROR_CODE_10010);
        if (multipartFile == null) {
            throw new BaseException(BaseException.ERROR_CODE_20020);// 上传文件不存在
        }
        Long uuidL = parseLong(uuid, BaseException.ERROR_CODE_20016);
        Integer blockIndexI = parseInt(blockIndex, BaseException.ERROR_CODE_20006);

        Map<String, Object> appMap = getAuthService().validateSigature(parameters);

        AccessToken accessToken = CasUtil.checkAccessToken(token, appMap);
        Long uid = accessToken.getUid();
        String bucketUrl = accessToken.getBucketUrl();
        // 从上传目录拷贝文件到工作目录
        String fileAbsulutePath = null;
        try {
            fileAbsulutePath = this.copyFile(multipartFile.getInputStream(), multipartFile.getOriginalFilename());
        } catch (IOException ioe) {
            log.error(ioe.getMessage(), ioe);
            throw new BaseException(BaseException.ERROR_CODE_20020);// 上传文件不存在
        }
        File uploadedFile = new File(Global.UPLOAD_TEMP_DIR + fileAbsulutePath);
        checkEmptyFile(uploadedFile);// file 非空验证

        Object rs = uploadService.upload(uuidL, blockIndexI, uid, uploadedFile, bucketUrl);
        setHttpStatusOk(response);
        return rs;
    }

    // TODO 查看下这里是否有问题
    // 上传文件非空验证
    private void checkEmptyFile(File file) {
        if (file == null || file.getAbsolutePath() == null) {
            throw new BaseException(BaseException.ERROR_CODE_20020);// 上传文件不存在
        }
    }

    /**
     * 写文件到本地文件夹
     *
     * @throws IOException
     *             返回生成的文件名
     */
    private String copyFile(InputStream inputStream, String fileName) {
        OutputStream outputStream = null;
        String tempFileName = null;
        int pointPosition = fileName.lastIndexOf(".");
        if (pointPosition < 0) {// myvedio
            tempFileName = UUID.randomUUID().toString();// 94d1d2e0-9aad-4dd8-a0f6-494b0099ff26
        } else {// myvedio.flv
            tempFileName = UUID.randomUUID() + fileName.substring(pointPosition);// 94d1d2e0-9aad-4dd8-a0f6-494b0099ff26.flv
        }
        try {
            outputStream = new FileOutputStream(Global.UPLOAD_TEMP_DIR + tempFileName);
            int readBytes = 0;
            byte[] buffer = new byte[10000];
            while ((readBytes = inputStream.read(buffer, 0, 10000)) != -1) {
                outputStream.write(buffer, 0, readBytes);
            }
            return tempFileName;
        } catch (IOException ioe) {
            // log.error(ioe.getMessage(), ioe);
            throw new BaseException(BaseException.ERROR_CODE_20020);// 上传文件不存在
        } finally {
            if (outputStream != null) {
                try {
                    outputStream.close();
                } catch (IOException e) {
                }
            }
            if (inputStream != null) {
                try {
                    inputStream.close();
                } catch (IOException e) {
                }
            }

        }

    }

    /**
     * 测试此服务是否可用
     *
     * @param response
     * @return
     * @author zwq7978
     */
    @RequestMapping("/core/v1/file/testServer")
    @ResponseBody
    public Object testServer(HttpServletResponse response) {
        setHttpStatusOk(response);
        return Global.SUCCESS_RESPONSE;
    }

    public UploadService getUploadService() {
        return uploadService;
    }

    public void setUploadService(UploadService uploadService) {
        this.uploadService = uploadService;
    }

    public void setAuthService(AuthService authService) {
        this.authService = authService;
    }

    public AuthService getAuthService() {
        return authService;
    }

}

        比如要上传的文件是 test450k.mp4。对照《Java 文件分块上传客户端源代码》中分块上传服务器对分块文件参数定义的名字"file"，upload 方法里使用的是 MultipartFile 接收该对象。对于每次的 HTTP 请求，使用 copyFile 方法将文件流输出到服务器本地的一个临时文件夹里，比如作者的是 D:/defonds/syncPath/uploadTemp，该文件下会有 50127019-b63b-4a54-8f53-14efd1e58ada.mp4 临时文件生成用于保存上传文件流。

        分块依次上传。当所有块都上传完毕之后，将这些临时文件都转移到服务器指定目录中，比如作者的这个目录是 D:/defonds/syncPath/file，在该文件夹下会有/1/temp_dir_5_1 目录生成，而 uploadTemp 的临时文件则被挨个转移到这个文件夹下，生成形如 5.part0001 的文件。以下是文件转移的源代码：

/**
 * 把所有块从临时文件目录移到指定本地目录或S2/S3
 *
 * @param preUpload
 */
private void moveBlockFiles(BlockPreuploadFileInfo preUpload) {
    @SuppressWarnings("unchecked")
    String[] s3BlockUrl=new String[preUpload.getBlockNumber()];
    String[] localBlockUrl=new String[preUpload.getBlockNumber()];//本地的块文件路径    以便以后删除
    List<BlockUploadInfo> blocks = (List<BlockUploadInfo>) getBaseDao().queryForList(
            "upload.getBlockUploadFileByUuid", preUpload.getUuid());

    String tempDirName = SyncUtil.getTempDirName(preUpload.getUuid(), preUpload.getUid());
    String parentPath = Global.UPLOAD_ABSOLUTE_PAHT_ + Global.PATH_SEPARATIVE_SIGN
            + String.valueOf(preUpload.getUid());
    String dirPath = parentPath + Global.PATH_SEPARATIVE_SIGN + tempDirName;
    new File(dirPath).mkdirs();//创建存放块文件的文件夹 （本地）
    int j=0;
    for (BlockUploadInfo info : blocks) {
        try {
            String strBlockIndex = createStrBlockIndex(info.getBlockIndex());
            String suffixPath = preUpload.getUuid() + ".part" + strBlockIndex;
            String tempFilePath = info.getTempFile();
            File tempFile = new File(tempFilePath);
            File tmpFile = new File(dirPath + suffixPath);
            if (tmpFile.exists()) {
                FileUtils.deleteQuietly(tmpFile);
            }
            FileUtils.moveFile(tempFile, tmpFile);
            localBlockUrl[j]=dirPath + suffixPath;
            j++;
            info.setStatus(Global.MOVED_TO_NEWDIR);
            getBaseDao().update("upload.updateBlockUpload", info);
            if (log.isInfoEnabled())
                log.info(preUpload.getUuid() + " " + info.getBuId() + " moveBlockFiles");
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            throw new BaseException("file not found");
        }
    }
    preUpload.setLocalBlockUrl(localBlockUrl);
    preUpload.setDirPath(dirPath);
    preUpload.setStatus(Global.MOVED_TO_NEWDIR);
    getBaseDao().update("upload.updatePreUploadInfo", preUpload);
}

private String createStrBlockIndex(int blockIndex) {
    String strBlockIndex;
    if (blockIndex < 10) {
        strBlockIndex = "000" + blockIndex;
    } else if (10 <= blockIndex && blockIndex < 100) {
        strBlockIndex = "00" + blockIndex;
    } else if (100 <= blockIndex && blockIndex < 1000) {
        strBlockIndex = "0" + blockIndex;
    } else {
        strBlockIndex = "" + blockIndex;
    }
    return strBlockIndex;
}

        最后是文件的组装源代码：

/**
 * 组装文件
 *
 */
private void assembleFileWithBlock(BlockPreuploadFileInfo preUpload) {
    String dirPath = preUpload.getDirPath();
    // 开始在指定目录组装文件
    String uploadedUrl = null;
    String[] separatedFiles;
    String[][] separatedFilesAndSize;
    int fileNum = 0;
    File file = new File(dirPath);
    separatedFiles = file.list();
    separatedFilesAndSize = new String[separatedFiles.length][2];
    Arrays.sort(separatedFiles);
    fileNum = separatedFiles.length;
    for (int i = 0; i < fileNum; i++) {
        separatedFilesAndSize[i][0] = separatedFiles[i];
        String fileName = dirPath + separatedFiles[i];
        File tmpFile = new File(fileName);
        long fileSize = tmpFile.length();
        separatedFilesAndSize[i][1] = String.valueOf(fileSize);
    }

    RandomAccessFile fileReader = null;
    RandomAccessFile fileWrite = null;
    long alreadyWrite = 0;
    int len = 0;
    byte[] buf = new byte[1024];
    try {
        uploadedUrl = Global.UPLOAD_ABSOLUTE_PAHT_ + Global.PATH_SEPARATIVE_SIGN + preUpload.getUid() + Global.PATH_SEPARATIVE_SIGN + preUpload.getUuid();
        fileWrite = new RandomAccessFile(uploadedUrl, "rw");
        for (int i = 0; i < fileNum; i++) {
            fileWrite.seek(alreadyWrite);
            // 读取
            fileReader = new RandomAccessFile((dirPath + separatedFilesAndSize[i][0]), "r");
            // 写入
            while ((len = fileReader.read(buf)) != -1) {
                fileWrite.write(buf, 0, len);
            }
            fileReader.close();
            alreadyWrite += Long.parseLong(separatedFilesAndSize[i][1]);
        }
        fileWrite.close();
        preUpload.setStatus(Global.ASSEMBLED);
        preUpload.setServerPath(uploadedUrl);
        getBaseDao().update("upload.updatePreUploadInfo", preUpload);

        if(Global.BLOCK_UPLOAD_TO!=Global.BLOCK_UPLOAD_TO_LOCAL)
        {
        //组装完毕没有问题  删除掉S2/S3上的block
        String[] path=preUpload.getS3BlockUrl();
        for (String string : path) {
            try {
                if(Global.BLOCK_UPLOAD_TO==Global.BLOCK_UPLOAD_TO_S2)
                {
                S2Util.deleteFile(preUpload.getBucketUrl(), string);
                }else
                {
                S3Util.deleteFile(preUpload.getBucketUrl(), string);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
        }
        if (log.isInfoEnabled())
            log.info(preUpload.getUuid() + " assembleFileWithBlock");
    } catch (IOException e) {
        log.error(e.getMessage(), e);
        try {
            if (fileReader != null) {
                fileReader.close();
            }
            if (fileWrite != null) {
                fileWrite.close();
            }
        } catch (IOException ex) {
            log.error(e.getMessage(), e);
        }
    }

}

        BlockPreuploadFileInfo 是我们自定义的业务文件处理 bean。
        OK，分块上传的服务器、客户端源代码及其工作流程至此已全部介绍完毕，以上源代码全部是经过项目实践过的，大部分现在仍运行于一些项目之中。有兴趣的朋友可以自己动手，将以上代码自行改造，看看能否运行成功。如果遇到问题可以在本博客下跟帖留言，大家一起讨论讨论。