围观：基于事件机制的内部解耦之心路历程

栏目: IT技术 · 发布时间: 4年前

每篇文章都有属于它自己的故事，没有故事的文章是没有灵魂的文章。而我就是这个灵魂摆渡人。

主人公张某某，这边不方便透露姓名，就叫小张吧。小张在一家小型的互联网创业团队中就职。

职位是 Java 后端开发，所以整体和业务代码打交道在所难免。

之前有个搜索相关的需求，而且数量量也算比较大，就采用了 ElasticSearch 来做搜索。第一版由于时间比较赶，做的比较粗糙。越到后面发现代码越难写下去了，主要是在更新索引数据的场景没处理好，才有了今天的故事。

基础入门

Spring Event

Spring 的事件就是观察者设计模式，一个任务结束后需要通知任务进行下一步的操作，就可以使用事件来驱动。

在 Spring 中使用事件机制的步骤如下：

自定义事件对象，继承 ApplicationEvent
自定义事件监听器，实现 ApplicationListener 或者通过 @EventListener 注解到方法上实现监听
自定义发布者，通过 applicationContext.publishEvent()发布事件

Spring Event 在很多开源框架中都有使用案例，比如 Spring Cloud 中的 Eureka 里面就有使用

event 包

定义 Event

发布 Event

Guava EventBus

EventBus 是 Guava 的事件处理机制，在使用层面和 Spring Event 差不多。这里不做过多讲解，今天主要讲 Spring Event。

业务背景

所有的数据都会有一个定时任务去同步数据到 ElasticSearch 中，业务中直接从 ElasticSearch 查询数据返回给调用方。

之所以把所有数据都存入 ElasticSearch 是为了方便，如果只存储搜索的字段，那么搜索出来后就还需要去数据库查询其他信息进行组装。

就是由于所有数据都会存储 ElasticSearch 中，所以当这些数据发生变更的时候我们需要去刷新 ElasticSearch 中的数据，这个就是我们今天文章的核心背景。

假设我们 ElasticSearch 中的数据是文章信息，也就是我们经常看的技术文章，这个文章中存储了访问量，点赞量，评论量等信息。

当这些动作发生的时候，都需要去更新 ElasticSearch 的数据才行，我们默认的操作都是更新数据库中的数据，ElasticSearch 是由定时任务去同步的，同步会有周期，做不到毫秒别更新。

实现方案-倔强青铜

倔强青铜就是在每个会涉及到数据变更的地方，去手动调用代码进行数据的刷新操作，弊端在于每个地方都要去调用，这还是简单的场景，有复杂的业务场景，一个业务操作可能会涉及到很多数据的刷新，也就是需要调用很多次，模拟代码如下：

// 浏览
public void visit() {
    articleIndexService.reIndex(articleId);
    XXXIndexService.reIndex(articleId);
    ........
}

// 评论
public void comment() {
    articleIndexService.reIndex(articleId);
}

实现方案-秩序白银

倔强青铜的弊端在于不解耦，而且是同步调用，如果在事务中会加长事务的时间。所以我们需要一个异步的方案来执行重建索引的逻辑。

经过大家激烈的讨论，而项目也是以 Spring Boot 为主，所以选择了 Spring Event 来作为异步方案。

定义一个重建文章索引的 Event，代码如下：

public class ArticleReIndexEvent extends ApplicationEvent {
  private String id;

  public ArticleReIndexEvent(Object source, String id) {
    super(source);
    this.id = id;
  }

  public String getId() {
    return id;
  }

}

然后写一个 EventListener 来监听事件，进行业务逻辑处理，代码如下：

@Component
public class MyEventListener {

  @EventListener
  public void onEvent(ArticleReIndexEvent event) {
    System.out.println(event.getId());
  }
}

使用的地方只需要发布一个 Event 就可以，这个动作默认是同步的，如果我们想让这个操作不会阻塞，变成异步只需要在@EventListener 上面再增加一个@Async 注解。

// 浏览
public void visit() {
  applicationContext.publishEvent(new ArticleReIndexEvent(this, articleId));
  applicationContext.publishEvent(new XXXReIndexEvent(this, articleId));
}

// 评论
public void comment() {
  applicationContext.publishEvent(new ArticleReIndexEvent(this, articleId));
}

实现方案-荣耀黄金

秩序白银的方案在代码层面确实解耦了，但是使用者发布事件需要关注的点太多了，也就是我改了某个表的数据，我得知道有哪些索引会用到这张表的数据，我得把这些相关的事件都发送出去，这样数据才会异步进行刷新。

当业务复杂后或者有新来的同事，不是那么的了解业务，压根不可能知道说我改了这个数据对其他那些索引有影响，所以这个方案还是有优化的空间。

荣耀黄金的方案是将所有的事件都统一为一个，然后在事件里加属性来区分修改的数据是哪里的。每个数据需要同步变更的索引都有自己的监听器，去监听这个统一的事件，这样对于发布者来说我只需要发送一个事件告诉你，我这边改数据了，你要不要消费，要不要更新索引我并不关心。

定义一个数据表发生修改的事件，代码如下：

public class TableUpdateEvent extends ApplicationEvent {
  private String table;
  private String id;

  public TableUpdateEvent(Object source, String id, String table) {
    super(source);
    this.id = id;
    this.table = table;
  }

  public String getId() {
    return id;
  }

  public String getTable() {
    return table;
  }

}

然后每个索引都需要消费这个事件，只需要关注这个索引中数据的来源表有没有变动，如果有变动则去刷新索引。

比如索引 A 的数据都是 article 表中过来的，所以只要是 article 表中的数据发生了变更，索引 A 都要做对应的处理，所以索引 A 的监听器只需要关注 article 表有没有修改即可。

@Component
public class MyEventListener {

  private List<String> consumerTables = Arrays.asList("article");

  @Async
  @EventListener
  public void onEvent(TableUpdateEvent event) {
    System.out.println(event.getId() + "\t" + event.getTable());
    if (consumerTables.contains(event.getTable())) {
      System.out.println("消费自己关注的表数据变动，然后处理。。。");
    }
  }

}

比如索引 B 的数据是从 comment 和 comment_reply 两个表中过来的，所以只要是 comment 和 comment_reply 两个表的数据发生了变更，索引 B 都需要做对应的处理，所以索引 B 的监听器只需要关注 comment 和 comment_reply 两个表有没有修改即可。

@Component
public class MyEventListener2 {

  private List<String> consumerTables = Arrays.asList("comment", "comment_replay");

  @Async
  @EventListener
  public void onEvent(TableUpdateEvent event) {
    System.out.println(event.getId() + "\t" + event.getTable());
    if (consumerTables.contains(event.getTable())) {
      System.out.println("消费自己关注的表数据变动，然后处理。。。");
    }
  }

}