Rxjava的最简单使用

水深无声 2022-06-13 03:52 194阅读 0赞

最近在项目开发中，需要使用到EventBus的来实现当用户信息发生改变后，进行所有与帐号有关的页面的UI进行更改,想着使用一下Rxjava,第一次看，发现有点复杂，在代码中使用了后发现，这个特别好用，自己特别记录一下：

*  **RxEvent**
 *  **EventObserver**
 *  **RxEventBus**
 *  **AccurateEventObserver**
 *  **在Activity中使用**
 *  **离线写博客**
 *  **导入导出Markdown文件**
 *  **丰富的快捷键**

--------------------

## RxEvent ##

代码特别简单，如下：

public class RxEvent {
        protected String action;
        protected Callback callBack;
    
        protected RxEvent() {
    
        }
    
        public RxEvent(String action) {
            this.action = action;
        }
    
        public void setAction(String action) {
            this.action = action;
        }
    
        public String getAction() {
            return this.action;
        }
    
        public void setCallBack(Callback callBack) {
            this.callBack = callBack;
        }
        protected  interface Callback {
            void onCallback(RxEvent event);
        }
    }

## EventObserver ##

public interface EventObserver {
        void onReceive(RxEvent rxEvent);
    }

## RxEventBus ##

这个类是整个使用的核心，

public class RxEventBus { 
    
        private static final String TAG = "Event.RxEventBus";
        private static volatile RxEventBus sEventBus;
        private Subject<RxEvent, RxEvent> mBus;
        private Map<String, List<ObserverWrapper>> mMaps;
          //使用单例模式
        private RxEventBus() {
            PublishSubject<RxEvent> ps = PublishSubject.create();
            mBus = new SerializedSubject<>(ps);
            mMaps = new ConcurrentHashMap<>();
        }
    
        public static RxEventBus getBus() {
            if (sEventBus == null) {
                synchronized (RxEventBus.class) {
                    if (sEventBus == null) {
                        sEventBus = new RxEventBus();
                    }
                }
            }
            return sEventBus;
        }
    
        /** * Publish a RxxEvent to the RxEventBus,it will be * accurate diapatched to a EventObserver which was register before * 发布事件 * @param event * @see #register * @see #unregister */
        public boolean publish(RxEvent event) {
            if (event == null) {
                Log.e(TAG, "publish event failed,event is null");
                return false;
            }
            mBus.onNext(event);
            //callback to event
            if (event.callBack != null) {
                event.callBack.onCallback(event);
            }
            Log.i(TAG, "publish event(%s) successfully", event);
    
            return true;
        }
    
        private Observable<RxEvent> getObserverableHook() {
            return mBus.observeOn(AndroidSchedulers.mainThread());//指定事件消费者所在线程
        }
    
          //注册
        public boolean register(final String action, final EventObserver observer) {
            if (observer == null) {
                Log.i(TAG, "register event observer failed ,observer is null");
                return false;
            }
            if (action == null || action.length() == 0) {
                Log.i(TAG, "register event observer failed,action is null");
                return false;
            }
            ObserverWrapper ow = new ObserverWrapper(observer);
    
            List<ObserverWrapper> list = mMaps.get(action);
    
            if (list == null) {
                list = new LinkedList<>();
                mMaps.put(action, list);
            } else if (list.contains(ow)) {
                Log.i(TAG, "register event observer failed ,the observer has been already register");
                return false;
            }
    
            ow.subscription = createSubScription(action, observer);
            list.add(ow);
            Log.i(TAG, "register event(%s) observer (%s) successfully", action, observer);
            return true;
    
        }
        //反注册
        public boolean unregister(final String action, final EventObserver observer) {
            if (action == null || action.length() == 0 || observer == null) {
                Log.w(TAG, "unregister event(%s) observer(%s) failed ,action or observer is null ", action, observer);
                return false;
            }
    
            List<ObserverWrapper> list = mMaps.get(action);
            if (list == null) {
                Log.w(TAG, "unregister event observer failed,can not find the observer(%s) list by action(%s). ", observer, action);
                return false;
            }
            ObserverWrapper ow = null;
            for (int i = 0; i < list.size(); i++) {
                if (observer.equals(list.get(i).observer)) {
                    ow = list.remove(i);
                    break;
                }
            }
            if (ow == null) {
                Log.e(TAG, "unregister event observer failed ,can not find the observer(%s) int the list by action(%S) ", observer, action);
                return false;
            }
            if (list.isEmpty()) {
                mMaps.remove(action);
            }
    
            Subscription subscription=ow.subscription;
            if (subscription == null) {
                Log.w(TAG, "unregister event(%s) observer(%s) successfully, but subscription is null.", action, observer);
                return true;
            }
            if (!subscription.isUnsubscribed()) {
                subscription.unsubscribe();
            }
            Log.i(TAG, "unregister event(%s) observer(%s) successfully.", action, observer);
            return true;
        }
    
        private Subscription createSubScription(final String action, final EventObserver observer) {
            Subscription subscription = getObserverableHook().subscribe(new Action1<RxEvent>() {
                @Override
                public void call(RxEvent event) {
                    if (action.equals(event.getAction())) {
                    //注意：这里就是我们要关心的地方，可以看到，当注册后有事件发布，就会回调 
                    //EventObserver的onReceive（）方法
                        observer.onReceive(event);
                    }
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    Log.e(TAG, throwable.getMessage());
                }
            });
            return subscription;
        }
        //
        private static class ObserverWrapper { 
            EventObserver observer;
            Subscription subscription;
    
            ObserverWrapper(EventObserver observer) {
                this.observer = observer;
            }
    
            ObserverWrapper(EventObserver observer, Subscription subscription) {
                this.observer = observer;
                this.subscription = subscription;
            }
    
            @Override
            public boolean equals(Object obj) {
                if (this == obj) {
                    return true;
                }
                if (!(obj instanceof ObserverWrapper)) {
                    return false;
                }
                ObserverWrapper ow = (ObserverWrapper) obj;
                return ow.observer == this.observer || (this.observer != null && this.observer.equals(ow.observer));
            }
        }
    }

### AccurateEventObserver ###

实现EventObserver接口

public abstract class AccurateEventObserver<EventType extends RxEvent> implements EventObserver { 
        private static final String TAG = "AccurateEventObserver";
    
        @Override
        public void onReceive(RxEvent rxEvent) {
            EventType event = null;
    
            try {
    
                event = (EventType) rxEvent;
            } catch (Exception e) {
                Log.v(TAG, "cast RxEvent failed. %s", e);
                return;
            }
            if (event != null) {
                onReceiveEvent(event);
            }
    
        }
    
        public abstract void onReceiveEvent(EventType event);
    }

### AccurateEventObserver ###

public abstract class AccountChangeActivity extends BaseActivity { 
        private EventObserver eventObserver = null;
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            eventObserver = new AccurateEventObserver<AccountUpdataEvent>() {
    
                @Override
                public void onReceiveEvent(AccountUpdataEvent event) {
                    onAccountUpdate(AccountManager.getInstance().getUserInfo());
                }
            };
            RxEventBus.getBus().register(AccountUpdataEvent.ID, eventObserver);
        }
    
    
        @Override
        protected void onDestroy() {
            RxEventBus.getBus().unregister(AccountUpdataEvent.ID, eventObserver);
            eventObserver = null;
            super.onDestroy();
        }
        /** * 当用户信息发生修改时,更新UI * @param info */
        public abstract void onAccountUpdate(UserInfo info);
    
    }

### 在Activity中使用 ###

public abstract class AccountChangeActivity extends BaseActivity { 
        private EventObserver eventObserver = null;
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            eventObserver = new AccurateEventObserver<AccountUpdataEvent>() {
    
                @Override
                public void onReceiveEvent(AccountUpdataEvent event) {
                    onAccountUpdate(AccountManager.getInstance().getUserInfo());
                }
            };
            //注册
            RxEventBus.getBus().register(AccountUpdataEvent.ID, eventObserver);
        }
    
    
        @Override
        protected void onDestroy() {
            //切记要记得反注册
            RxEventBus.getBus().unregister(AccountUpdataEvent.ID, eventObserver);
            eventObserver = null;
            super.onDestroy();
        }
        /** * 当用户信息发生修改时,更新UI，每当我们更新界面时，就会调用该方法 * @param info */
        public abstract void onAccountUpdate(UserInfo info);
    
    
    
    }

### publish(发布) ###

//通知相关的注册页面进行用户信息更新
                        RxEventBus.getBus().publish(new AccountUpdataEvent());

通过上面几个步骤就可以实现最简单的Rxjava使用，为自己使用记录一下。