Android使用RxJava并行上传图片到阿里云OSS
2018-08-18 本文已影响659人
丶丶TkoRn丶丶
前言
不要上错来车了，这里只是简单的使用 RxJava来多线程上传图片至OSS。 有需要的往下看，有哪里不足的希望各位提出宝贵的意见。
OSS 配置
参考：官方文档
1. 添加依赖
dependencies {
compile 'com.aliyun.dpa:oss-android-sdk:+'
compile 'com.squareup.okhttp3:okhttp:3.4.1'
compile 'com.squareup.okio:okio:1.9.0'
}
2. 权限设置
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<uses-permission android:name="android.permission.ACCESS_WIFI_STATE" />
<uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
<uses-permission android:name="android.permission.MOUNT_UNMOUNT_FILESYSTEMS" />
3. 混淆设置
-keep class com.alibaba.sdk.android.oss.** { *; }
-dontwarn okio.**
-dontwarn org.apache.commons.codec.binary.**
OSS 初始化
这里使用的是SDK代理更新token的方式，您需要告诉SDK如何获取Token。在SDK的应用中，您需要实现一个回调，这个回调通过您实现的方式去获取一个Federation Token(即StsToken)，然后返回。SDK会利用这个Token来进行加签处理，并在需要更新时主动调用这个回调获取Token
public static void initOSS(){
OSSCredentialProvider provider = new OSSFederationCredentialProvider() {
@Override
public OSSFederationToken getFederationToken() throws ClientException {
final StsTokenBean.CredentialsBean[] credentialsBean = new StsTokenBean.CredentialsBean[1];
final String ENDPOINT = "http://oss-cn-hangzhou.aliyuncs.com";
RequestBody body = RequestBody.create(MediaType.parse("application/json;charset=UTF-8"), "{}");
//从自己服务器获取临时的StsTkoen,使用同步的
RetrofitFactory.Companion
.createInstance(true)
.create(CommonApi.class)
.getStsToken(body)
.subscribe(new Observer<BaseBean<StsTokenBean>>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(BaseBean<StsTokenBean> stsTokenBeanBaseBean) {
if (stsTokenBeanBaseBean != null && stsTokenBeanBaseBean.getContent()!= null && stsTokenBeanBaseBean.getContent().getCredentials() != null){
credentialsBean[0] = stsTokenBeanBaseBean.getContent().getCredentials();
}
}
});
if (credentialsBean[0] != null){
return new OSSFederationToken(credentialsBean[0].getAccessKeyId(),credentialsBean[0].getAccessKeySecret(),credentialsBean[0].getSecurityToken(),credentialsBean[0].getExpiration());
}
return null;
}
};
ClientConfiguration conf = new ClientConfiguration();
conf.setConnectionTimeout(15 * 1000); // 连接超时，默认15秒
conf.setSocketTimeout(15 * 1000); // socket超时，默认15秒
conf.setMaxConcurrentRequest(5); // 最大并发请求书，默认5个
conf.setMaxErrorRetry(2); // 失败后最大重试次数，默认2次
oss = new OSSClient(STFApplication.Companion.getContext(), ENDPOINT, provider, conf);
}
OSS 上传图片
这里只是使用了简单上传本地文件的同步接口，有需要了解更多的请向往官方文档
public static Observable<PutObjectResult> upImages(final List<UpImageBean> pathList){
if (oss == null){
Log.e(TAG,"oss == null 没有初始化OSS");
return null;
}
if (pathList == null || pathList.size() == 0){
Log.e(TAG,"pathList == null 图片数组不能为空");
return null;
}
int threadNum = Runtime.getRuntime().availableProcessors()+1;
final ExecutorService executor = Executors.newFixedThreadPool(threadNum);
//为了方便阅读代码这里就没使用lambda
return Observable.from(pathList)
.flatMap(new Func1<UpImageBean, Observable<PutObjectResult>>() {
@Override
public Observable<PutObjectResult> call(final UpImageBean bean) {
return Observable.just(bean)
.map(new Func1<UpImageBean, PutObjectResult>() {
@Override
public PutObjectResult call(UpImageBean upImageBean) {
PutObjectRequest put = null;
try {
put = new PutObjectRequest(BUCKE_NAME, upImageBean.getName(), upImageBean.getPath());
return oss.putObject(put);
} catch (ClientException e) {
e.printStackTrace();
} catch (ServiceException e) {
e.printStackTrace();
}
return null;
}
})
.subscribeOn(Schedulers.from(executor));
}
})
.subscribeOn(Schedulers.io())
.finallyDo(() -> {
executor.shutdown();
})
.observeOn(AndroidSchedulers.mainThread());
}
注意：这里是没有做背压处理的，应为项目需求最大只上传9张图片。需要做背压处理的同学就要自己做改进了。
OSSUtils.java
献上全部代码给各位观众老爷，注意把 ENDPOINT ，BUCKE_NAME 改成自己的。
public class OSSUtils {
private static final String ENDPOINT = "http://oss-cn-hangzhou.aliyuncs.com";
private static final String BUCKE_NAME = "xxxxxx";
private static OSS oss;
private static final String TAG = OSSUtils.class.getSimpleName();
public static void initOSS(){
OSSCredentialProvider credentialProvider = new OSSFederationCredentialProvider(){
@Override
public OSSFederationToken getFederationToken() throws ClientException {
return null;
}
};
OSSCredentialProvider provider = new OSSFederationCredentialProvider() {
@Override
public OSSFederationToken getFederationToken() throws ClientException {
final StsTokenBean.CredentialsBean[] credentialsBean = new StsTokenBean.CredentialsBean[1];
RequestBody body = RequestBody.create(MediaType.parse("application/json;charset=UTF-8"), "{}");
//从自己服务器获取临时的StsTkoen,使用同步的
RetrofitFactory.Companion
.createInstance(true)
.create(CommonApi.class)
.getStsToken(body)
.subscribe(new Observer<BaseBean<StsTokenBean>>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(BaseBean<StsTokenBean> stsTokenBeanBaseBean) {
if (stsTokenBeanBaseBean != null && stsTokenBeanBaseBean.getContent()!= null && stsTokenBeanBaseBean.getContent().getCredentials() != null){
credentialsBean[0] = stsTokenBeanBaseBean.getContent().getCredentials();
}
}
});
if (credentialsBean[0] != null){
return new OSSFederationToken(credentialsBean[0].getAccessKeyId(),credentialsBean[0].getAccessKeySecret(),credentialsBean[0].getSecurityToken(),credentialsBean[0].getExpiration());
}
return null;
}
};
ClientConfiguration conf = new ClientConfiguration();
conf.setConnectionTimeout(15 * 1000); // 连接超时，默认15秒
conf.setSocketTimeout(15 * 1000); // socket超时，默认15秒
conf.setMaxConcurrentRequest(5); // 最大并发请求书，默认5个
conf.setMaxErrorRetry(2); // 失败后最大重试次数，默认2次
oss = new OSSClient(STFApplication.Companion.getContext(), ENDPOINT, provider, conf);
}
public static Observable<PutObjectResult> upImages(final List<UpImageBean> pathList){
if (oss == null){
Log.e(TAG,"oss == null 没有初始化OSS");
return null;
}
if (pathList == null || pathList.size() == 0){
Log.e(TAG,"pathList == null 图片数组不能为空");
return null;
}
int threadNum = Runtime.getRuntime().availableProcessors();
final ExecutorService executor = Executors.newFixedThreadPool(threadNum);
return Observable.from(pathList)
.flatMap(bean-> Observable.just(bean)
.map(upImageBean->{
PutObjectRequest put;
try {
put = new PutObjectRequest(BUCKE_NAME, upImageBean.getName(), upImageBean.getPath());
return oss.putObject(put);
} catch (ClientException e) {
e.printStackTrace();
} catch (ServiceException e) {
e.printStackTrace();
}
return null;
})
.subscribeOn(Schedulers.from(executor)))
.subscribeOn(Schedulers.io())
.finallyDo(()->executor.shutdown())
.observeOn(AndroidSchedulers.mainThread());
}
}
使用
public void upPhotos(List<UpImageBean> pathList){
final int[] imgCount = {0};
OSSUtils.upImages(pathList)
.subscribe(new Subscriber<PutObjectResult>() {
@Override
public void onCompleted() {
if (imgCount[0] == pathList.size()){//全部上传成功
}
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(PutObjectResult putObjectResult) {
if (putObjectResult!= null && putObjectResult.getStatusCode() == 200){//上传成功
imgCount[0]++;
}
}
});
}
