Broncho
0%

EventBus

发表于 更新于 分类于 ，
EventBus使用简介

EventBus

EventBus is a publish/subscribe event bus optimized for Android.

EventBus介绍

  • simplifies the communication between components
    • decouples event senders and receivers
    • performs well with Activities, Fragments, and background threads
    • avoids complex and error-prone dependencies and life cycle issues
  • makes your code simpler
  • is fast
  • is tiny (~50k jar)
  • is proven in practice by apps with 100,000,000+ installs
  • has advanced features like delivery threads, subscriber priorities, etc.

添加EventBus支持

Gradle:

1
2
gradle
compile 'org.greenrobot:eventbus:3.0.0'

download EventBus from Maven Central

EventBus 使用示例

定义事件

1
2
  public class LoadEvent {
}

onCreate中注册事件

1
EventBus.getDefault().register(this);

onDestroy中取消注册

1
EventBus.getDefault().unregister(this);

订阅事件

1
2
3
4
@Subscribe(threadMode = ThreadMode.MAIN, priority = 0, sticky = false)
public void onLoadEvent(LoadEvent event) {
    Log.d(TAG, "onLoadEvent");
}
  • threadMode
    • POSTING 在调用post所在的线程执行回调，直接运行。
    • MAIN 在主线程回调，如果post所在线程为主线程则直接执行，否则则通过mainThreadPoster来调度。
    • BACKGROUND 如果post所在线程为非UI线程则直接执行，否则则通过backgroundPoster来调度，这里只适合执行时间比较短的任务。
    • ASYNC（交给线程池来管理）：直接通过asyncPoster调度。
  • sticky 是否监听黏性事件

    If true, delivers the most recent sticky event (posted with EventBus#postSticky(Object)) to this subscriber (if event available).

发布事件

1
EventBus.getDefault().post(new LoadEvent());

完整示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class MainActivity extends AppCompatActivity {
    public static final String TAG = "MainActivity";

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

        EventBus.getDefault().register(this);

        Button start = (Button) findViewById(R.id.start_load);
        start.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                EventBus.getDefault().post(new LoadEvent());
            }
        });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        EventBus.getDefault().unregister(this);
    }

    @Subscribe(threadMode = ThreadMode.MAIN, priority = 0, sticky = false)
    public void onLoadEvent(LoadEvent event) {
        Log.d(TAG, "onLoadEvent");
    }

    public static class LoadEvent {

    }
}

EventBus的注解生成索引

项目的根目录build.gradle引入apt编译插件

1
classpath 'com.neenbedankt.gradle.plugins:android-apt:1.8'

app的build.gradle 应用apt插件，并设置apt生成的索引的包名和类名

1
2
3
4
5
6
apply plugin: 'com.neenbedankt.android-apt'
apt {
    arguments {
        eventBusIndex "com.android.gallery3d.MySubscriberInfoIndex"
    }
}

app的dependencies中引入EventBusAnnotationProcessor：

1
apt 'org.greenrobot:eventbus-annotation-processor:3.0.1'

把注解生成的索引添加到EventBus默认的单例中，注意installDefaultEventBus这个方法只能调用一次

1
EventBus.builder().addIndex(new MySubscriberInfoIndex()).installDefaultEventBus();

EventBus原理浅析

EventBus.getDefault()方法

1
2
3
4
5
6
7
8
9
10
public static EventBus getDefault() {
    if (defaultInstance == null) {
        synchronized (EventBus.class) {
            if (defaultInstance == null) {
                defaultInstance = new EventBus();
            }
        }
    }
    return defaultInstance;
}

EventBus.getDefault().register(this)方法

1
2
3
4
5
6
7
8
9
public void register(Object subscriber) {
    Class<?> subscriberClass = subscriber.getClass();
    List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
    synchronized (this) {
        for (SubscriberMethod subscriberMethod : subscriberMethods) {
            subscribe(subscriber, subscriberMethod);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;

private final Map<Object, List<Class<?>>> typesBySubscriber;  

private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
    Class<?> eventType = subscriberMethod.eventType;
    Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
    CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions == null) {
        subscriptions = new CopyOnWriteArrayList<>();
        subscriptionsByEventType.put(eventType, subscriptions);
    } else {
        if (subscriptions.contains(newSubscription)) {
            throw new EventBusException("Subscriber " + subscriber.getClass() + " already registered to event "
                    + eventType);
        }
    }

    int size = subscriptions.size();
    for (int i = 0; i <= size; i++) {
        if (i == size || subscriberMethod.priority > subscriptions.get(i).subscriberMethod.priority) {
            subscriptions.add(i, newSubscription);
            break;
        }
    }

    List<Class<?>> subscribedEvents = typesBySubscriber.get(subscriber);
    if (subscribedEvents == null) {
        subscribedEvents = new ArrayList<>();
        typesBySubscriber.put(subscriber, subscribedEvents);
    }
    subscribedEvents.add(eventType);

    if (subscriberMethod.sticky) {
        if (eventInheritance) {
            // Existing sticky events of all subclasses of eventType have to be considered.
            // Note: Iterating over all events may be inefficient with lots of sticky events,
            // thus data structure should be changed to allow a more efficient lookup
            // (e.g. an additional map storing sub classes of super classes: Class -> List<Class>).
            Set<Map.Entry<Class<?>, Object>> entries = stickyEvents.entrySet();
            for (Map.Entry<Class<?>, Object> entry : entries) {
                Class<?> candidateEventType = entry.getKey();
                if (eventType.isAssignableFrom(candidateEventType)) {
                    Object stickyEvent = entry.getValue();
                    checkPostStickyEventToSubscription(newSubscription, stickyEvent);
                }
            }
        } else {
            Object stickyEvent = stickyEvents.get(eventType);
            checkPostStickyEventToSubscription(newSubscription, stickyEvent);
        }
    }
}
1
2
3
4
5
6
7
private void checkPostStickyEventToSubscription(Subscription newSubscription, Object stickyEvent) {
    if (stickyEvent != null) {
        // If the subscriber is trying to abort the event, it will fail (event is not tracked in posting state)
        // --> Strange corner case, which we don't take care of here.
        postToSubscription(newSubscription, stickyEvent, Looper.getMainLooper() == Looper.myLooper());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    switch (subscription.subscriberMethod.threadMode) {
        case POSTING:
            invokeSubscriber(subscription, event);
            break;
        case MAIN:
            if (isMainThread) {
                invokeSubscriber(subscription, event);
            } else {
                mainThreadPoster.enqueue(subscription, event);
            }
            break;
        case BACKGROUND:
            if (isMainThread) {
                backgroundPoster.enqueue(subscription, event);
            } else {
                invokeSubscriber(subscription, event);
            }
            break;
        case ASYNC:
            asyncPoster.enqueue(subscription, event);
            break;
        default:
            throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
    }
}

EventBus.getDefault().post(new LoadEvent());

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void post(Object event) {
    PostingThreadState postingState = currentPostingThreadState.get();
    List<Object> eventQueue = postingState.eventQueue;
    eventQueue.add(event);

    if (!postingState.isPosting) {
        postingState.isMainThread = Looper.getMainLooper() == Looper.myLooper();
        postingState.isPosting = true;
        if (postingState.canceled) {
            throw new EventBusException("Internal error. Abort state was not reset");
        }
        try {
            while (!eventQueue.isEmpty()) {
                postSingleEvent(eventQueue.remove(0), postingState);
            }
        } finally {
            postingState.isPosting = false;
            postingState.isMainThread = false;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private void postSingleEvent(Object event, PostingThreadState postingState) throws Error {
    Class<?> eventClass = event.getClass();
    boolean subscriptionFound = false;
    if (eventInheritance) {
        List<Class<?>> eventTypes = lookupAllEventTypes(eventClass);
        int countTypes = eventTypes.size();
        for (int h = 0; h < countTypes; h++) {
            Class<?> clazz = eventTypes.get(h);
            subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);
        }
    } else {
        subscriptionFound = postSingleEventForEventType(event, postingState, eventClass);
    }
    if (!subscriptionFound) {
        if (logNoSubscriberMessages) {
            Log.d(TAG, "No subscribers registered for event " + eventClass);
        }
        if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
                eventClass != SubscriberExceptionEvent.class) {
            post(new NoSubscriberEvent(this, event));
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
  private boolean postSingleEventForEventType(Object event, PostingThreadState postingState, Class<?> eventClass) {
    CopyOnWriteArrayList<Subscription> subscriptions;
    synchronized (this) {
        subscriptions = subscriptionsByEventType.get(eventClass);
    }
    if (subscriptions != null && !subscriptions.isEmpty()) {
        for (Subscription subscription : subscriptions) {
            postingState.event = event;
            postingState.subscription = subscription;
            boolean aborted = false;
            try {
                postToSubscription(subscription, event, postingState.isMainThread);
                aborted = postingState.canceled;
            } finally {
                postingState.event = null;
                postingState.subscription = null;
                postingState.canceled = false;
            }
            if (aborted) {
                break;
            }
        }
        return true;
    }
    return false;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) {
    switch (subscription.subscriberMethod.threadMode) {
        case POSTING:
            invokeSubscriber(subscription, event);
            break;
        case MAIN:
            if (isMainThread) {
                invokeSubscriber(subscription, event);
            } else {
                mainThreadPoster.enqueue(subscription, event);
            }
            break;
        case BACKGROUND:
            if (isMainThread) {
                backgroundPoster.enqueue(subscription, event);
            } else {
                invokeSubscriber(subscription, event);
            }
            break;
        case ASYNC:
            asyncPoster.enqueue(subscription, event);
            break;
        default:
            throw new IllegalStateException("Unknown thread mode: " + subscription.subscriberMethod.threadMode);
    }
}
1
2
3
4
5
6
7
8
9
void invokeSubscriber(Subscription subscription, Object event) {
    try {
        subscription.subscriberMethod.method.invoke(subscription.subscriber, event);
    } catch (InvocationTargetException e) {
        handleSubscriberException(subscription, event, e.getCause());
    } catch (IllegalAccessException e) {
        throw new IllegalStateException("Unexpected exception", e);
    }
}

EventBus.getDefault().unregister(this);

1
2
3
4
5
6
7
8
9
10
11
12
13
private final Map<Object, List<Class<?>>> typesBySubscriber;

public synchronized void unregister(Object subscriber) {
    List<Class<?>> subscribedTypes = typesBySubscriber.get(subscriber);
    if (subscribedTypes != null) {
        for (Class<?> eventType : subscribedTypes) {
            unsubscribeByEventType(subscriber, eventType);
        }
        typesBySubscriber.remove(subscriber);
    } else {
        Log.w(TAG, "Subscriber to unregister was not registered before: " + subscriber.getClass());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private final Map<Class<?>, CopyOnWriteArrayList<Subscription>> subscriptionsByEventType;

private void unsubscribeByEventType(Object subscriber, Class<?> eventType) {
    List<Subscription> subscriptions = subscriptionsByEventType.get(eventType);
    if (subscriptions != null) {
        int size = subscriptions.size();
        for (int i = 0; i < size; i++) {
            Subscription subscription = subscriptions.get(i);
            if (subscription.subscriber == subscriber) {
                subscription.active = false;
                subscriptions.remove(i);
                i--;
                size--;
            }
        }
    }
}

EventBus缺点

  • 耦合性太低
  • 订阅的事件是利用反射的机制执行的
  • 不能跨进程

用RxJava实现简易RxBus

实现RxBus

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public class RxBus {
    private static volatile RxBus sInstance;
    private final Subject subject;

    private RxBus() {
        subject = new SerializedSubject<>(PublishSubject.create());
    }

    public static RxBus getInstance() {
        if (sInstance == null) {
            synchronized (RxBus.class) {
                if (sInstance == null) {
                    sInstance = new RxBus();
                }
            }
        }
        return sInstance;
    }

    public void post(Object event) {
        subject.onNext(event);
    }

    public <T> Observable<T> toObservable(Class<T> eventType) {
        return subject.ofType(eventType);
    }
}

订阅事件

1
2
3
4
5
6
7
8
9
RxBus.getInstance().toObservable(ForceUpdateEvent.class)
        .observeOn(AndroidSchedulers.mainThread())
        .compose(this.<ForceUpdateEvent>bindUntilEvent(ActivityEvent.DESTROY))
        .subscribe(new Action1<ForceUpdateEvent>() {
            @Override
            public void call(ForceUpdateEvent forceUpdateEvent) {
                //do some thing
            }
        });

发布事件

1
RxBus.getInstance().post(new ForceUpdateEvent(parameter));
1
2
public class ForceUpdateEvent {
}

参考链接