事件中心 - 解耦核心业务与辅助功能

/**
 * 事件中心，一般用来监听收集各种事件并分发给监听者列表.
 *
 * @author tenmao
 * @since 2019/12/12
 */
@Slf4j
public class EventHub<Event, Entity> {
    private final MdcThreadPoolExecutor mdcThreadPoolExecutor = MdcThreadPoolExecutor.newWithInheritedMdc(4, 16, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100));

    /**
     * 同步监听（一般可以用在事务一致性场景）.
     */
    private Map<Event, Set<Consumer<Entity>>> syncListenerMap = new HashMap<>();

    /**
     * 异步监听.
     */
    private Map<Event, Set<Consumer<Entity>>> asyncListenerMap = new HashMap<>();


    /**
     * 注册同步监听器.
     *
     * @param event        事件
     * @param syncListener 同步监听器
     */
    public void subscribeSync(Event event, Consumer<Entity> syncListener) {
        Preconditions.checkNotNull(event);
        Preconditions.checkNotNull(syncListener);
        log.info("subscribeSync: event[{}], listener[{}]", event, syncListener);
        addListener(syncListenerMap, event, syncListener);
    }

    /**
     * 注册异步监听器.
     *
     * @param event         事件
     * @param asyncListener 异步监听器
     */
    public void subscribeAsyncListener(Event event, Consumer<Entity> asyncListener) {
        Preconditions.checkNotNull(event);
        Preconditions.checkNotNull(asyncListener);
        log.info("subscribeAsyncListener: event[{}], listener[{}]", event, asyncListener);
        addListener(asyncListenerMap, event, asyncListener);
    }

    private synchronized void addListener(Map<Event, Set<Consumer<Entity>>> map, Event event, Consumer<Entity> listener) {
        Set<Consumer<Entity>> consumers = map.get(event);
        if (consumers == null) {
            consumers = new HashSet<>();
            map.putIfAbsent(event, consumers);
        }
        consumers.add(listener);
    }


    /**
     * 发布事件.
     *
     * @param event  事件
     * @param entity 事件实体
     */
    public void publishEvent(Event event, Entity entity) {
        //订单状态变更
        log.info("publishEvent: event[{}], entity[{}]", event, entity);
        for (Consumer<Entity> consumer : syncListenerMap.get(event)) {
            consumer.accept(entity);
        }

        for (Consumer<Entity> asyncObserver : asyncListenerMap.get(event)) {
            mdcThreadPoolExecutor.execute(() -> {
                try {
                    asyncObserver.accept(entity);
                } catch (Exception e) {
                    log.warn("fail to finish listener for event: event[{}], entity[{}]", event, entity);
                }
            });
        }
    }
}

/**
 * @author tenmao
 * @since 2019/12/12
 */
@Configuration
public class EventHubConfiguration {
    /**
     * 订单状态事件中心.
     */
    @Bean
    public EventHub<OrderStatus, Order> orderEventHub() {
        return new EventHub<>();
    }

    /**
     * 退款状态事件中心.
     */
    @Bean
    public EventHub<RefundStatus, Refund> refundEventHub() {
        return new EventHub<>();
    }
}

/**
 * 发送提醒消息.
 *
 * @author tenmao
 * @since 2019/12/12
 */
@Slf4j
@Component
public class EmailListener {
    @Resource
    private EmailRemoteManager EmailRemoteManager;

    @Resource
    private EventHub<OrderStatus, Order> orderEventHub;

    @PostConstruct
    private void init() {
        //监听多个不同的事件，监听到后会发送提醒消息
        orderEventHub.subscribeAsyncListener(OrderStatus.PAYED, this::handleOrderPayed);
        orderEventHub.subscribeAsyncListener(OrderStatus.ARRIVAL, this::handleOrderArrival);
    }

    private void handleOrderPayed(Order order) {
        //推送 Email
        EmailRequest EmailRequest = EmailRequest.build(order);
        EmailRemoteManager.sendEmail(EmailRequest);
    }

    private void handleOrderArrival(Order order) {
        //推送 Email
        EmailRequest EmailRequest = EmailRequest.build(order);
        EmailRemoteManager.sendEmail(EmailRequest);
    }
}

@Slf4j
@Component
public class OrderManager {
    @Resource
    private EventHub<OrderStatus, Order> orderEventHub;

    public void payOrder(String orderNo) {
        //TODO 订单付款
        Order order = getOne(orderNo);
        orderEventHub.publishEvent(OrderStatus.PAYED, order);
    }

    public void orderArrival(String orderNo) {
        //TODO 订单到货
        Order order = getOne(orderNo);
        orderEventHub.publishEvent(OrderStatus.ARRIVAL, order);
    }
    //TODO 还有其他事件
}