Android Rxbus事件总线

Posted 2023-02-26 ywl5320

我的视频课程：《FFmpeg打造Android万能音频播放器》

        最近在项目中使用了Rxjava和Rxandroid，确实感觉挺不错的，然后听说可以用RxBus来替换掉EventBus等事件分发的三方库，然后就上午找比较好的RxBus开源项目（以为是有统一标准的），然而后面才知道，RxBus没有绝对的，只是一种想法而已，借助Rxjava的特性来实现的。既然这样，我也就自己实现了一个（可以设置tag，并指定事件接收的线程类型），测试效果还是蛮不错的。先看效果图：

途中，在第一个activity中发送事件，并在其中接收到事件，文本框内容改变；打开第二个activity，发送相同tag的事件，所以第一个activity中也接收到了第二个activity发送的事件。

一：首先添加依赖库

dependencies 
    compile 'io.reactivex:rxjava:1.1.0'
    compile 'io.reactivex:rxandroid:1.1.0'

二：编写RxBus类

package com.ywl5320.rxbusdemo.RxBus;


import java.util.HashMap;
import java.util.Map;

import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;

/**
 * Created by ywl on 2016/5/20.
 */
public class RxBus 

    private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());
    private final Map<String, Object> tags = new HashMap<>();

    private static RxBus rxbus;

    public static RxBus getInstance()
    
        if(rxbus == null)
        
            synchronized (RxBus.class) 
                if(rxbus == null) 
                    rxbus = new RxBus();
                
            
        
        return rxbus;
    

    /**
     * 发送事件消息
     * @param tag 用于区分事件
     * @param object 事件的参数
     */
    public void post(String tag, Object object)
    
        _bus.onNext(object);
        if(!tags.containsKey(tag))
        
            tags.put(tag, object);
        
    

    /**
     * 主线程中执行
     * @param tag
     * @param rxBusResult
     */
    public void toObserverableOnMainThread(final String tag, final RxBusResult rxBusResult) 

            _bus.observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Object>() 
                @Override
                public void call(Object o) 
                    if (tags.containsKey(tag)) 
                        rxBusResult.onRxBusResult(o);
                    
                
            );
    

    /**
     * 子线程中执行
     * @param tag
     * @param rxBusResult
     */
    public void toObserverableChildThread(final String tag, final RxBusResult rxBusResult) 

        _bus.observeOn(Schedulers.io()).subscribe(new Action1<Object>() 
            @Override
            public void call(Object o) 
                if (tags.containsKey(tag)) 
                    rxBusResult.onRxBusResult(o);
                
            
        );
    

    /**
     * 移除tag
     * @param tag
     */
    public void removeObserverable(String tag)
    
            if(tags.containsKey(tag))
            
                tags.remove(tag);
            
    

    /**
     * 退出应用时，清空资源
     */
    public void release()
    
        tags.clear();
        rxbus = null;
    

在RxBus类中，使用到了Subject引用网上的解释：“Subject在ReactiveX是作为observer和observerable的一个bridge或者proxy。因为它是一个观察者，所以它可以订阅一个或多个可观察对象，同时因为他是一个可观测对象，所以它可以传递和释放它观测到的数据对象，并且能释放新的对象。”

Subject表示一个同时是Observable和Observer的对象，所以Subject即使订阅者也是观察者，利用这个特性，我们就可以利用它发送和接收消息。Subject的onNext(Object)方法会向所有订阅者发送事件，但是我们在项目中一般都是点对点的发送，不希望其他无关的订阅者也接收到事件，所以在RxBus中添加了一个Map来管理事件的tag，来区别该事件来自哪里，将要去哪里。

为了接收时写更少的代码，写了一个回调函数来处理接收到的信息：

package com.ywl5320.rxbusdemo.RxBus;

/**
 * Created by ywl on 2016/5/20.
 */
public interface RxBusResult 

    /**
     * 事件回调接口
     * @param o
     */
    void onRxBusResult(Object o);

二：怎么使用

1、首先定义RxBus并初始化

private RxBus rxBus = RxBus.getInstance();

2、发送消息

btnsend.setOnClickListener(new View.OnClickListener() 
            @Override
            public void onClick(View v) 
                rxBus.post("first", new String("hello rxbus"));
            
        );

其中，“first”即为区分事件的tag，第二个参数是发送的对象。

3、接收消息

rxBus.toObserverableOnMainThread("first", new RxBusResult() 
            @Override
            public void onRxBusResult(Object o) 
                final String msg = (String)o;
                mtvmsg.setText("first收到消息;" + msg);
                Toast.makeText(MainActivity.this, "收到消息;" + msg, Toast.LENGTH_SHORT).show();
            
        );

这里指定在主线程中接收tag为“first”的消息并处理。

这样发送接收就都完成了，但是为了更好的管理tag，我们在退出时最好把相应的tag清除掉，如：

@Override
    protected void onDestroy() 
        super.onDestroy();
        rxBus.removeObserverable("first");
    

最后：在应用程序完全退出时还需调用rxBus.release()方法释放资源（也避免了一个问题：就是如果不释放，第二次启动应用时，不能更新UI，可能是rxBus和第二次启动的应用不在同一个进程中造成的，因为第一次退出应用时，rxBus是没有销毁的，第二次不会重新创建，所以导致了这个问题）

最后把MainActivity和SecondActivity中代码贴出来:

package com.ywl5320.rxbusdemo;

import android.content.Intent;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;
import android.widget.Toast;

import com.ywl5320.rxbusdemo.RxBus.RxBus;
import com.ywl5320.rxbusdemo.RxBus.RxBusResult;

public class MainActivity extends AppCompatActivity 

    private Button btnsend;
    private Button btnsecond;
    private TextView mtvmsg;

    private RxBus rxBus = RxBus.getInstance();

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

        btnsend = (Button) findViewById(R.id.btn_send);
        btnsecond = (Button) findViewById(R.id.btn_second);
        mtvmsg = (TextView) findViewById(R.id.tv_msg);

        btnsend.setOnClickListener(new View.OnClickListener() 
            @Override
            public void onClick(View v) 
                rxBus.post("first", new String("hello rxbus"));
            
        );

        btnsecond.setOnClickListener(new View.OnClickListener() 
            @Override
            public void onClick(View v) 
                Intent intent = new Intent(MainActivity.this, SecondActivity.class);
                MainActivity.this.startActivity(intent);
            
        );

        rxBus.toObserverableOnMainThread("first", new RxBusResult() 
            @Override
            public void onRxBusResult(Object o) 
                final String msg = (String)o;
                mtvmsg.setText("first收到消息;" + msg);
                Toast.makeText(MainActivity.this, "收到消息;" + msg, Toast.LENGTH_SHORT).show();
            
        );

        rxBus.toObserverableOnMainThread("second", new RxBusResult() 
            @Override
            public void onRxBusResult(Object o) 
                String msg = (String)o;
                mtvmsg.setText("second收到消息;" + msg);
                Toast.makeText(MainActivity.this, "second收到消息;" + msg, Toast.LENGTH_SHORT).show();
            
        );

    

    /**
     * 退出时，释放rxbus
     */
    @Override
    protected void onDestroy() 
        super.onDestroy();
        rxBus.release();
    

package com.ywl5320.rxbusdemo;

import android.os.Bundle;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.view.View;
import android.widget.Button;

import com.ywl5320.rxbusdemo.RxBus.RxBus;

/**
 * Created by ywl on 2016/5/20.
 */
public class SecondActivity extends AppCompatActivity 

    private Button mbtnsend;
    private RxBus rxBus = RxBus.getInstance();

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) 
        super.onCreate(savedInstanceState);
        setContentView(R.layout.second_activity);

        mbtnsend = (Button) findViewById(R.id.btn_send);
        mbtnsend.setOnClickListener(new View.OnClickListener() 
            @Override
            public void onClick(View v) 
                rxBus.post("second", new String("第二个页面的数据"));
            
        );

    

    @Override
    protected void onDestroy() 
        super.onDestroy();
        rxBus.removeObserverable("first");
    

这样就完成了RxBus工具类的编写  Demo下载地址 (注：demo的RxBus有Bug，就是不能同时两个接收同一个发送的消息，这里可以以博客中为准，博客中已经修改了，可以对比下改了什么哟 哈哈哈)。

github：https://github.com/wanliyang1990/rxbusDemo