Rxjava拆分获取联系人

Observable.create(new ObservableOnSubscribe<Contact>() {
    @Override
    public void subscribe(ObservableEmitter<Contact> emitter) throws Exception {
        ContentResolver contentResolver = getContentResolver();
        Cursor cursor = contentResolver.query(
                ContactsContract.Contacts.CONTENT_URI,
                null,
                null,
                null,
                null
        );

        if (cursor != null && cursor.moveToFirst()) {
            do {
                String name = cursor.getString(cursor.getColumnIndex(
                        ContactsContract.Contacts.DISPLAY_NAME
                ));
                Contact contact = new Contact(name);
                emitter.onNext(contact);
            } while (cursor.moveToNext());
            cursor.close();
        }

        emitter.onComplete();
    }
})
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<Contact>() {
        @Override
        public void onSubscribe(Disposable d) {
            // 可选的订阅时回调
        }

        @Override
        public void onNext(Contact contact) {
            // 处理每个联系人
        }

        @Override
        public void onError(Throwable e) {
            // 处理错误
        }

        @Override
        public void onComplete() {
            // 处理完成
        }
    });