我的视频课程：《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 _bus = new SerializedSubject<>(PublishSubject.create());
private final Map 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() {
@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() {
@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
京公网安备 11010802041100号