关于java:基于spring环境的eventbus

eventbus作为事件总线能够提供消费者的注册和事件的散发性能，是实现事件驱动的必要工具也是我的项目中的解耦利器。

本组件是基于greenrobot的eventbus包装，思考到大部分我的项目都是基于spring开发，消费者提供了bean的主动注册。事件发布者也以单例bean的模式创立。

应用示例：https://github.com/neuSnail/s…

疾速开始
maven依赖：

<dependency>
    <groupId>com.github.neusnail</groupId>
    <artifactId>spring-eventbus</artifactId>
    <version>1.0</version>
</dependency>

配置公布器

@Configuration
public class EventBusConfig {
    @Bean
    public EventPublisher eventPublisher() {
        return EventPublisherBuilder.createDefault();
    }
}

创立消费者和事件

@EventSubscriber
@Slf4j
//用户行为事件消费者
public class UserBehaviorSubscriber {
    @Subscribe()
    public void userLeaveSubscriber(UserEntryEvent event) {
        log.info("get userLeaveEvent userName:{}", event.getUserName());
    }
 
    @Subscribe(threadMode = ThreadMode.ASYNC, priority = 1)
    public void userEntrySubscriber(UserLeaveEvent event) {
        log.info("get userEntryEvent userName:{}", event.getUserName());
    }
}

@Data
@EqualsAndHashCode(callSuper = true)
public class UserEntryEvent extends BaseEvent {
    private String userId;
    private String userName;
}

开启消费者扫描

倡议在main办法之后间接调用注册，也能够应用@EventListener、CommandLineRunner、实现aware等形式在容器启动后注册，但留神确保不要反复注册

@SpringBootApplication
public class DemoApplication{
 
    public static void main(String[] args) {
        ApplicationContext context = SpringApplication.run(DemoApplication.class, args);
        SubscriberRegister.register(context);
    }
}

发送事件

public void postEvent() {
    UserEntryEvent userEntryEvent = new UserEntryEvent();
    userEntryEvent.setUserName("bigBoss");
    eventPublisher.post(userEntryEvent);
}

留神这里提供的postAsync是生产方的异步，决定权在生产者这里，threadmode是生产方的异步，决定权在消费者里

参数配置
在注册publisher bean时候能够配置局部参数

@Configuration
public class EventBusConfig {
    @Bean
    public EventPublisher eventPublisher() {
        return EventPublisherBuilder.builder()
                .asyncPostExecutor(Executors.newFixedThreadPool(1))
                .EventBus(EventBus.builder()
                        .throwSubscriberException(false)
                        .sendNoSubscriberEvent(true)
                        .executorService(Executors.newCachedThreadPool())
                        .build()
                ).build();
    }
}

其中asyncPostExecutor是发布者异步发送时应用的线程池，eventbus中的executorService是消费者异步生产的线程池

对于eventbus builder的具体参数文档可参照：https://greenrobot.org/files/…

默认的发送者和消费者线程池都应用了阿里的ttl线程池，防止父子线程传递问题

private ExecutorService defaultExecutor() {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
            4, 8,
            5L, TimeUnit.MINUTES,
            new LinkedBlockingQueue<Runnable>(1000),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );
    return TtlExecutors.getTtlExecutorService(executor);
}

消费者线程模式和优先级

@subscribe中的threadmode有以下选项

ThreadMode.POSTING:和发送事件在同一个线程
ThreadMode.MAIN:主线程
ThreadMode.BACKGROUND:子线程
ThreadMode.ASYNC:异步线程
咱们在javaweb的开发中个别只会应用ThreadMode.POSTING（同步）和ThreadMode.ASYNC（异步）

priority是优先级，数值越大越先被生产