AsyncEventBus code example

EventCenter

封装AsyncEventBus，对外有register 和 post 方法。

@Service
public class EventCenter {
    private static final Logger LOG = LoggerFactory.getLogger(EventCenter.class);

    private static AsyncEventBus INSTANCE;
    private static ThreadPoolExecutor EXECUTOR;

    @PostConstruct
    private void init() {
        //default new ThreadPoolExecutor.AbortPolicy()
        this.EXECUTOR = new ThreadPoolExecutor(10, 50, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000));

        this.INSTANCE = new AsyncEventBus(EXECUTOR, new SubscriberExceptionHandler() {
            @Override
            public void handleException(Throwable exception, SubscriberExceptionContext context) {
                LOG.error("consume event exception : [subscribeMethod={}, event={} ]",context.getSubscriberMethod(), context.getEvent().toString(),exception);
            }
        });
    }

    private AsyncEventBus getInstance() {
        if(INSTANCE == null) {
            init();
        }
        return INSTANCE;
    }

    public void register(Object obj) {
        LOG.info("register listener {}", obj.getClass().getSimpleName());
        getInstance().register(obj);
    }

    public void post(Object event){
        LOG.info("post event {}", event.toString());
        getInstance().post(event);
    }
}

Listener

AbstractEventListener 为所有listener基类，init中注册到EventBus.
BizEventListener为业务类，消费EventBus发出的消息。

@Service
public abstract class EventListener {

    @Autowired
    private EventCenter eventCenter;

    @PostConstruct
    public void init() {
        eventCenter.register(this);
    }
}


@Service
public class BizEventListener extends AbstractEventListener{

    @AllowConcurrentEvents
    @Subscribe
    public void consumeTaskEvent(Event event) {
        //business code
    }
}