RxJava作为事件总线，只有一个事件时onNext被多次调用post

Question

我正在使用 RxJava 实现事件总线 (RxBus)。

RxBus.java

public class RxBus {

    private static final String TAG = LogUtils.makeTag(RxBus.class);
    private static final RxBus INSTANCE = new RxBus();

    private final Subject<Object, Object> mBusSubject = new SerializedSubject<>(PublishSubject.create());

    public static RxBus getInstance() {
        return INSTANCE;
    }

    public <T> Subscription register(final Class<T> eventClass, Action1<T> onNext) {
        return mBusSubject
                .filter(new Func1<Object, Boolean>() {
                    @Override
                    public Boolean call(Object event) {
                        return event.getClass().equals(eventClass);
                    }
                })
//                .filter(event -> event.getClass().equals(eventClass))
                .map(new Func1<Object, T>() {
                    @Override
                    public T call(Object obj) {
                        return (T) obj;
                    }
                })
//                .map(obj -> (T) obj)
                .subscribe(onNext);
    }

    public void post(Object event) {
        Log.d(TAG, "Apr12, " + "post event: " + event);
        mBusSubject.onNext(event);
    }
}

Post 来自 RecyclerView 的 viewHolder 的事件

public ViewHolder(LayoutInflater inflater, final ViewGroup parent) {
        super(inflater.inflate(R.layout.bill_card, parent, false));

        drawee = (SimpleDraweeView) itemView.findViewById(R.id.card_image);
        title = (TextView) itemView.findViewById(R.id.card_title);

        itemView.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {

                Log.d(TAG, "Apr12, item clicked.");
                RxBus.getInstance().post(new ItemSelectedEvent(position));
            }
        });

        TagImageButton = (ImageButton) itemView.findViewById(R.id.tag_button);
        TagImageButton.setOnClickListener(new View.OnClickListener(){
            @Override
            public void onClick(View v) {
                Log.d(TAG, "Tag button clicked.");
                RxBus.getInstance().post(new ApplyTagForItemEvent(position));
            }
        });
    }
}

订阅Fragment的活动

@Override
public void onActivityCreated(@Nullable Bundle savedInstanceState) {
    super.onActivityCreated(savedInstanceState);

    mActivity = getActivity();
    Log.d(TAG, "getActivity(): " + getActivity());
    mItemClickSubscription = RxBus.getInstance().register(ItemSelectedEvent.class,
            new Action1<ItemSelectedEvent>() {
                @Override
                public void call(ItemSelectedEvent event) {
                    Log.d(TAG, "Apr12, " + "call event: " + event);
                    if (mDetail == null) {
                        if (getParentFragment() instanceof IFragmentStackHolder) {
                            IFragmentStackHolder fsh = (IFragmentStackHolder) getParentFragment();

                            Fragment details = new DetailCardFragment();
                            Bundle args = new Bundle();
                            args.putInt(ContentHolder.INDEX, event.getPosition());
                            details.setArguments(args);

                            fsh.pushFragment(details, event.getPairs());
                        }
                    }
                }
            });

    mApplyTagSubscription = RxBus.getInstance().register(ApplyTagForItemEvent.class,
            new Action1<ApplyTagForItemEvent>() {
                @Override
                public void call(ApplyTagForItemEvent event) {
                    IFragmentStackHolder fsh = (IFragmentStackHolder) getParentFragment();

                    Fragment tagApplyFragment = new TagApplyFragment();
                    Bundle args = new Bundle();
                    args.putInt(ContentHolder.INDEX, event.getPosition());
                    tagApplyFragment.setArguments(args);

                    fsh.pushFragment(tagApplyFragment, null);
                }
            }
    );
}

问题是：当我点击itemView或TagImageButton时，RxBus.post()只调用了一次（这是正确的），但是Action1 call()被调用了多次（甚至不是固定时间）。请查看下面的日志。

D/**-CardContentView(31177): Apr12, item clicked.
D/**-RxBus(31177): Apr12, post event: com.*****.events.ItemSelectedEvent@1a11346e
D/**-CardDetailFragment(31177): Apr12, call event: com.*****.events.ItemSelectedEvent@1a11346e
D/**-CardDetailFragment(31177): Apr12, call event: com.*****.events.ItemSelectedEvent@1a11346e
D/**-CardDetailFragment(31177): Apr12, call event: com.*****.events.ItemSelectedEvent@1a11346e

如何让它只调用一次？

EDIT: 我发现如果这次Action1 call()被调用了N次，下次点击item时就会被调用N+1次。似乎 observable 正在向订阅者发送历史记录中所有随后观察到的项目。

Answer 1

终于找到解决办法了。

很简单：我应该在 onStop() 中调用 mItemClickSubscription.unsubscribe(); 和 mApplyTagSubscription.unsubscribe();。

PublishSubject用于事件总线。 PublishSubject 是一个主题：

Subject that, once an Observer has subscribed, emits all subsequently observed items to the subscriber.

因此，如果您不 unsubscribe() 订阅，此订阅将保留 "recording" 历史记录中发生的所有事件，并在 .subscribe(onNext) 执行后发出所有事件。

RxJava作为事件总线，只有一个事件时onNext被多次调用post

RxJava as an event bus, onNext is called multiple times when only one event post

android

event-bus

rx-java

rx-android