事件驅(qū)動模型的初次使用

個人理解：事件驅(qū)動，顧名思義坠七，當(dāng)某一個事件發(fā)生時水醋，會有其他監(jiān)聽該事件發(fā)生的方法伴隨著該事件的執(zhí)行而觸發(fā)旗笔。因此，事件驅(qū)動中的組成就很明確了拄踪，一個事件蝇恶、一個調(diào)用事件的事件源、多個監(jiān)聽者惶桐。

案例：在我的工作中撮弧，有這樣的需求，一個游戲服務(wù)器需要對玩家的數(shù)據(jù)進行持久化姚糊，但是玩家的數(shù)據(jù)又會關(guān)聯(lián)到很多其他的實體贿衍，例如玩家的背包、寵物等等叛拷。如果將所有的保存邏輯都放在玩家數(shù)據(jù)保存的代碼塊中舌厨，那么每當(dāng)我新增一個關(guān)聯(lián)玩家的實體時岂却，就要添加新的代碼進來忿薇，這樣對于一個分模塊的游戲服務(wù)器來說是很不友好的。因此躏哩，我打算將保存玩家數(shù)據(jù)作為一個事件署浩，在我完成對玩家數(shù)據(jù)的保存操作以后就調(diào)用該事件，讓監(jiān)聽該事件的其他方法隨之觸發(fā)扫尺，例如保存背包和保存寵物等筋栋。這樣，只需要把這些方法注冊為保存玩家數(shù)據(jù)事件的監(jiān)聽者正驻，就能避免在玩家數(shù)據(jù)保存的代碼塊中不斷的新增其余的邏輯了弊攘。

代碼如下：我這是參考了Google的EventBus事件總線，自己寫了一個簡化版的事件驅(qū)動姑曙。

/**
 * 事件抽象類襟交，代表基礎(chǔ)事件
 *
 * @author Administrator
 */
public interface BaseEvent {

    /**
     * 得到事件的擁有者
     * @return
     */
    default Object getOwner() {
        return "system";
    }

}

/**
 * 利用Spring管理Bean，來對bean進行掃描伤靠，注冊監(jiān)聽者捣域。
 *
 * @author Administrator
 */
@Component("eventRegisterBeanProcessor")
public class EventRegisterBeanProcessor implements BeanPostProcessor {

    @Resource
    EventBus eventBus;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        // 注冊bean中存在@Subscribe注解的方法（注冊消息訂閱）
        eventBus.register(bean);
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

}

/**
 * 消息注冊器，用于尋找符合的Bean宴合，此處為尋找添加了@Subscribe注解的方法焕梅，并注冊進入集合當(dāng)中。
 * 具體的流程在代碼中有注釋說明卦洽。
 *
 * @author Administrator
 */
public class SubscriberRegistry {

    /**
     * 事件類型贞言，**所有** 消息訂閱者 的集合
     */
    private final Map<Class<? extends BaseEvent>, CopyOnWriteArraySet<Subscriber>> eventSubscribersMap = new ConcurrentHashMap<>();





    /**
     * 注冊bean中存在@Subscribe注解的方法（注冊消息訂閱）
     *
     * @param bean listener
     */
    public void register(Object bean) {
        Map<Class<? extends BaseEvent>, Subscriber> eventSubscriberMap = getAllEventSubscriber(bean);
        for (Entry<Class<? extends BaseEvent>, Subscriber> subscriberEntry : eventSubscriberMap.entrySet()) {
            Class<? extends BaseEvent> event = subscriberEntry.getKey();
            eventSubscribersMap.putIfAbsent(event, new CopyOnWriteArraySet<>());
            CopyOnWriteArraySet<Subscriber> subscriberSet = eventSubscribersMap.get(event);
            subscriberSet.add(subscriberEntry.getValue());
        }
    }

    // 注意：bean（也就是一個listener）中有method 和 @Subscribe注解 和 event事件參數(shù)
    // 1.遍歷bean中的method，找到有@Subscribe注解的method阀蒂，得到method當(dāng)中的event事件參數(shù)的 類型
    // 2.保存eventSubscribeMap =  map<event, new subscriber(bean, method)>
    // 3.遍歷eventSubscribeMap的event该窗，為每個event創(chuàng)建一個CopyOnWriteArraySet（如果未創(chuàng)建）
    // 4.將eventSubscribeMap的value保存進CopyOnWriteArraySet

    private Map<Class<? extends BaseEvent>, Subscriber> getAllEventSubscriber(Object bean) {
        Map<Class<? extends BaseEvent>, Subscriber> eventSubscribeMap = new HashMap<>();
        Method[] declaredMethods = bean.getClass().getDeclaredMethods();
        for (Method method : declaredMethods) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            // 如果方法method上有@Subscribe注解打肝，即消息處理
            if (method.isAnnotationPresent(Subscribe.class)) {
                if (parameterTypes.length != 1) {
                    throw new IllegalArgumentException();
                }
                // 如果參數(shù)是event類型
                if (BaseEvent.class.isAssignableFrom(parameterTypes[0])) {
                    // 得到event類型
                    Class<? extends BaseEvent> event = (Class<? extends BaseEvent>) parameterTypes[0];
                    // 添加進map
                    eventSubscribeMap.put(event, new Subscriber(bean, method));
                }
            }
        }
        return eventSubscribeMap;
    }

    /**
     * 根據(jù)event類型找到所有的消息執(zhí)行方法
     * @param eventType
     * @return
     */
    public Set<Subscriber> getSubscribersByEvent(Class<? extends BaseEvent> eventType) {
        if (!eventSubscribersMap.containsKey(eventType)) {
            return Collections.emptySet();
        }
        return eventSubscribersMap.get(eventType);
    }

}

/**
 * @Subscribe注解，即文中提到的監(jiān)聽者挪捕，注解的方法會伴隨著事件的執(zhí)行而觸發(fā)粗梭。
 * 訂閱消息（行為）
 * 一個被標(biāo)注了@Subscribe注解的方法就是一個訂閱者。
 *
 * @author Administrator
 */
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface Subscribe {

}

/**
 * 消息訂閱者（類方法）
 * 實際上就是監(jiān)聽者的類级零。
 * 可以這樣理解断医，監(jiān)聽者所對應(yīng)的對象。
 * 即奏纪，method對應(yīng)標(biāo)注了@Subscribe注解的方法鉴嗤。
 * listener對應(yīng)其所在的類。
 * 舉個例子：Test類中有一個test()方法序调，test()方法是被標(biāo)注了@Subscribe注解的醉锅。
 * 那么，此處就會有一個Subscriber對象发绢，listener就是Test類硬耍；method就是test()方法。
 *
 * @author Administrator
 */
public class Subscriber {

    /**
     * 消息監(jiān)聽器
     */
    private Object listener;

    /**
     * 消息執(zhí)行者
     */
    private Method method;

    public Subscriber(Object listener, Method method) {
        this.listener = listener;
        this.method = method;
    }

    /**
     * 執(zhí)行消息
     * @param event 事件
     */
    public void handleEvent(BaseEvent event) {
        try {
            method.invoke(listener, event);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }
}

/**
 * 事件總線边酒，用于注冊監(jiān)聽者以及處理事件经柴。
 *
 * @author Administrator
 */
@Slf4j
@Component("eventBus")
public class EventBus {


    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 初始化事件總線線程池
     */
    public void initEventBusPool() {
        ThreadFactory businessThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("EventBus thread-%d")
                .setUncaughtExceptionHandler((thread, exception) -> exception.printStackTrace())
                .build();
        // 創(chuàng)建線程數(shù)量為1、無界任務(wù)隊列的線程池墩朦。
        threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), businessThreadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
    }

    private SubscriberRegistry registry = new SubscriberRegistry();

    /**
     * 注冊的目的是坯认，通過事件的類型，找到 所有 對應(yīng)的subscriber（跟事件關(guān)聯(lián)的方法）
     * @param bean
     */
    public void register(Object bean) {
        registry.register(bean);
    }


    /**
     * 處理消息
     * 當(dāng)有事件發(fā)布到事件總線中氓涣，事件總線遍歷所有的訂閱者進行事件處理
     *
     * @param event
     */
    public void post(BaseEvent event) {
        // 得到事件類型
        Class<? extends BaseEvent> eventType = event.getClass();
        // 通過事件類型找到所有要處理的消息（事件訂閱者）
        Set<Subscriber> subscribers = registry.getSubscribersByEvent(eventType);
        for (Subscriber subscriber : subscribers) {
            // 處理事件
            subscriber.handleEvent(event);
        }
    }

    /**
     * 異步處理事件
     * @param event
     */
    public void asyncPost(BaseEvent event) {
        this.threadPoolExecutor.execute(() -> post(event));
    }
}

/**
 * 一個測試Subscriber
 * @author Administrator
 */
@Slf4j
public class Test {

    /**
    * 測試案例中的保存玩家數(shù)據(jù)
    */
    @Subscribe
    public void testPlayerSaveEvent(PlayerSaveEvent playerSaveEvent) {
        Player p = playerSaveEvent.getOwner();
        log.info("=========" + p.getName() + "=========");
    }

}

/**
 * 角色數(shù)據(jù)保存事件
 *
 * @author Administrator
 */
public class PlayerSaveEvent implements BaseEvent {

    private Player player;

    public PlayerSaveEvent(Player player) {
        this.player = player;
    }

    @Override
    public Player getOwner() {
        return player;
    }
}

/**
* 玩家管理類牛哺，保存玩家數(shù)據(jù)，并調(diào)用對應(yīng)的保存事件
*/
public TestSavePlayer {
    
    /**
     * 保存角色數(shù)據(jù)
     * 不管角色增加了什么劳吠，都不影響引润，不用修改這里的代碼。
     * 保存角色是一個事件赴背，用監(jiān)聽器監(jiān)聽保存角色椰拒，一旦監(jiān)聽成功，驅(qū)動保存背包凰荚、保存寵物等事件一起執(zhí)行
     *
     * @param playerId
     */
    public void savePlayer(Long playerId) {
        // 通過playerId得到Player
        Player player = playerCache.getPlayerByPlayerId(playerId);
        if (player != null) {
            playerDao.save(dbPlayer);
            // 處理玩家數(shù)據(jù)保存事件
            eventBus.post(new PlayerSaveEvent(player));

        }
    }
}

代碼就是這樣燃观，一旦保存了玩家數(shù)據(jù)，即調(diào)用了TestSavePlayer中的savePlayer()方法時便瑟，Test中的testPlayerSaveEvent()方法就會隨之觸發(fā)缆毁。