java - 我的 rxjava 程序中的 Scheduler.Worker 线程引发致命异常

Question

我做了这个项目:

public class MainActivity
        extends ActionBarActivity {

    private Realm realm;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        RealmConfiguration defaultConfig = new RealmConfiguration.Builder().deleteRealmIfMigrationNeeded().build();
        realm = Realm.getInstance(defaultConfig);
        /**
         * Set up Android CardView/RecycleView
         */

        RecyclerView mRecyclerView = (RecyclerView) findViewById(R.id.recycler_view);
        mRecyclerView.setHasFixedSize(true);
        mRecyclerView.setLayoutManager(new LinearLayoutManager(this));
        final CardAdapter mCardAdapter = new CardAdapter();
        mRecyclerView.setAdapter(mCardAdapter);

        /**
         * START: button set up
         */
        final Button bClear = (Button) findViewById(R.id.button_clear);
        Button bFetch = (Button) findViewById(R.id.button_fetch);
        bClear.setOnClickListener(new View.OnClickListener() {
            public void onClick(View v) {
                mCardAdapter.clear();
            }
        });

        bFetch.setOnClickListener(new View.OnClickListener() {
            public void onClick(View v) {
                Retrofit retrofit = new Retrofit.Builder()
                        .addConverterFactory(GsonConverterFactory.create())
                        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                        .baseUrl(GithubService.SERVICE_ENDPOINT).build();

                GithubService service = retrofit.create(GithubService.class);

                service.getAirport()
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(new Subscriber<List<Airport>>() {
                            List<Airport> airps = new ArrayList<Airport>();

                            @Override
                            public void onCompleted() {
                                for(final Airport air : airps) {
                                    realm.executeTransactionAsync(new Realm.Transaction() {
                                        @Override
                                        public void execute(Realm mRealm) {
                                            AirportR airport = mRealm.createObject(AirportR.class);
                                            airport.setId(air.getId());
                                        }
                                    }, new Realm.Transaction.OnSuccess() {
                                        @Override
                                        public void onSuccess() {
                                            Log.wtf("ok", "ok");
                                        }
                                    }, new Realm.Transaction.OnError() {
                                        @Override
                                        public void onError(Throwable error) {
                                            Log.e("ok", "non vaaa");
                                        }
                                    });
                                }
                            }

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

                            @Override
                            public void onNext(List<Airport> airports) {
                                airps = airports;
                            }
                        });
            }
        });
        /**
         * END: button set up
         */
    }

我想在我的离线 Realm 数据库中保存所有机场。当我启动它时，出现此错误:

E/AndroidRuntime: FATAL EXCEPTION: main
              Process: com.example.githubdemo.app, PID: 30527
              Theme: themes:{}
              java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.Worker thread.
                  at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:62)
                  at android.os.Handler.handleCallback(Handler.java:739)
                  at android.os.Handler.dispatchMessage(Handler.java:95)
                  at android.os.Looper.loop(Looper.java:148)
                  at android.app.ActivityThread.main(ActivityThread.java:5461)
                  at java.lang.reflect.Method.invoke(Native Method)
                  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)
                  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616)
               Caused by: rx.exceptions.OnCompletedFailedException: Task java.util.concurrent.FutureTask@d76273a rejected from io.realm.internal.async.RealmThreadPoolExecutor@c1e8feb[Running, pool size = 9, active threads = 9, queued tasks = 100, completed tasks = 0]
                  at rx.observers.SafeSubscriber.onCompleted(SafeSubscriber.java:90)
                  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:262)
                  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:199)
                  at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
                  at android.os.Handler.handleCallback(Handler.java:739) 
                  at android.os.Handler.dispatchMessage(Handler.java:95) 
                  at android.os.Looper.loop(Looper.java:148) 
                  at android.app.ActivityThread.main(ActivityThread.java:5461) 
                  at java.lang.reflect.Method.invoke(Native Method) 
                  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726) 
                  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616) 
               Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@d76273a rejected from io.realm.internal.async.RealmThreadPoolExecutor@c1e8feb[Running, pool size = 9, active threads = 9, queued tasks = 100, completed tasks = 0]
                  at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2014)
                  at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:794)
                  at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1340)
                  at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:82)
                  at io.realm.internal.async.RealmThreadPoolExecutor.submitTransaction(RealmThreadPoolExecutor.java:71)
                  at io.realm.Realm.executeTransactionAsync(Realm.java:1340)
                  at app.activity.MainActivity$2$1.onCompleted(MainActivity.java:88)
                  at rx.observers.SafeSubscriber.onCompleted(SafeSubscriber.java:84)
                  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:262) 
                  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:199) 
                  at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) 
                  at android.os.Handler.handleCallback(Handler.java:739) 
                  at android.os.Handler.dispatchMessage(Handler.java:95) 
                  at android.os.Looper.loop(Looper.java:148) 
                  at android.app.ActivityThread.main(ActivityThread.java:5461) 
                  at java.lang.reflect.Method.invoke(Native Method) 
                  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726) 
                  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616) 

我该如何解决这个问题？

还有其他方法可以使用 rxJava 来保存 Realm 中的机场吗？

谢谢

Answer 1

不要为从 API 收到的每个 Airport 实例排队异步事务 - for batch inserts, you should consider using 1 asynchronous transaction instead, and the insert API .

所以这个:

for(final Airport air : airps) {
    realm.executeTransactionAsync(new Realm.Transaction() {
        @Override
        public void execute(Realm mRealm) {
            AirportR airport = mRealm.createObject(AirportR.class);
            airport.setId(air.getId());
        }
    }, new Realm.Transaction.OnSuccess() {
        @Override
        public void onSuccess() {
            Log.wtf("ok", "ok");
        }
    }, new Realm.Transaction.OnError() {
        @Override
        public void onError(Throwable error) {
            Log.e("ok", "non vaaa");
        }
    });
}

应该是这样的:

realm.executeTransactionAsync(new Realm.Transaction() {
    @Override
    public void execute(Realm realm) {
        AirportR airport = new AirportR();
        for(final Airport air : airps) {
            airport.setId(air.getId());
            realm.insertOrUpdate(airport);
        }
    }, new Realm.Transaction.OnSuccess() {
        @Override
        public void onSuccess() {
            Log.wtf("ok", "ok");
        }
    }, new Realm.Transaction.OnError() {
        @Override
        public void onError(Throwable error) {
            Log.e("ok", "non vaaa");
        }
    });
}