EventBus(事件总线)的使用和源码的简单解析

news2024/12/23 0:26:09

Google Guava EventBus(事件总线)的使用和源码的简单解析

什么是EventBus?

事件总线(EventBus)是一种广泛用于软件架构中的设计模式,用于实现解耦和松散耦合的通信机制。它可以帮助组织和管理应用程序中不同组件之间的通信,以提高应用程序的可维护性、可扩展性和灵活性。

在事件总线模式中,不同的组件通过订阅和发布事件来进行通信。发布者发布一个事件,订阅者可以订阅该事件并在事件发生时执行相关的操作。事件总线充当了中介者的角色,它负责管理所有的事件和订阅者,并确保事件正确地传递到所有订阅者。

事件总线模式可以在很多场景中使用,例如 Android 应用程序中的通信、分布式系统中的消息传递等。常见的事件总线实现包括 Google Guava 的 EventBus 和 Square 的 Otto。

总的来说,EventBus就是应用了发布者/订阅者模式,用来帮助我们进行各组件间通信的工具。我们这里解析的是Google Guava 的 EventBus,它的官方文档在这里

EventBus三要素

github上的图:
来自官网的图片
EventBus里有三个比较重要的概念:

  1. Event 消息事件
  2. Publisher 消息发布者
  3. Subscriber 消息订阅者

乍一看这订阅者/发布者模式还和观察者模式有几分相似之处,不同之处就在于EventBus的存在,它作为一个中间件充当了消息传递的助手,使得订阅者不必直接订阅发布者,订阅事件总线即可。

EventBus的五种线程模型

EventBus的线程模型指的是根据事件的发布和订阅所在的线程,决定应该在哪个线程中处理事件。EventBus中有以下四种线程模型:

    1. POSTING(默认):在发布事件的线程中执行,即同步执行,速度快。
    1. MAIN:在主线程(UI线程)中执行,如果当前线程是主线程,直接执行订阅方法;否则,通过主线程的Poster来执行。
    1. BACKGROUND:在后台线程中执行,如果当前线程是主线程,通过后台的Poster来执行;否则,直接执行订阅方法。
    1. ASYNC:在新的子线程中执行,每个事件都会创建一个新的子线程,即异步执行,速度较慢。

除此之外,还有一种模型:

    1. MAIN_ORDERED :MAIN_ORDERED 模式也是运行在主线程上的模式,不同于 MAIN 模式的是,它保证了事件的顺序性。也就是说,当一个事件在主线程中被发布时,它会先进入一个队列中,之后再一个个的被处理。这样可以保证相同事件类型的事件按照发送的顺序依次被执行。如果当前线程不是主线程,那么它就直接被执行,这也是为了避免在子线程中的事件被阻塞。

我们会在消息处理方法中利用@Subscribe注解指定线程模型。

EventBus的简单使用

使用EventBus三步走

这里我们根据github上的三步分为四步走:

  1. 定义Event事件类:
public class MessageEvent {
  private String message;
  public MessageEvent(String message){
      this.message = message;
  }

  public String getMessage(){
      return message;
  }

  public void setMessage(String message){
      this.message = message;
  }
}

这里定义了一个MessageEvent用于传递事件。

  1. 声明订阅者,注册订阅方法并且指定线程模型
    @Subscribe(threadMode = ThreadMode.MAIN)//选择线程模型,说明事件将在主线程中处理
    public void onMoonEvent(MessageEvent messageEvent){
        tv_message.setText(messageEvent.getMessage());
    }

这里在MainActivity里注册了一个订阅方法(消息处理方法),指定了线程模型为MAIN,说明订阅方法在主线程里执行。完整Demo在后面放出。

3.注册订阅者与订阅默认的事件总线

   bt_subscription.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                //MainActivity注册了这条事件总线
                if(!EventBus.getDefault().isRegistered(MainActivity.this)){
                    EventBus.getDefault().register(MainActivity.this);
                }
            }
        });

这里先判断MainActivity是否已经与默认的事件总线订阅,如果没有订阅就进行订阅。

4.发送消息事件–触发订阅方法

    bt_message.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                EventBus.getDefault().post(new MessageEvent("欢迎来到,德莱联盟"));
                finish();
            }
        });

这里我在第二个Activity里发送了一个消息事件到默认的事件总线中,这样MainActivity中的订阅方法就会被触发。

2.小Demo

这里我贴出我的小Demo:

  1. MainActivity:
public class MainActivity extends AppCompatActivity {

    private TextView tv_message;
    private Button bt_message;
    private Button bt_subscription;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        tv_message = findViewById(R.id.tv_message);
        bt_message = findViewById(R.id.bt_message);
        bt_subscription = findViewById(R.id.bt_subscription);

        bt_message.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                startActivity(new Intent(MainActivity.this,SecondActivity.class));
            }
        });

        bt_subscription.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                //MainActivity注册了这条事件总线
                if(!EventBus.getDefault().isRegistered(MainActivity.this)){
                    EventBus.getDefault().register(MainActivity.this);
                }
            }
        });
    }

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

    //通过参数类型来区分应该执行哪个方法

    @Subscribe(threadMode = ThreadMode.MAIN)//选择线程模型,说明事件将在主线程中处理
    public void onMoonEvent(MessageEvent messageEvent){
        tv_message.setText(messageEvent.getMessage());
    }

    @Subscribe(threadMode = ThreadMode.MAIN,sticky = true)
    public void ononStickyEvent(MessageEvent messageEvent){
        tv_message.setText(messageEvent.getMessage());
    }

    @Subscribe(threadMode = ThreadMode.MAIN)
    public void secondBack(String mes){
        Toast.makeText(this, mes, Toast.LENGTH_SHORT).show();
    }
}
  1. SecondActivity:
public class SecondActivity extends AppCompatActivity {

    TextView tv_message;
    Button bt_message;
    Button bt_sticky;
    Button mes;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_second);
        tv_message = findViewById(R.id.tv_message1);
        bt_message = findViewById(R.id.bt_message1);
        bt_sticky = findViewById(R.id.bt_sticky);
        mes = findViewById(R.id.mes_event);
        bt_message.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                EventBus.getDefault().post(new MessageEvent("欢迎来到,德莱联盟"));
                finish();
            }
        });

        bt_sticky.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                EventBus.getDefault().postSticky(new MessageEvent("粘性事件"));
                finish();
            }
        });

        mes.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                EventBus.getDefault().post(new String("另一个事件"));
                finish();
            }
        });
    }
}
  1. MessageEvent:
//自定义的Event事件
public class MessageEvent {
    private String message;
    public MessageEvent(String message){
        this.message = message;
    }

    public String getMessage(){
        return message;
    }

    public void setMessage(String message){
        this.message = message;
    }
}

https://github.com/MonsterTTL/Android-Impove/tree/master/EventBusDemo
具体的实例代码放在我的github上👆了,里面还涉及到了粘性事件以及多个订阅事件。

源码解析

getDefault()方法

先来看我们最常用的获取EventBus的方法

	static volatile EventBus defaultInstance;
    public static EventBus getDefault() {
        EventBus instance = defaultInstance;
        if (instance == null) {
            synchronized (EventBus.class) {
                instance = EventBus.defaultInstance;
                if (instance == null) {
                    instance = EventBus.defaultInstance = new EventBus();
                }
            }
        }
        return instance;
    }

很显然这里用到了DCL单例模式,确保缺省状态下EventBus的实例只有一个。我们顺着这个方法往下捋,看new EventBus()方法。

new EventBus()方法

	private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();
	
    public EventBus() {
        this(DEFAULT_BUILDER);
    }
    
    EventBus(EventBusBuilder builder) {
        logger = builder.getLogger();
        subscriptionsByEventType = new HashMap<>();
        typesBySubscriber = new HashMap<>();
        stickyEvents = new ConcurrentHashMap<>();
        mainThreadSupport = builder.getMainThreadSupport();
        mainThreadPoster = mainThreadSupport != null ? mainThreadSupport.createPoster(this) : null;
        backgroundPoster = new BackgroundPoster(this);
        asyncPoster = new AsyncPoster(this);
        indexCount = builder.subscriberInfoIndexes != null ? builder.subscriberInfoIndexes.size() : 0;
        subscriberMethodFinder = new SubscriberMethodFinder(builder.subscriberInfoIndexes,
                builder.strictMethodVerification, builder.ignoreGeneratedIndex);
        logSubscriberExceptions = builder.logSubscriberExceptions;
        logNoSubscriberMessages = builder.logNoSubscriberMessages;
        sendSubscriberExceptionEvent = builder.sendSubscriberExceptionEvent;
        sendNoSubscriberEvent = builder.sendNoSubscriberEvent;
        throwSubscriberException = builder.throwSubscriberException;
        eventInheritance = builder.eventInheritance;
        executorService = builder.executorService;
    }

很显然,获取default EventBus的调用流程是这样的:getDefault() -> EventBus()->EventBus(DEFAULT_BUILDER)。说道Builder,那么应该是采取建造者模式创建的,我们接下来分析EventBusBuilder。

EventBusBuilder

builder里延伸下去的内容有点多,我们直接先看EventBusBuilder的build方法:

public EventBus build() {
        return new EventBus(this);
    }

很显然,该类的作用是构建一个EventBus实例,并提供一些配置选项。该类具有多个属性,如是否记录订阅者异常、是否发送没有订阅者事件等,以及设置自定义线程池和日志处理程序等。类中的各种方法可用于配置这些属性,例如logSubscriberExceptions()方法可用于设置是否记录订阅者异常。 这里我们就不关注打印日志,记录异常等配置了,我们关注具体的创建过程。

我们先来看看以下几个可选的方法:

/*** 设置 EventBus 是否开启事件类继承的支持。默认情况下,EventBus会考虑事件类继承关系,即父类有订阅者时,子类的订阅者也会被
通知。但是,关闭这个特性可以提高事件发布的速度。如果事件类的继承层级比较复杂,关闭继承支持的性能提升应该会大于20%。但需要注意
的是,事件发布通常只占应用程序内部CPU时间的一小部分,除非高速发布事件,例如每秒钟发布数百/数千个事件。                                ***/
 public EventBusBuilder eventInheritance(boolean eventInheritance) {
        this.eventInheritance = eventInheritance;
        return this;
    }


    /**
     * 这是用来指定线程池类型的,除了使用Builder内部自带的线程池,我们也可以使用自己传入的线程池
     */
    public EventBusBuilder executorService(ExecutorService executorService) {
        this.executorService = executorService;
        return this;
    }

    /**
     * 这是用来跳过指定订阅者的方法验证的,当你注册一个类作为事件订阅者时,EventBus会自动检查该类是否具有合法的事件订阅方法。
     * 这些方法必须是公共的,没有返回值,只有一个参数,并且在方法上需要使用@Subscribe注解。然而,有些情况下,你可能希望让某
     * 个类作为订阅者,但是它并不符合这些要求。例如,它可能有一些不是用@Subscribe注解的方法,但是你仍然希望这些方法能够被
     * EventBus识别并调用。在这种情况下,你可以使用skipMethodVerificationFor()方法来跳过验证,允许这些方法被注册为事件订阅    
     * 方法。
     */
    public EventBusBuilder skipMethodVerificationFor(Class<?> clazz) {
        if (skipMethodVerificationForClasses == null) {
            skipMethodVerificationForClasses = new ArrayList<>();
        }
        skipMethodVerificationForClasses.add(clazz);
        return this;
    }

    /**  允许在生成了索引的情况下也强制使用反射进行订阅者的查找。默认情况下,如果有生成的索引,则会使用它来查找订阅者,
    而不是使用反射。通过调用这个方法并将其参数设置为 true,可以强制使用反射进行查找。 */
    public EventBusBuilder ignoreGeneratedIndex(boolean ignoreGeneratedIndex) {
        this.ignoreGeneratedIndex = ignoreGeneratedIndex;
        return this;
    }

    /** 开启严格方法验证 . 在 EventBus 中,严格检查是指在注册和订阅事件时检查相关方法的参数和返回类型是否正确匹配,以确保事
    件可以正确地被分发。如果开启了严格检查,那么在注册或订阅事件时,如果发现有方法的参数或返回类型与事件类型不匹配,EventBus 
    会抛出一个异常,防止程序出现意料之外的错误。*/
    public EventBusBuilder strictMethodVerification(boolean strictMethodVerification) {
        this.strictMethodVerification = strictMethodVerification;
        return this;
    }

    /** 用于添加 SubscriberInfoIndex 对象到 EventBusBuilder 中。SubscriberInfoIndex 接口用于提供订阅者类和订阅方法
    信息的索引,以便 EventBus 能够快速找到订阅者及其订阅方法。在 EventBus 中, */
    public EventBusBuilder addIndex(SubscriberInfoIndex index) {
        if (subscriberInfoIndexes == null) {
            subscriberInfoIndexes = new ArrayList<>();
        }
        subscriberInfoIndexes.add(index);
        return this;
    }

接下来看下比较重要的几个方法:

 public EventBus installDefaultEventBus() {
        synchronized (EventBus.class) {
            if (EventBus.defaultInstance != null) {
                throw new EventBusException("Default instance already exists." +
                        " It may be only set once before it's used the first time to ensure consistent behavior.");
            }
            EventBus.defaultInstance = build();
            return EventBus.defaultInstance;
        }
    }

这个方法是用来安装默认EventBus里的defaultInstance的:

class EventBus{
...
	static volatile EventBus defaultInstance;//默认的实例
	private static final EventBusBuilder DEFAULT_BUILDER = new EventBusBuilder();//默认的Builder
...
}

这段代码是通过构建者模式构建一个默认的EventBus实例,并安装在默认的EventBus上,只能在第一次使用默认的EventBus之前调用,否则会抛出EventBusException异常。在synchronized块内,首先判断是否已经存在默认的EventBus实例,如果存在则抛出异常,否则构建EventBus实例并将其设置为默认实例,最后返回默认实例。

接下来看另一个方法:

    MainThreadSupport getMainThreadSupport() {
        if (mainThreadSupport != null) {
            return mainThreadSupport;
        } else if (AndroidComponents.areAvailable()) {
            return AndroidComponents.get().defaultMainThreadSupport;
        } else {
            return null;
        }
    }

这个方法是用来获取主线程支持的,那什么是主线程支持呢,我们点进去看看:

public interface MainThreadSupport {

    boolean isMainThread();

    Poster createPoster(EventBus eventBus);
}

这里我们可以看到,MainThreadSupport是一个接口,定义了isMainThread和createPoster方法,根据这两个方法名,我们可以推测这个接口的作用就是用来支持消息在主线程里传递的,isMainThread用于判断当前线程是否是主线程,createPoster用于创建一个poster以在主线程里传递消息。

当然这只是一个接口,我们要看具体的实现还是得折回去看getMainThreadSupport方法,我们看这一段:

	else if (AndroidComponents.areAvailable()) {
	            return AndroidComponents.get().defaultMainThreadSupport;
	        }

这段代码的作用就是判断当前平台是不是Android平台,如果是Android平台,就调用Android.get().defaultMainThreadSupport获取默认的主线程支持类。那这里又得涉及到AndroidComponents类了,我们开一个子标题讨论这个类。

AndroidComponents类

AndroidComponents类是一个抽象类,内容比较短,这里先贴出它的全部源码:

public abstract class AndroidComponents {

    private static final AndroidComponents implementation;

    static {
        implementation = AndroidDependenciesDetector.isAndroidSDKAvailable()
            ? AndroidDependenciesDetector.instantiateAndroidComponents()
            : null;
    }

    public static boolean areAvailable() {
        return implementation != null;
    }

    public static AndroidComponents get() {
        return implementation;
    }

    public final Logger logger;
    public final MainThreadSupport defaultMainThreadSupport;

    public AndroidComponents(Logger logger, MainThreadSupport defaultMainThreadSupport) {
        this.logger = logger;
        this.defaultMainThreadSupport = defaultMainThreadSupport;
    }
}

这里将其声明为抽象的意义应该就是防止其实例化,因为AndroidComponents的具体实现是由AndroidDependenciesDetector来决定的,这个类会检测当前是否在Android环境下,如果是,则返回实际的实现,否则返回null。而AndroidDependenciesDetector这个类是不允许被外部直接实例化的,因此AndroidComponents也应该遵循同样的规则。

真正重要的代码部分是在静态代码块内,代码块在初始化时会会检测当前是否在Android环境下,如果是,则返回实际的实现,否则返回null。

接下来我们再深入向下扒,找到MainThreadSupport的具体实现类,再往下看我们可以发现这个类是由反射机制实现的,不过我们先不关心这个,直接找到具体实现类DefaultAndroidMainThreadSupport,直接看它的源码:

public class DefaultAndroidMainThreadSupport implements MainThreadSupport {

    @Override
    public boolean isMainThread() {
        return Looper.getMainLooper() == Looper.myLooper();
    }

    @Override
    public Poster createPoster(EventBus eventBus) {
        return new HandlerPoster(eventBus, Looper.getMainLooper(), 10);
    }
}

可以发现,这个类实际上十分简单,通过Looper判断是否是主线程,接下来看他的Poster类。
具体的Poster类:

public class HandlerPoster extends Handler implements Poster {

    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;
    private final EventBus eventBus;
    private boolean handlerActive;

    public HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
        super(looper);
        this.eventBus = eventBus;
        this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
        queue = new PendingPostQueue();
    }

    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!handlerActive) {
                handlerActive = true;
                if (!sendMessage(obtainMessage())) {
                    throw new EventBusException("Could not send handler message");
                }
            }
        }
    }
...

我们先来看这前半部分,在构造方法中新出现了一个queue,是一个post队列,是用于存储post请求的队列,具体的PendingPostQueue类我们就不在具体看了,只要知道其内部使用链表的结构连接的,而PendingPostQueue类里记录了这个post队列的头部和尾部。

enqueue方法中将发送的Post请求进行入队,如果handler不处于活跃状态,就将其置位活跃状态,然后获取一个Message对象并将其发送给MessageQueue中,接下来我们看它是如何处理post请求的:

public void handleMessage(Message msg) {
        boolean rescheduled = false;
        try {
            long started = SystemClock.uptimeMillis();
            while (true) {
                PendingPost pendingPost = queue.poll();
                if (pendingPost == null) {
                    synchronized (this) {
                        // Check again, this time in synchronized
                        pendingPost = queue.poll();
                        if (pendingPost == null) {
                            handlerActive = false;
                            return;
                        }
                    }
                }
                eventBus.invokeSubscriber(pendingPost);
                long timeInMethod = SystemClock.uptimeMillis() - started;
                if (timeInMethod >= maxMillisInsideHandleMessage) {
                    if (!sendMessage(obtainMessage())) {
                        throw new EventBusException("Could not send handler message");
                    }
                    rescheduled = true;
                    return;
                }
            }
        } finally {
            handlerActive = rescheduled;
        }
    }

由于HandlerPoster本身就是一个Handler,所以上一步发送的Message实际上就是由它自己处理的,就是在这个方法中处理,Post请求具体是在这里执行的:

eventBus.invokeSubscriber(pendingPost);

这个方法会将Event事件推送到具体的订阅者类中去执行,接下来我们重新回到EventBus类中看看这个过程是怎样执行的。

Subscription类

在正式介绍EventBus是怎样执行推送之前,我们需要先了解两个前置类的信息,首先我们先来看Subscriptiont类:

final class Subscription {
    final Object subscriber;
    final SubscriberMethod subscriberMethod;
    /**
     * Becomes false as soon as {@link EventBus#unregister(Object)} is called, which is checked by queued event delivery
     * {@link EventBus#invokeSubscriber(PendingPost)} to prevent race conditions.
     */
    volatile boolean active;

    Subscription(Object subscriber, SubscriberMethod subscriberMethod) {
        this.subscriber = subscriber;
        this.subscriberMethod = subscriberMethod;
        active = true;
    }

    @Override
    public boolean equals(Object other) {
        if (other instanceof Subscription) {
            Subscription otherSubscription = (Subscription) other;
            return subscriber == otherSubscription.subscriber
                    && subscriberMethod.equals(otherSubscription.subscriberMethod);
        } else {
            return false;
        }
    }

    @Override
    public int hashCode() {
        return subscriber.hashCode() + subscriberMethod.methodString.hashCode();
    }
}

这个类里面的具体的SubscriberMethod和Object subscriber就不详细介绍了,根据名字我们也可以知道,final Object subscriber 存储的是订阅事件的对象,而 final SubscriberMethod subscriberMethod 则包含了订阅事件的方法的信息,例如该方法的名称、参数类型等等。通过这两个属性,EventBus 可以将事件正确地分发给订阅者的对应方法。

PendingPost类

然后我们再来看PendingPost类

final class PendingPost {
    private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();

    Object event;
    Subscription subscription;
    PendingPost next;
    ...

}

我们看到这个类里有一个静态的变量:

private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();

就是说,所有的PendingPost类的实例都会共享这一个池子,这个池子主要是用来优化GC性能的,用于重用已创建的对象,避免了大量的对象创建和销毁,减小了内存开销。

1.构造方法

    private PendingPost(Object event, Subscription subscription) {
        this.event = event;
        this.subscription = subscription;
    }

其创建方法主要是传入发送的事件和订阅信息相关的Subscription,然后将其包装成一个PendingPost对象。

static PendingPost obtainPendingPost(Subscription subscription, Object event) {
        synchronized (pendingPostPool) {
            int size = pendingPostPool.size();
            if (size > 0) {
                PendingPost pendingPost = pendingPostPool.remove(size - 1);
                pendingPost.event = event;
                pendingPost.subscription = subscription;
                pendingPost.next = null;
                return pendingPost;
            }
        }
        return new PendingPost(event, subscription);
    }

obtainPendingPost方法是根据pendingPostPool这个共享池来创建PendingPost对象,如果共享池里还有可用对象,就复用这个对象来创建新的PendingPost对象,否则就直接创建一个新的对象。

 static void releasePendingPost(PendingPost pendingPost) {
        pendingPost.event = null;
        pendingPost.subscription = null;
        pendingPost.next = null;
        synchronized (pendingPostPool) {
            // Don't let the pool grow indefinitely
            if (pendingPostPool.size() < 10000) {
                pendingPostPool.add(pendingPost);
            }
        }
    }

最后是release方法,这个方法是用来将PendingPost对象释放会共享池中的,不过这个共享池的大小也是有限制的,最大不能超过10000。

EventBus类中如何推送事件

上面说到,具体是通过invokeSubscriber这个方法来推送事件的,事不宜迟,我们马上来看看这个方法的内容:

    void invokeSubscriber(PendingPost pendingPost) {
        Object event = pendingPost.event;
        Subscription subscription = pendingPost.subscription;
        PendingPost.releasePendingPost(pendingPost);
        if (subscription.active) {
            invokeSubscriber(subscription, event);
        }
    }

可以发现,具体又调用了invokeSubscriber(subscription, event);方法,我们紧接着看invokeSubscriber(subscription, event)方法:

    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);
        }
    }

很显然,这里是通过了反射机制,subscription.subscriberMethod.method.invoke(subscription.subscriber, event)表示将event事件发送到subscriber中并触发subscriber的处理方法。method是SubscriberMethod中的一个字段,表示订阅方法本身,通过invoke方法来调用这个方法,并将subscriber作为方法的调用者,event作为参数传递进去。到这里,我们就看完Event事件是如何传递给订阅者的了。

Register(订阅事件总线)

接下来我们看第二个内容,订阅事件总线时做了什么:

public void register(Object subscriber) {
		...
        Class<?> subscriberClass = subscriber.getClass();
        List<SubscriberMethod> subscriberMethods = subscriberMethodFinder.findSubscriberMethods(subscriberClass);
        synchronized (this) {
            for (SubscriberMethod subscriberMethod : subscriberMethods) {
                subscribe(subscriber, subscriberMethod);
            }
        }
    }

核心方法在于subscribe方法:

    private void subscribe(Object subscriber, SubscriberMethod subscriberMethod) {
        Class<?> eventType = subscriberMethod.eventType;
        Subscription newSubscription = new Subscription(subscriber, subscriberMethod);
        CopyOnWriteArrayList<Subscription> subscriptions = subscriptionsByEventType.get(eventType);//获取同一个类型的消息事件下的所有Subscrption对象
        if (subscriptions == null) {
            subscriptions = new CopyOnWriteArrayList<>();
            subscriptionsByEventType.put(eventType, subscriptions);//如果原来还没有注册这一类型的消息事件,则新创建一个List并加入
        } 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;
            }
        }//这一段是用来将新的subscription对象插入到同一类型的消息事件队列的合适位置中--将订阅者与注册方法关联起来

        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);
            }
        }
    }

具体来说,subscribe方法做了三件事情:
1. 维护了当前EventBus中的subscriptionsByEventType表;
2. 维护了当前EventBus中的 typesBySubscriber表;
3. 处理了粘性事件;

详细的说明我放在上面的注释里面了,我们先说维护的两个表,第一个表是代表消息事件类型和Subscription关联的一个表,
第二个表是表示订阅者Subscriber和消息事件类型关联的两个表,这两个表可以在发布事件时可以快速地找到对应的订阅者并调用其处理方法。

然后是粘性事件的处理,这里有一个标志位eventInheritance,这个标志位我们在之前提到过,我们可以回到EventBuilder的内容查看,主要是涉及到继承问题的,除去继承的问题,粘性事件主要就是调用:

   checkPostStickyEventToSubscription(newSubscription, stickyEvent);

直接点开这个方法会发现其调用了:

    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, isMainThread());
        }
    }
	
	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 MAIN_ORDERED:
                if (mainThreadPoster != null) {
                    mainThreadPoster.enqueue(subscription, event);
                } else {
                    // temporary: technically not correct as poster not decoupled from subscriber
                    invokeSubscriber(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);
        }
    }

实际上是调用了postToSubscription方法,该方法根据回调方法指定的线程模型来传递粘性事件,至于上面提到的三种poster,我们之后再解析。

Post方法

订阅者注册完毕后,我们就要用Post方法进行消息的发送了,Post方法又可以分为post和postSticky,接下来我们先看post方法:

    public void post(Object event) {
        PostingThreadState postingState = currentPostingThreadState.get();
        List<Object> eventQueue = postingState.eventQueue;//获取发送线程的eventQueue
        eventQueue.add(event);//将当前要发送时间添加进eventQueue中

        if (!postingState.isPosting) {//如果当前线程处于空闲状态
            postingState.isMainThread = isMainThread();
            postingState.isPosting = true;
            if (postingState.canceled) {
                throw new EventBusException("Internal error. Abort state was not reset");
            }
            try {
                while (!eventQueue.isEmpty()) {//eventQueue不为空
                    postSingleEvent(eventQueue.remove(0), postingState);//通过postSingleEvent方法依次将eventQueue中的事件发送
                }
            } finally {
                postingState.isPosting = false;
                postingState.isMainThread = false;
            }
        }
    }

这里又冒出来了一个PostingThreadState,我们看它是从哪里来的:

    private final ThreadLocal<PostingThreadState> currentPostingThreadState = new ThreadLocal<PostingThreadState>() {
        @Override
        protected PostingThreadState initialValue() {
            return new PostingThreadState();
        }
    };
...
    final static class PostingThreadState {
        final List<Object> eventQueue = new ArrayList<>();
        boolean isPosting;
        boolean isMainThread;
        Subscription subscription;
        Object event;
        boolean canceled;
    }

这里涉及到一个ThreadLocal类,我们先来看看:

在Java中,ThreadLocal是一个非常有用的类,它允许您在单个线程中存储和检索数据,而不会与其他线程共享。ThreadLocal通过创建一个副本来解决线程安全问题,每个线程都有自己的副本,所以每个线程都可以独立地改变自己的副本,而不会影响其他线程。
ThreadLocal提供了三个主要方法:
get() - 返回当前线程的变量副本(如果有);如果没有,则返回默认值。
set(T value) - 设置当前线程的变量副本。
remove() - 移除当前线程的变量副本。
ThreadLocal通常用于在单个线程中存储和检索上下文信息,例如数据库连接、事务对象等。使用ThreadLocal可以避免在多线程环境下对这些资源进行同步操作,从而提高性能。

所以在这里的语境下,PostingThreadState就是每个发送post请求线程中的局部变量,且各个发送线程之间是不会共享这个PostingThreadState变量的,每个发送post请求的线程都会有自己私有的PostingThreadState变量。

post方法做的大致如下:

  1. 先获取到当前发送线程的PostingThreadState变量

  2. 接着获取当前线程要发送的事件队列eventQueue,然后将当前要发送的事件添加进入eventQueue队列中

  3. 如果当前发送线程处于空闲状态,就将eventQueue中的事件通过postSingleEvent方法依次发送

    既然说调用了postSingleEvent方法,接下来我们理所当然地查看postSingleEvent方法:

    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) {
                logger.log(Level.FINE, "No subscribers registered for event " + eventClass);
            }
            if (sendNoSubscriberEvent && eventClass != NoSubscriberEvent.class &&
                    eventClass != SubscriberExceptionEvent.class) {
                post(new NoSubscriberEvent(this, event));
            }
        }
    }

这里又涉及到了eventInheritance标志位,这里我们不管这个标志位,查看这个方法的核心在于:

 subscriptionFound |= postSingleEventForEventType(event, postingState, clazz);

我们接着查看这个postSingleEventForEventType方法:

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;
                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;
    }

这个方法接收了发送的事件event,发送线程的局部变量postingState,还有事件类型eventClass。接下来它获取了对应事件类型下的subscriptions队列,对其中的每个subscription对象调用了postToSubscription方法,如果一切成功,就返回true,否则返回false。至于postToSubscription这个方法我们在之前也提到过,就是:

 private void postToSubscription(Subscription subscription, Object event, boolean isMainThread) 

上次提到它是用来传递粘性事件,当然它也可以用来传递普通事件,这个方法将根据指定的线程模型来选择如何传递事件给订阅者回调处理。

至于粘性事件的POST:

    public void postSticky(Object event) {
        synchronized (stickyEvents) {
            stickyEvents.put(event.getClass(), event);
        }
        // Should be posted after it is putted, in case the subscriber wants to remove immediately
        post(event);
    }

不同点就在于在执行post方法之前还将其加入到了粘性事件队列stickyEvents,以便在其他订阅者完成注册时能接收到粘性事件。

三种Poster

之前提到过EventBus里有三种Poster,在MainThreadSuppot中我们已经认识了HandlerPoster,接下来我们看看其他两种Poster,这里我们为了方便,贴出postToSubscription的源码方便理解:

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 MAIN_ORDERED:
                if (mainThreadPoster != null) {
                    mainThreadPoster.enqueue(subscription, event);
                } else {
                    // temporary: technically not correct as poster not decoupled from subscriber
                    invokeSubscriber(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);
        }
    }

很显然,根据不同的线程模型,该方法会采取的不同的策略和poster,mainThreadPoster就是我们一开始解析过的HandlerPoster,接下来先看backgroundPoster:

	private final PendingPostQueue queue;
    private final EventBus eventBus;
	private volatile boolean executorRunning;
	
    BackgroundPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }

BackgroundPoster内部会有自己的一条PendingPostQueue,接下来我们看入队方法:

 public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!executorRunning) {
                executorRunning = true;
                eventBus.getExecutorService().execute(this);
            }
        }
    }

这里BackgroundPoster用了同步代码块,说明虽然Background模型是在子线程中运行的,但是其会遵循顺序依次执行,同一个发送队列发送的事件触发的多个Background线程模型的回调方法不能同时运行。

至于任务该如何执行,eventBus中会有一个线程池,这个线程池类型是根据builder决定的,默认情况下,这个线程池的类型将会是CachedThreadPool。接下来再看它的run方法:

public void run() {
        try {
            try {
                while (true) {
                    PendingPost pendingPost = queue.poll(1000);
                    if (pendingPost == null) {
                        synchronized (this) {
                            // Check again, this time in synchronized
                            pendingPost = queue.poll();
                            if (pendingPost == null) {
                                executorRunning = false;
                                return;
                            }
                        }
                    }
                    eventBus.invokeSubscriber(pendingPost);
                }
            } catch (InterruptedException e) {
                eventBus.getLogger().log(Level.WARNING, Thread.currentThread().getName() + " was interruppted", e);
            }
        } finally {
            executorRunning = false;
        }
    }

run方法很简单,就是不断将queue中的PendingPost取出放入线程池中执行eventBus的invokeSubscriber方法:
在这里插入图片描述
接下来我们看AsncPoster:

AsyncPoster(EventBus eventBus) {
        this.eventBus = eventBus;
        queue = new PendingPostQueue();
    }

    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        queue.enqueue(pendingPost);
        eventBus.getExecutorService().execute(this);
    }

    @Override
    public void run() {
        PendingPost pendingPost = queue.poll();
        if(pendingPost == null) {
            throw new IllegalStateException("No pending post available");
        }
        eventBus.invokeSubscriber(pendingPost);
    }

AsyncPoster和BackgroundPoster其实有点像,不同之处就在于AsyncPoster的post请求不必顺序执行,多个回调可以同时进行。

总结

到此为止,我们就已经解析了EventBus源码的核心基础部分了,让我们总结一下流程图:在这里插入图片描述
这里的流程图是简化的版本,这里具体实现的细节被隐藏起来了,主要是帮助我们理解机制流程。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/484246.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

【SQL篇】面试之高级查询和连接

603 连续空余座位 select distinct c1.seat_id from Cinema c1 join Cinema c2 on abs(c2.seat_id-c1.seat_id) 1 where c1.free1 and c2.free1 order by c1.seat_id;总结 思路&#xff1a;为什么我们这里需要abs和distinct&#xff0c;如果是如下代码&#xff0c;为什么不可…

[架构之路-178]-《软考-系统分析师》-17-嵌入式系统分析与设计- 3- 分区操作系统(Partition Operating System)概述

目录&#xff1a; 本文概述&#xff1a; 1.1 什么是分区操作系统 1.2 分区操作系统出现背景 1. 前后台系统(Foreground/Background System) 2. 实时操作系统(RTOS) 本文概述&#xff1a; 随着嵌入式系统日趋复杂化以及对安全性要求的不断提高&#xff0c;采用空间隔离、时…

[计算机图形学]光场,颜色与感知(前瞻预习/复习回顾)

一、Light Field / Lumigraph—光场 1.我们看到的是什么 我们的眼睛能够把3D世界转换为2D的成像信号被我们感知&#xff0c;如上面第一幅图&#xff0c;这就是我们看到整个世界的过程&#xff0c;那么如果我们把之前记录的光的信息都完美的放在一个幕布上&#xff0c;那么我们…

第15章 信息(文档)和配置管理

文章目录 软件文档的分类&#xff08;1&#xff09;开发文档&#xff1a;描述开发过程 本身&#xff08;2&#xff09;产品文档&#xff1a;描述开发过程的 产物&#xff08;3&#xff09;管理文档&#xff1a;记录项目管理的信息 文档的质量可以分为四级&#xff08;1&#xf…

第二十五章 刚体Rigidbody

在物理学中&#xff0c;静止和匀速直线运动是物体的平衡状态&#xff0c;如果给该物体施加某一个力的话&#xff0c;物体的平衡状态就会改变&#xff0c;当然这个真理的前提是理想状态。我们知道在现实世界中&#xff0c;由于重力和摩擦力的存在&#xff0c;任何一个物体都不可…

【SQL篇】窗口函数和公共表达式

1077 项目员工 III # Write your MySQL query statement below select project_id, employee_id from (select project_id, e.employee_id, rank() over(partition by project_id order by experience_years desc) as rkfrom Employee ejoin Project pon e.employee_id p.empl…

【Fluent】接着上一次计算的结果继续计算,利用计算过程中得到的物理场(温度、速度、压力等)插值Interpolate文件初始化模型的方法

一、问题背景 因为fluent中支持的初始化无非三种类型。 1、Standard initialization 标准初始化 2、Hybridinitialization 混合初始化 3、FMG initialization FMG初始化 另外&#xff0c;还可以用UDF通过坐标判断的方式予以初始化。 但是这些初始化方法都没办法利用以前计算过…

通关MyBatis(上)

作者&#xff1a;~小明学编程 文章专栏&#xff1a;spring框架 格言&#xff1a;热爱编程的&#xff0c;终将被编程所厚爱。 目录 什么是MyBatis 如何使用Mybatis 添加依赖 创建数据库 配置数据库连接字符串 MyBatis的操作流程 数据持久层 配置mybatis的xml文件 mapp…

Windows自动虚拟机WSL和VMware虚拟机兼容问题(此平台不支持虚拟化的 Intel VT-x/EPT)

问题背景与原因分析 在安装了WSL2之后&#xff0c;忽然发现VMware Workstation无法正常启动了。就是在开启虚拟机时遇到了这种情况&#xff1a; “ 此平台不支持虚拟化的 Intel VT-x/EPT” 问题描述&#xff1a;出现以上问题&#xff0c;发现WSL2和 VMware Workstation 是不兼…

idea使用git遇到的小问题

idea使用git遇到的小问题 前置说明颜色含义中文插件修改提交的用户名 前置说明 idea版本为2022专业版 github需要自己会科学上网 颜色含义 在idea中使用github后&#xff0c;会发现项目中会有各种各样的颜色&#xff0c;如图所示文件全为绿色 这颜色含义分别为&#xff1a;…

函数-函数递归及练习

目录 1、什么是递归&#xff1f; 2、递归的两个必要条件 3、递归的练习 3.1 接受一个整型值&#xff08;无符号&#xff09;&#xff0c;按照顺序打印它的每一位 3.2 编写函数不允许创建临时变量&#xff0c;求字符串的长度 3.3 求第n个斐波那契数 3.4 字符串逆序&…

UG NX二次开发(C++)-建模-修改NXObject或者Feature的颜色(二)

文章目录 1、前言2、在UG NX中修改Body的颜色操作3、采用NXOpen(C)实现3.1 创建修改对象颜色的方法3.2 在do_it()中添加调用的代码3.3 测试效果 1、前言 在UG NX中&#xff0c;改变NXObject和Feature的操作是不相同的&#xff0c;所以其二次开发的代码也不一样&#xff0c;我们…

企业级信息系统开发讲课笔记4.1 Spring Boot入门程序

文章目录 零、学习目标一、Spring Boot框架概述&#xff08;一&#xff09;由Spring到Spring Boot&#xff08;二&#xff09;Spring Boot框架的核心功能&#xff08;三&#xff09;Spring Boot框架的应用 二、使用Maven方式构建Spring Boot项目&#xff08;一&#xff09;创建…

二维字符数组的三种输入方式浅析(scanf()、gets()和fgets())

二维字符数组的输入 目录 二维字符数组的输入1.scanf函数知识点scanf()关于回车的问题&#xff1a; 2.gets函数3.fgets函数参考链接 1.scanf函数 知识点 按照常规输入数组的办法&#xff0c;通过 for 循环实现 将整个字符串输入时&#xff0c;在数组名前不加&&#xff0…

Java 基础进阶篇(二)—— static 静态关键字与单例模式

文章目录 一、static 静态关键字1.1 静态成员变量与实例成员变量1.2 静态成员方法与实例成员方法1.3 static 访问注意事项1.4 内存使用情况 二、工具类三、代码块四、单例模式4.1 饿汉单例4.2 懒汉单例 一、static 静态关键字 static&#xff1a;代表静态的意思&#xff0c;可…

Java 基础进阶篇(六)—— 接口详解

文章目录 一、接口概述二、接口的基本使用三、接口从 JDK 8 开始新增的方法四、接口的注意事项&#xff08;了解&#xff09;补充&#xff1a;接口与接口的关系 一、接口概述 规范的基本特征是约束和公开。 接口就是一种规范&#xff0c;其约束别人必须干什么事情。 所以&…

【五一创作】Matlab 绘制风速、风向统计玫瑰花图【优化】

在之前&#xff0c;有个博客专门讲matlab 绘制风速、风向统计玫瑰花图&#xff1b;这里面存在不少细节问题&#xff0c;目前对该部分代码做了优化。以前的博客链接见下&#xff1a; Matlab 绘制风速、风向统计玫瑰花图 最近接了一个任务&#xff0c;需要绘制风速、风向的统计玫…

Java 基础进阶篇(五)—— 抽象类与模板方法设计模式

文章目录 一、抽象类、抽象方法概述二、抽象类的特征三、模板方法设计模式3.1使用场景3.2 实现步骤3.3 写作文案例 补充&#xff1a;final 和 abstract 是什么关系? 一、抽象类、抽象方法概述 在 Java 中 abstract 是抽象的意思&#xff0c;可以修饰类、成员方法。 abstract …

Java 基础进阶篇(七)—— 面向对象三大特征之三:多态

文章目录 一、多态的概述二、多态中成员访问特点 ★三、多态的优势与劣势四、多态下的类型转换4.2 自动类型转换&#xff08;从子到父&#xff09;4.2 强制类型转换&#xff08;从父到子&#xff09;4.3 instanceof 关键字 一、多态的概述 多态&#xff1a;是指执行同一个行为…

Java 基础进阶篇(四)—— 权限修饰符、final 关键字与枚举

文章目录 一、权限修饰符二、final 关键字2.1 final 作用2.2 final 修饰变量举例2.3 常量 三、枚举3.1 枚举的格式3.2 枚举的特征3.3 枚举的应用 一、权限修饰符 权限修饰符 用于约束成员变量、构造器、方法等的访问范围。 权限修饰符&#xff1a; 有四种作用范围由小到大 (p…