Reactor-EventBus事件驱动开发阻塞问题解决

问题发现

近期，公司一个项目中使用到了Reator-EventBus，网上有关资料比较少，没有完全熟悉、了解工作机制，导致在最后项目现网上出现了一个耗时任务阻塞导致其他任务停滞的严重问题。

使用Spring Reactor-EventBus进行事件驱动开发

使用reactor EventBus首先需要项目是在SpringBoot的基础上搭建的。
然后导入需要的依赖：

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
            <version>2.0.8.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-context</artifactId>
            <version>2.0.7.RELEASE</version>
        </dependency>

按照网上的一些教程，需要写一个类对EventBus进行配置，写一个Listener用来处理消息的事务，使用eventBus.on方法对进行消息和消息处理类绑定。
具体步骤如下：
①创建消息处理类（消息监听器），该类主要用于处理对应消息事件

@Component
public class MyEventListener implements Consumer<Event<MyEvent>> {
    private static final Logger LOGGER = LoggerFactory.getLogger(MyEventListener.class);

    @Override
    public void accept(Event<MyEvent> eventContextEvent) {

        MyEvent event = eventContextEvent.getData();

        LOGGER.info("thread {} ,receive event:{}",Thread.currentThread().getName(),event.getData());
    }
}

②将“消息“和”消息处理类“绑定

@Configuration
public class EventConfig {
    
    @Autowired
    public MyEventListener myEventListener;

    @Bean
    Environment environment() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    @Autowired
    public EventBus eventBus(Environment environment, MyEventListener myEventListener) {
        EventBus eventBus = EventBus.create(environment, Environment.THREAD_POOL);
        //此处通过eventBus.on方法对“myevent”消息和其消息处理类myEventListener进行绑定
        eventBus.on($("myevent"), myEventListener);

        return eventBus;
    }
}

③消息通知类通知消息进行处理

@Autowired
    EventBus eventBus;

    public void publishEvent(String data){
        eventBus.notify("myevent", Event.wrap(new MyEvent(data)));
    }

在实际开发中，可以在消息处理类的处理方法上直接使用@Selector(value = "myevent")注解，减少消息处理类和消息绑定的流程。
上面的例子中，改造后，不再需要EventConfig类，改造的消息处理类如下：

@Consumer
public class EventMsgHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventMsgHandler.class);
        
    //必须注入eventBus
    @Autowired
    private EventBus eventBus;
    
    //@Selector注解的value属性填写消息名称
    @Selector(value = "myevent")
    public void processMrPrepare5Mi(Event ev) {
          //ev.getData()方法用于获取传入参数
          LOGGER.info("thread {} ,receive event:{}",Thread.currentThread().getName(),ev.getData());
    }
}

程序会根据@Seclector注解中的value字段自动找到处理类
同时可以通过Event类型的参数获取通知消息进行处理时传入的data

EventBus的阻塞问题

在实际开发过程中，我们发现，不同的事件使用的是相同的EventBus，从而导致一个消息事件处理的过程中，其他类型的消息事件将会被阻塞。
因此，考虑为不同的消息类型注入不同的EventBus。

使用@Configuration注解为EventBus写一个配置类
@Primary注解表示同一种类型的Bean存在多个时，若未特别注明使用哪个，则使用@Primary注解加注的Bean进行注入

@Configuration
public class EventConfig {

    @Bean(name="eventBus")
    @Primary
    EventBus createEventBus(Environment env) {  
        return EventBus.create(env, Environment.THREAD_POOL);  
    }
    
    @Bean(name="eventBus1")
    EventBus createEventBus1(Environment env) {  
        return EventBus.create(new MpscDispatcher("eventBus1Dispatcher"));  
    } 
    
    @Bean(name="eventBus2")
    EventBus createEventBus2(Environment env) {  
        return EventBus.create(new MpscDispatcher("eventBus2Dispatcher"));  
    } 

    ………………
}

在上面的代码中，我们定义了3个EventBus(当然还可以定义更多)，每个EventBus有自己的名字。每个EventBus的创建方法中使用EventBus.create()方法传入一个Dispatcher。
这里我们使用的是间接实现reactor.core.Dispatcher接口的MpscDispatcher类，这个类的好处是可以对同名的EventBus标注的消息实现方法进行阻塞，对不同名的不阻塞，这也是我们项目工程中所需要的。
当然，reactor.core.Dispatcher接口的实现类和间接实现类还有很多，每个都有不同的效果，甚至我们还可以自己定义Dispatcher，在这里就不再赘述。

在创建完EventBus之后，对消息处理类和消息通知类的改造如下：
消息处理类：

@Consumer
public class EventMsgHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventMsgHandler.class);
        
    //必须注入eventBus
    @Autowired
    @Qualifier("eventBus1")
    private EventBus eventBus;
    
    //@Selector注解的value属性填写消息名称
    @Selector(value = "myevent")
    public void processMrPrepare5Mi(Event ev) {
          //ev.getData()方法用于获取传入阐述
          LOGGER.info("thread {} ,receive event:{}",Thread.currentThread().getName(),ev.getData());
    }
}

消息通知类：

    @Autowired
    @Qualifier("eventBus1")
    EventBus eventBus;

    public void publishEvent(String data){
        eventBus.notify("myevent", Event.wrap(new MyEvent(data)));
    }

我们可以通过@Qualifier注解，对不同的业务流程注入并使用不同的eventBus（同一流程的消息处理类和消息通知类必须使用相同的EventBus），从而避免因为EventBus造成的阻塞。

总结

在使用Reactor-EventBus进行事件驱动开发时，需要定义三个类：①消息处理类；②消息绑定类；③消息通知类。在业务流程中，消息通知类触发消息通知，并根据消息在消息绑定类中绑定的消息处理类提交到对应的消息处理类进行处理。
实际开发过程中，可以使用@Selector注解代替消息绑定类。
不同的业务流程默认使用相同的EventBus，这样是会造成不同流程之间相互阻塞的。因此需要通过@Configuration注解创建不同的EventBus，并在消息处理类和消息通知类中注入，这样可以避免相互阻塞问题。