Rxjava 3 + Retrofit2 - 多次插入数据库问题

Question

我正在尝试执行以下操作；使用 Retrofit 将云数据库同步到设备上的本地 SqLite 数据库（房间）。数据库可能会变大，大约 100,000 个寄存器或更多，因此同步过程可能需要一些时间。所以它发送第一个Retrofit请求来获取寄存器的数量，这样它就可以计算出总页数，然后它会发送多个Retrofit请求，从API中获取所有数据，每次请求后，它将数据保存到房间。

现在，我在组合两个 RxJava 调用或进程时遇到问题，同样在第二个 RxJava 进程上，在 Retrofit 调用之后，有一个对象列表的房间插入，但是在 hole 进程结束之后，我注意到并不是所有记录的 100% 都被插入，每次我运行过程中，插入的记录数都会发生变化，大约是 80% - 98%，但从来没有 100%，即使所有Retrofit 调用已发送。

请帮助我：

如何在一次 RxJava 调用中完成所有进程，而不是像我那样调用 2 个现在？
如何将100%的记录插入房间？

代码如下：

Gradle

def room_version = "2.2.5"
//RxJava 2
implementation "io.reactivex.rxjava2:rxjava:2.2.19"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
//Retrofit
implementation 'com.squareup.retrofit2:retrofit:2.8.1'
implementation 'com.squareup.retrofit2:converter-gson:2.8.1'
//Retrofit2 Adapter for RxJava 2
implementation "com.squareup.retrofit2:adapter-rxjava2:2.8.1"
//okhttp3 Logging Interceptor
implementation "com.squareup.okhttp3:logging-interceptor:4.5.0"
//Room
implementation "androidx.room:room-runtime:$room_version"
annotationProcessor "androidx.room:room-compiler:$room_version"
//RxJava support for Room
implementation "androidx.room:room-rxjava2:$room_version"

ItemSyncDetails

...
public class ItemSyncDetails {
    @SerializedName("CurrentPage")
    int currentPage;
    @SerializedName("PageCount")
    int pageCount;
    @SerializedName("PageSize")
    int pageSize;
    @SerializedName("RecordCount")
    int recordCount;
    @SerializedName("Results")
    List<Item> mItemList;
...
}

ItemDao

Note: I haven't used Observer/Flowable/Maybe/Single, because I having been able to make it work with RxJava

import io.reactivex.Flowable;

@Dao
public interface ItemDao {

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    long insert(Item item);

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    List<Long> insertAll(List<Item> items);
...

DataApi

import io.reactivex.rxjava3.core.Observable;
...

public interface DataApi {

    @GET("item")
    Observable<ItemSyncDetails> getItemsByPage(
            @Query("pageSize") Integer pageSize,
            @Query("currentPage") Integer currentPage,
            @Query("sortBy") Integer sortBy
    );

项目资源库

import io.reactivex.Observable;
    ...

    public class ItemRepository {
    ...

        public ItemRepository(Application application) {
            mDataApi = RetrofitClient.getRetrofitInstance("http://192.168.1.100").create(DataApi.class);
            RfidDatabase db = RfidDatabase.getAppDatabase(application);
            itemDao = db.itemDao();
            itemList = itemDao.getAllItems();
            inserts = 0;
        }

        public List<Long> insertAllLocal (List<Item> itemList) {
            List<Long> items = itemDao.insertAll(itemList);
            inserts += items.size();
            Log.i(TAG, "************insertAllLocal - ItemRepository: " + inserts + "*************");
            Log.i(TAG, "************insertAllLocal - ItemRepository: " + items);
            return items;
        }

        public Observable<ItemSyncDetails> getRecordsCount(){
            return mDataApi.getItemsByPage(1,1,1);
        }

        public Observable<ItemSyncDetails> getItemsPerPage(int pageSize,int currentPage){
            return mDataApi.getItemsByPage(pageSize,currentPage,1);
        }
    ...

SyncConfigFragment 

    import io.reactivex.Observable;
    import io.reactivex.android.schedulers.AndroidSchedulers;
    import io.reactivex.disposables.CompositeDisposable;
    import io.reactivex.functions.Function;
    import io.reactivex.schedulers.Schedule
    ...

    public class SyncConfigFragment extends Fragment {


        private ItemViewModel itemViewModel;
        private ImageView imageSyncItems;
        private ProgressDialog progressDialog;
        private TextView tvSyncDescriptionItems;
        private DataApi service;
        private ItemSyncDetails mItemSyncDetails;
        private List<Item> mItemlist;
        private CompositeDisposable mCompositeDisposable;
        private int mNumPages;
        private int syncProgress;
        ...

        @Override
        public View onCreateView(LayoutInflater inflater, ViewGroup container, Bundle savedInstanceState) {
            View view =  inflater.inflate(R.layout.fragment_config_sync,container, false);
            progressDialog = new ProgressDialog(getActivity());
            sharedPref = getActivity().getSharedPreferences(
                    getString(R.string.sharepref_filename), Context.MODE_PRIVATE);
            mItemlist = new ArrayList<Item>();
            mCompositeDisposable = new CompositeDisposable();
            itemViewModel = ViewModelProviders.of(this).get(ItemViewModel.class);
            tvSyncDescriptionItems = view.findViewById(R.id.tvDescriptionSyncItems);
            if(sharedPref.contains("last_sync_item")) {
                tvSyncDescriptionItems.setText("Última actualización " + sharedPref.getString("last_sync_item",""));
            } else{
                tvSyncDescriptionItems.setText("No se ha Sincronizado");
            }
            imageSyncItems = view.findViewById(R.id.imageViewSyncItems);
            imageSyncItems.setOnClickListener(clickListener);
            return view;
        }

        private View.OnClickListener clickListener = new View.OnClickListener() {
            public void onClick(View v) {
                    if (v.equals(imageSyncItems)) {
                //If I uncomment the next line it does not work
                        //mCompositeDisposable.add(
                        mNumPages = 0;
                        syncProgress = 0;
                        showProgressDialog("Items");
                        getRecordsCount();
                       //); Closing round bracket for mCompositeDisposable
                }
            }
        };//End View.OnClickListener 

        private void getRecordsCount(){
            itemViewModel.getRecordsCount()
                    .subscribeOn(Schedulers.io())
                    .retry(3)
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(this::HandleResults, this::handleError,this::getNumPagesHandlerComplete );
        }

        private void HandleResults(ItemSyncDetails itemSyncDetails) {
            this.mItemSyncDetails = itemSyncDetails;
            int pageSize = 100;
            int numPages = itemSyncDetails.getRecordCount()/pageSize;
            if (itemSyncDetails.getRecordCount() < pageSize || itemSyncDetails.getRecordCount()%pageSize != 0){
                numPages++;
            }
            this.mNumPages = numPages;
        }

        private void getNumPagesHandlerComplete() {
            getAllRecords(mNumPages);
        }

        private void handleError(Throwable throwable) {
            tvSyncDescriptionItems.setText("**********Error de conexión...");
            closeProgressDialog();
        }

        private void getAllRecords(int numPages){
            //numPages: total of pages are the number of times to send the request to API
            Observable.range(1, numPages)
                    .flatMap(i -> itemViewModel.getItemsPerPage(100,i))
                    .map(new Function<ItemSyncDetails, Integer>() {
                        @Override
                        public Integer apply(ItemSyncDetails itemSyncDetails) throws Throwable {
                            return itemViewModel.insertAllLocal(itemSyncDetails.getItemList()).size();
                        }
                    })
                    .subscribeOn(Schedulers.newThread())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(this::getAllHandleResults, this::handleError,this::handleComplete);
        }

        private void getAllHandleResults(Integer i) {
            progressDialog.setProgress(getProgress(i));
        }

        private void handleComplete() {
            //last request finished
            closeProgressDialog();
        }

        private int getProgress(int newItems){
            syncProgress += newItems;
            int progress = 0;
            if (syncProgress == mItemSyncDetails.getRecordCount()){
                progress = 100;
            } else {
                progress = (100 * syncProgress)/mItemSyncDetails.getRecordCount();
            }
            return progress;
        }
    ...
    }

http://192.168.1.10:82/api/v1.0/item?pageSize=1&currentPage=1&sortBy=1

Note: The page size could change, I am using a fixed size of a 100 items per page.

{
  Results: [
  {
    epc: "202020202020202030303031",
    barcode: "0001",
    name: "Televisor Samnsung",
    description: "0001",
    creation_date: "2020-02-26T10:55:06",
    last_update: "2020-02-26T10:55:06",
    last_seen: "2020-02-26T10:55:06",
    brand: "Samnsung",
    serial_number: "0001",
    parent: "",
    fk_category: 1,
    responsable: "",
    purchase_date: "2020-02-26T10:55:06",
    cost: 0,
    fk_location: 1008,
    fk_item_state: 1,
    inventory_date: "2020-02-26T10:55:06"
  }
 ],
 CurrentPage: 1,
 PageCount: 65565,
 PageSize: 1,
 RecordCount: 65565
}

Answer 1

您在编辑之前在此处发布了 json 回复。

    CurrentPage: 1,
    PageCount: 65566,
    PageSize: 1,
    RecordCount: 65566

如果我没理解错的话，那么你有 65k 个项目，每页有 1 个项目。意思是 65k 页面，这意味着 65k 网络调用。好多啊。你可以先改进这个设计。

将整个记录分成几页（甚至可能是 10 或 20 页）。如果整个记录有 10 万个项目，1 页仍然有数千个项目。
然后使用 gzip 压缩压缩每个页面的 json 响应并从服务器提供。或者不要将记录分成几页并将它们全部传递到一个用 gzip 压缩的响应中（如果它不是那么大的话）。
解压缩 android 上的响应，解析它然后做任何你想做的事。

这样可以减少大量网络调用，并可能减少同步的等待时间。

关于您实际的 rx 问题：

val pageSize = 100
viewModel.getRecordsCount()
    .map {
        // logic from `HandleResults` function
        // do some calculation
        var numPages: Int = it.records / pageSize
        if (it.records < pageSize || it.records % pageSize != 0) {
            numPages++
        }
        return@map numPages
    }
    .flatMap { pages -> Observable.range(1, pages) }
    .flatMap { page -> viewModel.getItemsPerPage(pageSize, page) }
    .flatMap { itemSyncDetails ->
        val items = viewModel.insertAllLocal(itemSyncDetails.getItemList())
        return@flatMap Observable.just(items.size)
    }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(....)

I notice that not 100% of all the records are inserted, every time that I run the process, the number of records inserted change, it is around 80% - 98%, but never 100%, even though all the Retrofit calls are sent.

记录 handleError 函数中的错误，看看实际问题是什么。

Rxjava 3 + Retrofit2 - 多次插入数据库问题

Rxjava 3 + Retrofit2 - multiple inserts to DB problem

android

retrofit2

android-room

rx-java3