import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
public void storeLargeFile(String key, File file, byte[] md5Hash)
    throws IOException {
  S3Object object = new S3Object(key);
  object.setDataInputFile(file);
  object.setContentType("binary/octet-stream");
  object.setContentLength(file.length());
  object.setServerSideEncryptionAlgorithm(serverSideEncryptionAlgorithm);
  if (md5Hash != null) {
    object.setMd5Hash(md5Hash);
  }

  List<StorageObject> objectsToUploadAsMultipart =
      new ArrayList<StorageObject>();
  objectsToUploadAsMultipart.add(object);
  MultipartUtils mpUtils = new MultipartUtils(multipartBlockSize);

  try {
    mpUtils.uploadObjects(bucket.getName(), s3Service,
                          objectsToUploadAsMultipart, null);
  } catch (Exception e) {
    handleException(e, key);
  }
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
/**
 * Test upload object as multiparts.
 *
 * @throws Exception the exception
 */
@Test
public void testUploadObjectAsMultiparts() throws Exception {
	s3RESTService.createBucket(AWS_S3_BUCKET);//create bucket for test
	// Create a large (11 MB) file
       File file = File.createTempFile("testLargeFile", ".txt");
       BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(file));
       int offset = 0;
       while (offset < 11 * 1024 * 1024) {
           bos.write((offset++ % 256));
       }
       bos.close();
       
       StorageObject objectKey = new StorageObject(file);
	s3RESTService.uploadObjectAsMultiparts(AWS_S3_BUCKET,objectKey ,
			MultipartUtils.MIN_PART_SIZE);
	S3Object s3Obj = s3RESTService.getObject(AWS_S3_BUCKET, objectKey.getKey());
	assertEquals(true, s3Obj!=null);
	if(file!=null){
		file.delete();//Delete when operation completes.
	}
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
@Override
public void uploadObjectsAsMultiparts(final String bucketName, final List<StorageObject> objects)
		throws ServiceException, Exception {
	LOGGER.info("putObjectsAsMultiparts invoked, bucketName: {}", bucketName);
	final MultipartUtils multipartUtils = new MultipartUtils();
       multipartUtils.uploadObjects(bucketName, s3Service, objects, null);	
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
@Override
public void uploadDirectoryAsMultiparts(final String bucketName, final File dirPath)
		throws ServiceException, Exception {
	LOGGER.info("uploadDirectoryAsMultiparts invoked, bucketName: {} and dirPath: {}", bucketName, dirPath);
	final Set<File> setOfFilePath = DirectoryTraverser.getFileUris(dirPath);
	final List<StorageObject> objects = new ArrayList<StorageObject>();
	for (final File file : setOfFilePath) {
		objects.add(ObjectUtils.createObjectForUpload(file.getName(), file, null, // encryptionUtil
				false // gzipFile
		));
	}
	
	final MultipartUtils multipartUtils = new MultipartUtils();
	multipartUtils.uploadObjects(bucketName, s3Service, objects, null);
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
@JsonCreator
public S3FileSink(
        @JsonProperty("localFileSink") LocalFileSink localFileSink,
        @JsonProperty("bucket") String bucket,
        @JsonProperty("s3Endpoint") String s3Endpoint,
        @JsonProperty("maxPartSize") long maxPartSize,
        @JsonProperty("concurrentUpload") int concurrentUpload,
        @JsonProperty("notice") Notice notice,
        @JsonProperty("prefixFormatter") RemotePrefixFormatter prefixFormatter,
        @JsonProperty("batchUpload") boolean batchUpload,
        @JsonProperty("s3Acl") String s3Acl,
        @JsonProperty("s3AclRetries") int s3AclRetries,
        @JacksonInject MultipartUtils mpUtils,
        @JacksonInject AWSCredentialsProvider credentialProvider) {
    super(localFileSink, prefixFormatter, concurrentUpload, batchUpload);

    this.bucket = bucket;
    this.s3Endpoint = s3Endpoint == null ? "s3.amazonaws.com" : s3Endpoint;
    this.maxPartSize = maxPartSize == 0 ? 20 * 1024 * 1024 : maxPartSize;
    this.notice = notice == null ? new QueueNotice<String>() : notice;

    this.mpUtils = mpUtils;
    this.credentialsProvider = credentialProvider;

    this.s3Acl = s3Acl;
    this.s3AclRetries = s3AclRetries > 0 ? s3AclRetries : 5;

    Preconditions.checkNotNull(bucket, "bucket is needed");
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
private Injector getInjector() {
    return Guice.createInjector(
        new SuroSinkPlugin(),
        new AbstractModule() {
                @Override
                protected void configure() {
                    bind(ObjectMapper.class).to(DefaultObjectMapper.class);
                    bind(AWSCredentialsProvider.class)
                        .toInstance(new AWSCredentialsProvider() {
                            @Override
                            public AWSCredentials getCredentials() {
                                return new AWSCredentials() {
                                    @Override
                                    public String getAWSAccessKeyId() {
                                        return "accessKey";
                                    }

                                    @Override
                                    public String getAWSSecretKey() {
                                        return "secretKey";
                                    }
                                };
                            }

                            @Override
                            public void refresh() {
                            }
                        });

                    MultipartUtils mpUtils = mock(MultipartUtils.class);
                    try {
                        doAnswer(new Answer() {
                            @Override
                            public Object answer(InvocationOnMock invocation) throws Throwable {
                                Thread.sleep(1000);
                                return null;
                            }
                        }).when(mpUtils).uploadObjects(
                            any(String.class),
                            any(RestS3Service.class),
                            any(List.class),
                            any(S3ServiceEventListener.class));

                        bind(MultipartUtils.class).toInstance(mpUtils);
                    } catch (Exception e) {
                        Assert.fail(e.getMessage());
                    }
                    bind(SpaceChecker.class).toInstance(mock(SpaceChecker.class));
                }
        }
    );
}
 

import org.jets3t.service.utils.MultipartUtils; //导入依赖的package包/类
private void multipartUpload(S3Object object) throws Exception {

    List<StorageObject> objectsToUploadAsMultipart = Arrays.<StorageObject>asList(object);

    MultipartUtils mpUtils = new MultipartUtils(configuration.getUploadPartSize());
    mpUtils.uploadObjects(s3Bucket.getName(), s3Service, objectsToUploadAsMultipart, null);
  }