다중 스레드 다운로드를 위한 RxJava 및 retrofit

RxJava가 스레드 전환할 때 매우 강하다는 것을 감개무량하게 여겼는데 최근 프로젝트에서 다운로드 기능을 사용했기 때문에 Retrofit와 결합하여 이 방면의 시도를 하려고 합니다.
장면은 매우 간단하다. 서버는 3개의 파일 다운로드 주소를 제공한다.클라이언트가 단추를 눌러서 이 파일들을 비동기적으로 다운로드합니다.이것은 다중 스레드 다운로드와 관련이 있다.
Retrofit 섹션:
BaseApi:

/**
 * songwenju on 16-8-5 : 16 : 09.
 *  ：[email protected]
 */
public class BaseApi {
    // 
    public static OkHttpClient mOkHttpClient = new OkHttpClient.Builder()
            .readTimeout(AppConstant.READ_TIMEOUT, TimeUnit.SECONDS)// 
            .writeTimeout(AppConstant.WRITE_TIMEOUT,TimeUnit.SECONDS)// 
            .connectTimeout(AppConstant.CONNECT_TIMEOUT,TimeUnit.SECONDS)// 
            .build();
}

DownloadApi: 여기는 하나의 단례입니다. 게으름뱅이와 굶주린 식의 특성을 결합한 것입니다. 이것은 라인이 안전하고 실례가 하나입니다.

/**
 * songwenju on 16-8-23 : 16 : 32.
 *  ：[email protected]
 */
public class DownloadApi extends BaseApi{
    // DownloadApi
    private static DownloadApi mDownloadApi = new DownloadApi();
    private DownloadService mDownloadService;

    public static DownloadApi getInstance(){
        return mDownloadApi;
    }

    private DownloadApi(){
        Retrofit retrofit = new  Retrofit.Builder()
                .client(mOkHttpClient)
                .baseUrl(AppConstant.BASE_URL)
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .build();
        mDownloadService = retrofit.create(DownloadService.class);
    }

    public DownloadService getDownloadService() {
        return mDownloadService;
    }
}

Download Service: Streaming 메모를 사용합니다. 이 메모는 큰 파일을 다운로드하는 데 사용됩니다.메모가 추가되면 다운로드 파일은 모든 다운로드 내용을 메모리에 불러오지 않습니다.

/**
 * songwenju on 16-8-23 : 16 : 32.
 *  ：[email protected]
 */
public interface DownloadService {
   @Streaming
   @GET
   Observable downloadFile(@Url String fileUrl);
}

WriteFile Manager: ResponseBody를 받은 내용을 파일에 기록합니다.

/**
 *  
 */
public class WriteFileManager {
    private static final String TAG = "DownLoadManager";

    public static boolean writeResponseBodyToDisk(ResponseBody body,String downloadName) {
        String path = FileUtil.getVideoPath() + downloadName;
        LogUtil.i(TAG, "WriteFileManager.startToWrite.path:" + path);

        File futureFile = new File(path);
        InputStream inputStream = null;
        OutputStream outputStream = null;
        long fileSize = body.contentLength();
        LogUtil.d(TAG,"WriteFileManager.writeResponseBodyToDisk.fileSize:"+fileSize);

        try {
            try {
                byte[] fileReader = new byte[1024 * 1024];
                long fileSizeDownloaded = 0;
                inputStream = body.byteStream();
                outputStream = new FileOutputStream(futureFile);

                while (true) {
                    int read = inputStream.read(fileReader);

                    if (read == -1) {
                        break;
                    }
                    outputStream.write(fileReader, 0, read);
                    fileSizeDownloaded += read;
//                    LogUtil.i(TAG, "file download: " + fileSizeDownloaded + " of " + fileSize);
                }
                LogUtil.d(TAG, "file download: " + fileSizeDownloaded + " of " + fileSize);
                outputStream.flush();
                return true;
            } catch (IOException e) {
                e.printStackTrace();
                return false;
            } finally {
                if (inputStream != null) {
                    inputStream.close();
                }
                if (outputStream != null) {
                    outputStream.close();
                }
            }
        } catch (IOException e) {
            return false;
        }

    }
}

activity에서:merge를 사용하여observable를 통합합니다.

public class MainActivity extends AppCompatActivity {

    private DownloadService mDownloadService;
    private List mDownloadList;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        DownloadApi downloadApi = DownloadApi.getInstance();
        mDownloadService = downloadApi.getDownloadService();
        final String downloadUrl1 = AppConstant.DOWNLOAD_URL1;
        final String downloadUrl2 = AppConstant.DOWNLOAD_URL2;
        final String downloadUrl3 = AppConstant.DOWNLOAD_URL3;

        mDownloadList = new ArrayList<>();
        mDownloadList.add(downloadUrl1);
        mDownloadList.add(downloadUrl2);
        mDownloadList.add(downloadUrl3);
    }

    /**
     *  
     * @param view view
     */
    public void download(View view) {
        LogUtil.i(this, "MainActivity.download.");

        List> observables = new ArrayList>();
        // Observable List 
        for (int i = 0; i < mDownloadList.size(); i++) {
            final String downloadUrl = mDownloadList.get(i);
            observables.add(mDownloadService.downloadFile(downloadUrl)
                    .subscribeOn(Schedulers.io())
                    .map(new Func1() {
                        @Override
                        public Boolean call(ResponseBody responseBody) {
                            return WriteFileManager.writeResponseBodyToDisk(responseBody, downloadUrl);
                        }
                    }).subscribeOn(Schedulers.io()));
        }

        //Observable merge Observable Observable, observable 。
        Observable.merge(observables).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {
                        e.printStackTrace();
                    }

                    @Override
                    public void onNext(Boolean b) {
                        if (b) {
                            Toast.makeText(MainActivity.this, "Download is sucess", Toast.LENGTH_LONG).show();
                        }
                    }
                });
    }

}

log:

01-09 05:24:04.474 28625-28625/? I/swj_htv_MainActivity: MainActivity.download.
01-09 05:24:04.574 28625-28786/? I/swj_htv_DownLoadManager: DownloadManager.startToWrite.path:/storage/emulated/0/Movies/mvp.wmv
01-09 05:24:04.575 28625-28786/? D/swj_htv_DownLoadManager: DownloadManager.writeResponseBodyToDisk.fileSize:400937223
01-09 05:24:04.575 28625-28785/? I/swj_htv_DownLoadManager: DownloadManager.startToWrite.path:/storage/emulated/0/Movies/chess.mp4
01-09 05:24:04.575 28625-28785/? D/swj_htv_DownLoadManager: DownloadManager.writeResponseBodyToDisk.fileSize:6366633
01-09 05:24:04.575 28625-28784/? I/swj_htv_DownLoadManager: DownloadManager.startToWrite.path:/storage/emulated/0/Movies/protein.mp4
01-09 05:24:04.575 28625-28784/? D/swj_htv_DownLoadManager: DownloadManager.writeResponseBodyToDisk.fileSize:374081533
01-09 05:24:06.039 28625-28785/? D/swj_htv_DownLoadManager: file download: 6366633 of 6366633
01-09 05:25:12.286 28625-28784/? D/swj_htv_DownLoadManager: file download: 374081533 of 374081533
01-09 05:25:14.302 28625-28786/? D/swj_htv_DownLoadManager: file download: 400937223 of 400937223

프로젝트에 대응하는 데모가 github에 업로드되었습니다.
RxJavaRetrofitDownload