作为一个Android开发者，如果你还没开始使用Kotlin，那么你可能要思考下自己的步伐是不是慢了。本文带着大家动手来写一个Kotlin版本的EventBus，一方面来看看Kotlin长什么样，一方面理解下EventBus的设计思想。
EventBus原理
EventBus 是基于观察者模式，核心是事件。通过事件的发布和订阅实现组件之间的通信，EventBus默认是一个单例存在，在Java中还需要使用Synchronized来保证线程安全。
通俗来讲，EventBus通过注册将所有订阅事件的方法储存在集合中，当有事件发布的时候，根据某些规则，匹配出符合条件的方法，调用执行，从而实现组件间的通信。
发布的事件相当于被观察者，注册的对象相当于观察者，被观察者和观察者是一对多的关系。当被观察者状态发生变化，即发布事件的时候，观察者对象将会得到通知并作出响应，即执行对应的方法。
入口-EventBus
还是直接粘贴上代码吧，毕竟这是最直观的。
package core.zs.eventbus
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
/**
* EventBus.
*
* Created by Jonash on 2018/2/5.
*/
object EventBus {
// 因为注册,取消可能耗时，因此使用多线程进行管理
private val executorService: ExecutorService = Executors.newCachedThreadPool()
// 默认的事件tag为""
private const val TAG_DEFAULT = ""
// 很据事件类型找到所有的注解方法信息
private val subscriberMap = mutableMapOf<EventType, CopyOnWriteArrayList<Subscription>>()
// Method find helper
private val methodFinder = SubscriberMethodFinder(subscriberMap)
/**
* 注册观察者。
* @param obj 观察者对象
*/
fun register(obj: Any) = executorService.execute {
methodFinder.findSubscribeMethods(obj)
}
/**
* 发送事件通知。
* @param event 事件
* @param tag 事件标签，默认值为“”
*/
@JvmOverloads
fun post(event: IEvent, tag: String = TAG_DEFAULT) {
val eventType = EventType(event.javaClass, tag)
val list = methodFinder.getMatchEventType(eventType)
list?.let {
EventDispatcher.dispatchEvent(event, it)
}
}
/**
* 取消观察者。
* @param obj 观察者对象
*/
fun unregister(obj: Any) = executorService.execute {
methodFinder.removeSubscriberMethod(obj)
}
fun getExecutorService() = executorService
}
设计要点：
- EventBus类型为object,实际上编译为Java后是一个单例的存在。
- 注册和取消注册使用线程，因为查找的的过程是一个耗时的存在。(你可以考虑使用编译时注解)
- post使用了默认参数，默认参数为"",你甚至可以忽略它的存在；但是当你需要特别定制接收同一事件类型的方法时，它会体现出用处。
- 为什么没使用Builder模式来实现EventBus?答:没事件。
- 根据事件的类型[EventType]来存储观察者信息。
事件类型EventType
package core.zs.eventbus
/**
* 记录事件类型信息。
*
* Created by Jonash on 2018/2/5.
*/
class EventType(private val eventClass: Class<*>, private val tag: String) {
override fun equals(other: Any?): Boolean {
// 比较内存引用地址，相同则返回 true
if (this === other) {
return true
}
// 判断是否为空，是否属于同一种类型
if (other == null || (other.javaClass.name !== this.javaClass.name)) {
return false
}
// 能执行到这里，说明obj和this同类且非 null
val eventType = other as EventType
val tagJudge = tag == eventType.tag
val eventJudge = eventClass.name == eventType.eventClass.name
// tag相同
// EventType是同一个类型
return tagJudge && eventJudge
}
override fun hashCode(): Int {
var hash = 7
hash = hash * 31 + eventClass.hashCode()
hash = hash * 31 + tag.hashCode()
return hash
}
}
设计要点:
- 此类的存在主要是为了存储事件类型信息。
- 因为此类是作为Map的key存在，所以eventClass和tag的类型均为val不可变。
- 同样是作为key的原因，我们重写了equals和hashCode方法。
注解-Subscriber
注解对于大家太常见了，在使用一些注入框架时经常见到，Subscriber注解用于标注订阅事件的方法。
package core.zs.eventbus.annotation
import core.zs.eventbus.ThreadMode
/**
* 事件接受方法的注解类
* tag:事件tag
* mode:观察者处理事件的线程，默认为post
*
* Created by Jonash on 2018/2/5.
*/
@Target(AnnotationTarget.FUNCTION) // 在方法上使用
@Retention(AnnotationRetention.RUNTIME) // 运行时注解，因为要使用反射
annotation class Subscriber(val tag: String, val mode: ThreadMode = ThreadMode.POSTING)
设计要点：
- 注解使用在方法上。
- 因为要反射解析注解，因此使用运行时注解。
IEvent
package core.zs.eventbus
/**
* 事件实现接口.
*
* Created by Jonash on 2018/2/5.
*/
abstract class IEvent
设计点：
- 仅仅是作为一个标示存在，检查订阅方法的参数必须继承了IEvent。
- 如果对事件有要求，可以修改该类。
Subscription
package core.zs.eventbus
import java.lang.ref.WeakReference
import java.lang.reflect.Method
/**
* 订阅方法的详细信息
*
* Created by Jonash on 2018/2/5.
*/
class Subscription(val subscriber: WeakReference<Any>,
private val targetMethod: Method,
val threadMode: ThreadMode) {
override fun equals(other: Any?): Boolean {
if (this === other) {
return true
}
if (other == null || (other::class !== this::class)) {
return false
}
val subscription = other as Subscription
val judgeSubscriber = this.subscriber.get() === subscription.subscriber.get()
val judgeMethod = this.targetMethod.name == subscription.targetMethod.name
return judgeSubscriber && judgeMethod
}
override fun hashCode(): Int {
var hash = 7
hash = hash * 31 + subscriber.hashCode()
hash = hash * 31 + targetMethod.hashCode()
hash = hash * 31 + threadMode.hashCode()
return hash
}
/**
* 根据传入的实例，反射调用实例方法。
*/
internal fun invoke(event: IEvent) {
targetMethod.invoke(subscriber.get(), event)
}
}
设计要点：
- 该类用于存储订阅者的详细信息。
- 为了防止重复存储，重写了equals和hashCode方法。
- 为了后面的反射调用方法，设计了invoke方法。
SubscriberMethodFinder-发现者
详细如果你仔细阅读了EventBus的代码，已经对该类优点眼熟，它主要用于实际的查找观察者、移除观察者。
package core.zs.eventbus
import core.zs.eventbus.annotation.Subscriber
import java.lang.ref.WeakReference
import java.lang.reflect.Modifier
import java.util.concurrent.CopyOnWriteArrayList
/**
* 订阅者辅助类
*
* Created by Jonash on 2018/2/5.
*/
class SubscriberMethodFinder(private val subscriberMap: MutableMap<EventType, CopyOnWriteArrayList<Subscription>>) {
companion object {
private val BRIDGE = 0x40
private val SYNTHETIC = 0x1000
private val MODIFIERS_IGNORE = Modifier.ABSTRACT or Modifier.STATIC or BRIDGE or SYNTHETIC
}
/**
* 查找订阅者中所有的注解信息。
* @param subscriber 订阅者
*/
@Synchronized
fun findSubscribeMethods(subscriber: Any) {
var clazz: Class<*>? = subscriber.javaClass
while (clazz != null && !isSystemClass(clazz.name)) {
var methods = try {
// 返回所有的方法，包括public/private/protected/default
clazz.declaredMethods
} catch (e: Exception) {
// public方法
clazz.methods
}
for (method in methods) {
val modifiers = method.modifiers
// 过滤方法的修饰符
if (Modifier.PUBLIC and modifiers != 0 && modifiers and MODIFIERS_IGNORE == 0) {
// 获取到注解
val annotation = method.getAnnotation(Subscriber::class.java)
// 如果注解不为空
if (annotation != null) {
val parameterTypes = method.parameterTypes
// 方法只接收一个参数
parameterTypes?.let {
if (parameterTypes.size == 1) {
var type = parameterTypes[0]
if (isAssignableFrom(IEvent::class.java, type)) {
val eventType = EventType(type as Class<IEvent>, annotation.tag)
val subscription = Subscription(WeakReference(subscriber), method, annotation.mode)
// 保存订阅信息
subscribe(eventType, subscription)
}
}
}
}
}
}
clazz = clazz.superclass
}
}
@Synchronized
private fun subscribe(type: EventType, subscription: Subscription) {
var subscriptionLists: CopyOnWriteArrayList<Subscription>? = getMatchEventType(type)
if (subscriptionLists == null) {
subscriptionLists = CopyOnWriteArrayList()
}
// 这就是为什么我们要重写equals和hashCode方法的原因
if (subscriptionLists.contains(subscription)) {
return
}
subscriptionLists.add(subscription)
// 将事件类型key和订阅者信息存储到map中
subscriberMap.put(type, subscriptionLists)
}
internal fun getMatchEventType(type: EventType): CopyOnWriteArrayList<Subscription>? {
val keys = subscriberMap.keys
return keys.firstOrNull { it == type }?.let { subscriberMap[it] }
}
@Synchronized
fun removeSubscriberMethod(subscriber: Any) {
// 注意删除的时候要使用迭代器
var iterator = subscriberMap.values.iterator()
while (iterator.hasNext()) {
val subscriptions: MutableList<Subscription>? = iterator.next()
subscriptions?.let { it: MutableList<Subscription>? ->
val subIterator = it!!.iterator()
while (subIterator.hasNext()) {
val subscription = subIterator.next()
// 获取引用
val cacheObject = subscription.subscriber.get()
cacheObject?.let {
if (isSameObject(cacheObject, subscriber)) {
subscriptions.remove(subscription)
}
}
}
}
// 如果针对某个Event的订阅者数量为空了,那么需要从map中清除
if (subscriptions == null || subscriptions.isEmpty()) {
iterator.remove()
}
}
}
/** 判断是否是同一个对象 */
private fun isSameObject(subOne: Any, subTwo: Any) = subOne === subTwo
// 判断类是否为系统类
private fun isSystemClass(clazzName: String): Boolean {
if (clazzName.startsWith("java.") || clazzName.startsWith("javax.") || clazzName.startsWith("android.")) {
return true
}
return false
}
// 判断某个类是否实现了IEvent接口
private fun isAssignableFrom(a: Class<*>, b: Class<*>): Boolean = a.isAssignableFrom(b)
}
EventDispatcher-事件分发者
看下怎样发送一个事件：
/**
* 发送事件通知。
* @param event 事件
* @param tag 事件标签，默认值为“”
*/
@JvmOverloads
fun post(event: IEvent, tag: String = TAG_DEFAULT) {
val eventType = EventType(event.javaClass, tag)
// 找出所有的事件订阅者
val list = methodFinder.getMatchEventType(eventType)
list?.let {
// 分发事件
EventDispatcher.dispatchEvent(event, it)
}
}
EventDispatcher的实现：
package core.zs.eventbus
import android.os.Looper
import core.zs.eventbus.handler.*
import java.util.concurrent.CopyOnWriteArrayList
/**
* 事件分发者
*
* Created by Jonash on 2018/2/5.
*/
object EventDispatcher {
private val postHandler = PostEventHandler()
private val mainHandler = MainEventHandler(Looper.getMainLooper())
private val asyncHandler = AsyncEventHandler()
private val bgHandler = BackgroundHandler()
fun dispatchEvent(event: IEvent, list: CopyOnWriteArrayList<Subscription>) =
list.forEach { it: Subscription ->
it.let {
val subscriber = it.subscriber.get()
subscriber?.let { subscriber: Any ->
val eventHandler = getEventHandler(it.threadMode)
eventHandler.handleEvent(it, event)
}
}
}
// 很据ThreadMode获取对应的事件处理器
private fun getEventHandler(mode: ThreadMode): EventHandler = when (mode) {
ThreadMode.POSTING -> postHandler
ThreadMode.ASYNC -> asyncHandler
ThreadMode.MAIN -> mainHandler
ThreadMode.BACKGROUND -> bgHandler
}
}
实现的是如此的easy。
EventHandler
我们在设计注解的时候，定义了一个订阅者执行线程的模式。
package core.zs.eventbus
/**
* ThreadMode枚举类
*
* Created by Jonash on 2018/2/5.
*/
enum class ThreadMode {
POSTING, MAIN, ASYNC, BACKGROUND
}
说实话这个类完全是看的Java EventBus的代码。
EventHandler的实现：
package core.zs.eventbus.handler
import core.zs.eventbus.IEvent
import core.zs.eventbus.Subscription
interface EventHandler {
fun handleEvent(subscription: Subscription, event: IEvent)
}
设计要点：
- EventHandler就是一个接口。
- 如果你需要，你完全可以实现该接口。
- 为了对应ThreadMode的几种模式，设计实现了四种类型的事件处理。
PostEventHandler
package core.zs.eventbus.handler
import core.zs.eventbus.IEvent
import core.zs.eventbus.Subscription
class PostEventHandler : EventHandler {
override fun handleEvent(subscription: Subscription, event: IEvent) =subscription.invoke(event)
}
package core.zs.eventbus.handler
import android.os.Handler
import android.os.Looper
import android.os.Message
import core.zs.eventbus.IEvent
import core.zs.eventbus.PendingPost
import core.zs.eventbus.PendingPostQueue
import core.zs.eventbus.Subscription
class MainEventHandler(looper: Looper) : Handler(looper), EventHandler {
private val queue: PendingPostQueue = PendingPostQueue()
private var handlerActive = false
override fun handleMessage(msg: Message?) {
while (true) {
var pendingPost = queue.poll()
if (pendingPost == null) {
synchronized(this) {
pendingPost = queue.poll()
if (pendingPost == null) {
handlerActive = false
return
}
}
}
pendingPost!!.subscription!!.invoke(pendingPost!!.event!!)
PendingPost.releasePendingPost(pendingPost!!)
}
}
override fun handleEvent(subscription: Subscription, event: IEvent) {
val post = PendingPost.obtainPendingPost(subscription, event)
synchronized(this) {
queue.enqueue(post)
if (!handlerActive) {
handlerActive = true
sendMessage(Message.obtain())
}
}
}
}
AsyncEventHandler
package core.zs.eventbus.handler
import core.zs.eventbus.*
class AsyncEventHandler : EventHandler {
private val queue: PendingPostQueue = PendingPostQueue()
override fun handleEvent(subscription: Subscription, event: IEvent) {
val pendingPost = PendingPost.obtainPendingPost(subscription, event)
queue.enqueue(pendingPost)
EventBus.getExecutorService().execute {
val pendingPost = queue.poll() ?: throw IllegalStateException("No pending post available")
pendingPost.subscription!!.invoke(pendingPost!!.event!!)
PendingPost.releasePendingPost(pendingPost)
}
}
}
BackgroundHandler
package core.zs.eventbus.handler
import core.zs.eventbus.*
class BackgroundHandler : Runnable, EventHandler {
private val queue: PendingPostQueue = PendingPostQueue()
@Volatile
private var executorRunning: Boolean = false
override fun handleEvent(subscription: Subscription, event: IEvent) {
val pendingPost = PendingPost.obtainPendingPost(subscription, event)
synchronized(this) {
queue.enqueue(pendingPost)
if (!executorRunning) {
executorRunning = true
EventBus.getExecutorService().execute(this)
}
}
}
override fun run() = try {
try {
while (true) {
var pendingPost = queue.poll(1000)
if (pendingPost == null) {
synchronized(this) {
pendingPost = queue.poll()
if (pendingPost == null) {
executorRunning = false
return
}
}
}
pendingPost!!.subscription!!.invoke(pendingPost!!.event!!)
PendingPost.releasePendingPost(pendingPost!!)
}
} catch (e: InterruptedException) {
e.printStackTrace()
}
} finally {
executorRunning = false
}
}
PendingPost缓存池
缓存池的实现主要包含两块内容：PendingPostQueue和PendingPost，使用了常规的实现。
PendingPostQueue
package core.zs.eventbus
/**
* PendingPost队列
*
* Created by Jonash on 2018/2/6.
*/
class PendingPostQueue {
private var head: PendingPost? = null
private var tail: PendingPost? = null
@Synchronized
fun enqueue(post: PendingPost) = when {
tail != null -> {
tail!!.next = post
tail = post
}
head == null -> {
head = post
tail = post
}
else -> throw IllegalStateException("Head present, but no tail")
}
@Synchronized
fun poll(): PendingPost? {
val post = head
if (head != null) {
head = head!!.next
if (head == null) {
tail = null
}
}
return post
}
@Synchronized
@Throws(InterruptedException::class)
fun poll(maxMillisToWait: Int): PendingPost? {
if (head == null) {
Thread.sleep(maxMillisToWait.toLong())
}
return poll()
}
}
PendingPost:
package core.zs.eventbus
/**
* 待处理请求。<br/>
* 使用缓存池进行管理请求。
*
* Created by Jonash on 2018/2/6.
*/
class PendingPost(var event: IEvent?, var subscription: Subscription?, var next: PendingPost? = null) {
companion object {
private val pool = arrayListOf<PendingPost>()
@JvmStatic
fun obtainPendingPost(subscription: Subscription, event: IEvent): PendingPost {
if (pool.size > 0) {
val pendingPost = pool.removeAt(pool.size - 1)
pendingPost.next = null
pendingPost.subscription = subscription
pendingPost.event = event
return pendingPost
}
return PendingPost(event, subscription)
}
@JvmStatic
fun releasePendingPost(pendingPost: PendingPost) {
pendingPost.event = null
pendingPost.subscription = null
pendingPost.next = null
synchronized(pool) {
if (pool.size < 1000) {
pool.add(pendingPost)
}
}
}
}
}
总结语
EventBus在我们的开发中如此常见，以至于我们才动手实现了这个简化Kotlin版本，希望你看完有所收获。
