乐趣区

关于java:基于spring环境的eventbus

jiezi

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 是优先级，数值越大越先被生产

退出移动版