关于java:Halo-开源项目学习六事件监听机制

根本介绍

Halo 我的项目中，当用户或博主执行某些操作时，服务器会公布相应的事件，例如博主登录管理员后盾时公布 “ 日志记录 ” 事件，用户浏览文章时公布 “ 拜访文章 ” 事件。事件公布后，负责监听的 Bean 会做出相应的解决，这种设计称为事件监听机制，其作用是能够实现业务逻辑之间的解耦，进步程序的扩展性和可维护性。

ApplicationEvent 和 Listener

Halo 应用 ApplicationEvent 和 Listener 来实现事件的公布与监听，二者由 Spring 提供，其中 ApplicationEvent 是须要公布的事件，Listener 则是监听器。用户可在监听器中自定义事件的解决逻辑，当事件产生时，只须要将事件公布，监听器会依据用户定义的逻辑主动解决该事件。

定义事件

事件须要继承 ApplicationEvent 类，且须要重载构造方法，以 LogEvent 为例：

public class LogEvent extends ApplicationEvent {

    private final LogParam logParam;

    /**
     * Create a new ApplicationEvent.
     *
     * @param source the object on which the event initially occurred (never {@code null})
     * @param logParam login param
     */
    public LogEvent(Object source, LogParam logParam) {super(source);

        // Validate the log param
        ValidationUtils.validate(logParam);

        // Set ip address
        logParam.setIpAddress(ServletUtils.getRequestIp());

        this.logParam = logParam;
    }

    public LogEvent(Object source, String logKey, LogType logType, String content) {this(source, new LogParam(logKey, logType, content));
    }

    public LogParam getLogParam() {return logParam;}
}

构造方法中的 source 指的是触发事件的 Bean，也称为事件源，通常用 this 关键字代替，其它参数可由用户任意指定。

公布事件

ApplicationContext 接口的 publishEvent 办法可用于公布事件，例如博客初始化实现后公布 LogEvent 事件（InstallConroller 中的 installBlog 办法）：

public BaseResponse<String> installBlog(@RequestBody InstallParam installParam) {
    // 省略局部代码

    eventPublisher.publishEvent(new LogEvent(this, user.getId().toString(), LogType.BLOG_INITIALIZED, "博客已胜利初始化")
    );

    return BaseResponse.ok("装置实现！");
}

监听器

监听器的创立形式有多种，例如实现 ApplicationListener 接口、SmartApplicationListener 接口，或者增加 @EventListener 注解。我的项目中应用注解来定义监听器，如 LogEventListener：

@Component
public class LogEventListener {

    private final LogService logService;

    public LogEventListener(LogService logService) {this.logService = logService;}

    @EventListener
    @Async
    public void onApplicationEvent(LogEvent event) {
        // Convert to log
        Log logToCreate = event.getLogParam().convertTo();

        // Create log
        logService.create(logToCreate);
    }
}

用户可在 @EventListener 注解润饰的办法中定义事件的解决逻辑，办法接管的参数为监听的事件类型。@Async 注解的作用是实现异步监听，以上文中的 installBlog 办法为例，如果不增加该注解，那么程序须要期待 onApplicationEvent 办法执行完结后能力返回 “ 装置实现！”。加上 @Async 注解后，onApplicationEvent 办法会在新的线程中执行，installBlog 办法能够立刻返回。若要使 @Async 注解失效，还须要在启动类或配置类上增加 @EnableAsync 注解。

事件处理

接下来咱们剖析一下 Halo 我的项目中不同事件的处理过程：

日志记录事件

日志记录事件 LogEvent 由 LogEventListener 中的 onApplicationEvent 办法解决，该办法的解决逻辑非常简单，就是在 logs 表中插入一条系统日志，插入的记录用于在管理员界面展现：

须要留神的是，不同类型日志的 logKey、logType 以及 content 会有所区别，例如用户登录时，logKey 为用户的 userName，logType 为 LogType.LOGGED_IN，content 为用户的 nickName：

eventPublisher.publishEvent(new LogEvent(this, user.getUsername(), LogType.LOGGED_IN, user.getNickname()));

公布文章时，logKey 为文章的 id，logType 为 LogType.POST_PUBLISHED，content 为文章的 title：

LogEvent logEvent = new LogEvent(this, createdPost.getId().toString(),
    LogType.POST_PUBLISHED, createdPost.getTitle());
eventPublisher.publishEvent(logEvent);

文章拜访事件

文章拜访事件 PostVisitEvent 由 AbstractVisitEventListener 中的 handleVisitEvent 办法解决，该办法的解决的逻辑是将以后文章的访问量加一：

protected void handleVisitEvent(@NonNull AbstractVisitEvent event) throws InterruptedException {Assert.notNull(event, "Visit event must not be null");
    // 获取文章 id
    // Get post id
    Integer id = event.getId();

    log.debug("Received a visit event, post id: [{}]", id);

    // 如果以后 postId 具备对应的 BlockingQueue, 那么间接返回该 BlockingQueue, 否则为以后 postId 创立一个新的 BlockingQueue
    // Get post visit queue
    BlockingQueue<Integer> postVisitQueue =
        visitQueueMap.computeIfAbsent(id, this::createEmptyQueue);
    // 如果以后 postId 具备对应的 PostVisitTask, 不做任何解决, 否则为以后 postId 创立一个新的 PostVisitTask 工作
    visitTaskMap.computeIfAbsent(id, this::createPostVisitTask);
    // 将以后 postId 存入到对应的 BlockingQueue
    // Put a visit for the post
    postVisitQueue.put(id);
}

上述办法首先获取以后被拜访文章的 postId，而后查问 visitQueueMap 中是否存在 postId 对应的阻塞队列（理论类型为 LinkedBlockingQueue），如果存在那么间接返回该队列, 否则为以后 postId 创立一个新的阻塞队列并存入到 visitQueueMap。接着查问 visitTaskMap 中是否存在 postId 对应的 PostVisitTask 工作（工作的作用是将文章的访问量加一），如果没有，那么就为 postId 创立一个新的 PostVisitTask 工作，并将该工作交给线程池 ThreadPoolExecutor（Executors.newCachedThreadPool()）执行。之后将 postId 增加到对应的阻塞队列，这一步的目标是治理 PostVisitTask 工作的执行次数。

visitQueueMap 和 visitTaskMap 都是 ConcurrentHashMap 类型的对象，应用 ConcurrentHashMap 是为了保障线程平安，因为监听器的事件处理办法被 @Async 注解润饰。默认状况下，@Async 注解润饰的办法会由 Spring 创立的线程池 ThreadPoolTaskExecutor 中的线程执行，因而当某一篇文章被多个用户同时浏览时，ThreadPoolTaskExecutor 中的多个线程可能会同时在 visitQueueMap 中创立阻塞队列，或在 visitTaskMap 中创立 PostVisitTask 工作。

上面看一下 PostVisitTask 工作中 run 办法的解决逻辑：

public void run() {while (!Thread.currentThread().isInterrupted()) {
        try {BlockingQueue<Integer> postVisitQueue = visitQueueMap.get(id);
            Integer postId = postVisitQueue.take();

            log.debug("Took a new visit for post id: [{}]", postId);

            // Increase the visit
            basePostService.increaseVisit(postId);

            log.debug("Increased visits for post id: [{}]", postId);
        } catch (InterruptedException e) {
            log.debug("Post visit task:" + Thread.currentThread().getName() + "was interrupted",
                e);
            // Ignore this exception
        }
    }

    log.debug("Thread: [{}] has been interrupted", Thread.currentThread().getName());
}

线程池 ThreadPoolExecutor 中的一个线程解决该工作：

1. 从 visitQueueMap 获取 postId 对应的阻塞队列（这里的 id 其实就是 postId），并取出队首元素。
2. 将 postId 对应的文章的点赞量加一。
3. 只有线程不被中断，就始终反复步骤 1 和步骤 2，如果队列为空，那么线程进入阻塞。

综上，文章拜访事件的解决流程总结如下：

当 id 为 postId 的文章被拜访时，零碎会为其创立一个 LinkedBlockingQueue 类型的阻塞队列和一个负责将文章点赞量加一的 PostVisitTask 工作。而后 postId 入队，线程池 ThreadPoolExecutor 调配一个线程执行 PostVisitTask 工作，阻塞队列有多少个 postId 该工作就执行多少次。

结语

事件监听机制是一个十分重要的知识点，理论开发中，如果某些业务解决起来比拟耗时，且与次要业务的关联性并不是很强，那么能够思考做工作拆分，利用事件监听机制将串行执行异步化，改为并行执行（当然也能够应用音讯队列）。Halo 中还有新增评论、主题更新等事件，这些事件的的解决思路与文章拜访事件类似，所以本文就不再过多陈说了 (⊙‿⊙)。