使用java实现s3协议多线程下载对象
package GeneratePresignedUrlAndUploadObject;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.Protocol;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.OutputSerialization;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
public class Filedownload {
//创建一个计数器锁。初始值为线程数量，每执行结束一个线程后计数器减去1 ，当计数器为0的时候await等待的线程会被唤醒继续执行。
public static CountDownLatch latch = new CountDownLatch(10);
public static void main(String[] args) throws IOException {
String akey = "BXM4ATD5KZ3EJLP8IEM0";
String skey = "NfkfMSzAv8XETXmpsj8XDrAjB3fIuJnkxE3NzbGF";
String endpoint = "http://10.255.20.180:8060";
String bucket = "bucket1";
String filename = "centos-74.qcow2";
RandomAccessFile file = null;
try {
AWSCredentials credentials = new BasicAWSCredentials(akey, skey);
ClientConfiguration clientConfig = new ClientConfiguration();
clientConfig.setProtocol(Protocol.HTTP);
AmazonS3 s3Client = new AmazonS3Client(credentials, clientConfig);
s3Client.setEndpoint(endpoint);
//获取对象大小
ObjectMetadata metadata = s3Client.getObjectMetadata(bucket, filename);
long filesize = metadata.getInstanceLength();
System.out.println("文件大小："+filesize);
ExecutorService service = Executors.newFixedThreadPool(10);
long length = filesize;
long packageLength = length/10;
long leftLength = length%10;
long pos = 0;
long end = packageLength ;
file = new RandomAccessFile(filename,"rw");
//计算每个线程请求文件的开始和结束位置
for (int i=0;i<10;i++) {
if (leftLength >0) {
packageLength ++;
leftLength --;
}
System.out.println("pos: "+pos +" endpos: "+packageLength );
service.execute(new download(pos, packageLength,file));
pos = packageLength ;
packageLength = packageLength +end;
}
//等待其他线程结束后继续向下执行
try {
latch.await();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//关闭线程池
service.shutdown();
}
catch(AmazonServiceException e) {
// 服务端错误
e.printStackTrace();
}finally {
if(file != null) {
file.close();
}
}
}
}
class download implements Runnable{
private long from;
private long end;
private RandomAccessFile file;
public download(long from,long end,RandomAccessFile file){
this.from = from;
this.end = end;
this.file = file;
}
public void run() {
String akey = "BXM4ATD5KZ3EJLP8IEM0";
String skey = "NfkfMSzAv8XETXmpsj8XDrAjB3fIuJnkxE3NzbGF";
String endpoint = "http://10.255.20.180:8060";
String bucket = "bucket1";
String filename = "centos-74.qcow2";
S3Object objectPortion = null;
InputStream input =null;
BufferedInputStream buffer =null;
try {
AWSCredentials credentials = new BasicAWSCredentials(akey, skey);
ClientConfiguration clientConfig = new ClientConfiguration();
clientConfig.setProtocol(Protocol.HTTP);
clientConfig.setConnectionTimeout(300000);
clientConfig.setMaxConnections(100);
AmazonS3 s3Client = new AmazonS3Client(credentials, clientConfig);
s3Client.setEndpoint(endpoint);
//获取对象指定范围的流写入文件
GetObjectRequest rangeObjectRequest = new GetObjectRequest(bucket, filename).withRange(from,end);
objectPortion = s3Client.getObject(rangeObjectRequest);
input = objectPortion.getObjectContent();
buffer = new BufferedInputStream(input);
byte[] buf = new byte[1024];
int len;
long start = this.from;
long stop = this.end;
for(;;) {
if ((len = buffer.read(buf))==-1) {
break;
}
synchronized (file) {
file.seek(from);
file.write(buf, 0, len);
}
from += len;
}
System.out.println("文件片段 "+ start +"~"+stop+"下载完成");
new Filedownload().latch.countDown();//线程结束计数器减1
}catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally {
if(buffer != null) {
try {
buffer.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(input != null) {
try {
input.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(objectPortion != null) {
try {
objectPortion.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
}