「一招制敌」老板再也不用为“搜索不到数据”而操心了

2022-11-07 09:04:57来源：今日头条

1. 概览

相信负责过“搜索服务”的伙伴，最害怕的一句话就是：“数据怎么又搜索不出来了！！！”。每当收到这句话，都会心中一颤，因为面对几千万甚至几亿的索引数据，我真的无从下手，不知道业务要搜索什么，也不知道是哪些数据出了问题….

1.1. 背景

目前，“搜索”已经成为后端管理平台的必备功能，在这个业务场景中，很多人都会基于 elasticsearch 强大的检索能力构建自己的搜索服务。但实际开发中，elasticsearch 的引入是非常小的一部分，往往大头是索引模型的数据管理，在整个过程中，我们

需要根据业务需求构建检索模型和ES存储模型；需要从多个数据源中获取数据，并填充到检索模型；需要关注所有数据源的数据变化，并对变更数据进行索引重建；需要对不一致的数据进行识别和处理…

如此繁琐的事情，哪一环出现问题都会收到业务的投诉。

(资料图片仅供参考)

1.2. 目标

对搜索场景中的最佳实践进行封装，从而：

降低开发成本，开发人员将精力放在模型构建上，抛开繁琐的技术细节；对数据索引、关联数据更新有很好的支持；引入数据实时巡检能力，对于数据不一致的情况进行自动修复；引入天级对账机制，保障数据的一致性；2. 快速入门2.1. 准备环境

首先，增加对 spring data elasticsearch 的支持，具体 maven 坐标如下：

    org.springframework.boot    spring-boot-starter-data-elasticsearch

在 application.yml 中添加 es 的配置信息，具体如下：

spring:  elasticsearch:    uris: http://localhost:9200    connection-timeout: 10s    socket-timeout: 30s

新建 SpringESConfiguration 配置信息，指定 ES Repository 的包信息，居然如下：

@Configuration@EnableElasticsearchRepositories(basePackages = "com.geekhalo.lego.wide.es")public class SpringESConfiguration {}

最后，引入 lego-starter，具体如下：

com.geekhalo.legolego-starter0.1.14-wide-SNAPSHOT

至此，就完成了项目的准备工具，可以着手构建索引模型。

2.2. 构建模型

构造模型之前，需要构建一个 Enum 用以管理模型中所有关联数据，具体如下：

public enum WideOrderType implements WideItemType {    ORDER, // 订单主数据    USER, // 用户数据    ADDRESS, // 用户地址数据    PRODUCT // 购买商品数据}

WideOrderType 枚举实现 WideItemType 接口，用于与框架进行集成。

接下来，构建待索引的宽表模型，具体如下：

@Data@NoArgsConstructor@AllArgsConstructor@Document(indexName = "wide_order")public class WideOrder extends BindFromBasedWide {    @org.springframework.data.annotation.Id    private Long id;    @BindFrom(sourceClass = Order.class, field = "userId")    private Long userId;    @BindFrom(sourceClass = Order.class, field = "addressId")    private Long addressId;    @BindFrom(sourceClass = Order.class, field = "productId")    private Long productId;    @BindFrom(sourceClass = Order.class, field = "descr")    private String orderDescr;    @BindFrom(sourceClass = User.class, field = "name")    private String userName;    @BindFrom(sourceClass = Address.class, field = "detail")    private String addressDetail;    @BindFrom(sourceClass = Product.class, field = "name")    private String productName;    @BindFrom(sourceClass = Product.class, field = "price")    private Integer productPrice;    public WideOrder(Long orderId){        setId(orderId);    }    @Override    public Long getId(){        return id;    }    @Override    public boolean isValidate(){        return userId != null && addressId != null && productId != null;    }    @Override    public List getItemsKeyByType(WideOrderType wideOrderType){        switch (wideOrderType){            case ORDER:                return Collections.singletonList(new WideItemKey(wideOrderType, getId()));            case USER:                return Collections.singletonList(new WideItemKey(wideOrderType, getUserId()));            case ADDRESS:                return Collections.singletonList(new WideItemKey(wideOrderType, getAddressId()));            case PRODUCT:                return Collections.singletonList(new WideItemKey(wideOrderType, getProductId()));        }        return Collections.emptyList();    }}

该模型有如下几个特点：

存在很多属性，是由多个表数据共同组成的“宽表”；除 id 属性外，其他属性上都有 @BindFrom 注解，用于标明该字段的数据是来自于哪个实体的那个字段；继承自 BindFromBasedWide，其中 Long 为模型主键，WideOrderType 为刚建的枚举，BindFromBasedWide 将根据字段上的 @BindFrom 注解自动完成数据更新和数据比对；Long getId() 方法返回模型的主键信息;boolean isValidate() 用于对数据的有效性进行验证，无效数据将不会进行持久化处理List getItemsKeyByType(WideOrderType wideOrderType) 根据关联数据类型（WideOrderType）返回不同键信息，以进行数据组装；

至此，模型就建立完毕。

2.3. 数据提供器

有了模型后，我们需要构建一些组件用于为“宽表”提供数据，这就是 WideItemDataProvider 体系。

我们以 OrderProvider 为例，具体如下：

@Component@org.springframework.core.annotation.Order(value = Ordered.HIGHEST_PRECEDENCE)public class OrderProvider implements WideItemDataProvider {    @Autowired    private OrderDao orderDao;    @Override    public List apply(List key){        return orderDao.findAllById(key);    }    @Override    public WideOrderType getSupportType(){        return WideOrderType.ORDER;    }}

该类有如下特点：

实现 WideItemDataProvider 接口，其中 WideOrderType 为刚刚定义的枚举，Long 为 Order 模型的关联键类型，Order 为要提供的数据；List apply(List key)，根据 key 获得对应的数据；WideOrderType getSupportType()，获取该组件所支持的关联类型；@Component 标记该类为 Spring 的托管 Bean；@Order(value = Ordered.HIGHEST_PRECEDENCE) 指定组件的顺序，由于为 WideOrder 提供主数据，优先级调到最高；

每一类关联数据都会提供自己的数据提供器，简单看下 UserProvider 实现，具体如下：

@Componentpublic class UserProvider implements WideItemDataProvider {    @Autowired    private UserDao userDao;    @Override    public List apply(List key){        return userDao.findAllById(key);    }    @Override    public WideOrderType getSupportType(){        return WideOrderType.USER;    }}

和 OrderProvider 没有本质区别，当然，demo 中还提供了多种实现，如：

OrderProvider，提供订单主数据；UserProvider，提供用户信息；AddressProvider，提供用户地址信息；ProductProvider，提供商品信息；2.4. 构建宽表仓库

数据都准备好了，需要将 “宽表” 进行持久化，将其放入最合适的存储引擎，以便更好的处理查询请求。

基于 ElasticsearchRepository 的 WideOrderRepository 具体如下：

@Repositorypublic class WideOrderRepository implements WideCommandRepository {    @Autowired    private WideOrderESDao wideOrderDao;    @Override    public void save(List wides){        wideOrderDao.saveAll(wides);    }    @Override    public List findByIds(List masterIds){        return Lists.newArrayList(wideOrderDao.findAllById(masterIds));    }    @Override    public  void consumeByItem(WideOrderType wideOrderType, KEY key, Consumer wideConsumer){        switch (wideOrderType){            case PRODUCT:                this.wideOrderDao.findByProductId((Long) key).forEach(wideConsumer);            case ADDRESS:                this.wideOrderDao.findByAddressId((Long) key).forEach(wideConsumer);            case ORDER:                this.wideOrderDao.findById((Long) key).ifPresent(wideConsumer);            case USER:                this.wideOrderDao.findByUserId((Long) key).forEach(wideConsumer);        }    }    @Override    public boolean supportUpdateFor(WideOrderType wideOrderType){        return false;    }    @Override    public  void updateByItem(WideOrderType wideOrderType, KEY key, Consumer wideConsumer){        Consumer updateAndSave = wideConsumer.andThen(wideOrder -> wideOrderDao.save(wideOrder));        switch (wideOrderType){            case PRODUCT:                this.wideOrderDao.findByProductId((Long) key).forEach(updateAndSave);            case ADDRESS:                this.wideOrderDao.findByAddressId((Long) key).forEach(updateAndSave);            case ORDER:                this.wideOrderDao.findById((Long) key).ifPresent(updateAndSave);            case USER:                this.wideOrderDao.findByUserId((Long) key).forEach(updateAndSave);        }    }    @Override    public  void updateByItem(WideOrderType wideOrderType, KEY key, WideItemData item){    }}

仓库具有如下特征：

实现 WideCommandRepository 接口，其中 Long 是模型主键（也是宽表主键），WideOrderType 是之前定义的枚举，WideOrder 是宽表；void save(List wides) 提供批量保存方法；List findByIds(List masterIds) 提供根据主键批量查询方法；void consumeByItem(WideOrderType wideOrderType, KEY key, Consumer wideConsumer)，该方法主要用于数据巡检，根据类型和键信息从底层引擎中获取数据，并进行部分比对，用于发现数据不一致情况；boolean supportUpdateFor(WideOrderType wideOrderType)，该实现用于判断是否支持特定类型的批量更新，及依赖引擎能力批量对数据进行更新操作；void updateByItem(WideOrderType wideOrderType, KEY key, WideItemData item)，supportUpdateFor 返回为 true 时，调用该方法，使用引擎的更新能力批量对数据进行更新；void updateByItem(WideOrderType wideOrderType, KEY key, Consumer wideConsumer)，supportUpdateFor 返回为 false 时，调用该方法，根据类型和键信息依次查询所有数据，在内存中完成更新，并写回存储引擎；

所依赖的 WideOrderESDao 基于 ElasticsearchRepository 实现，具体如下：

public interface WideOrderESDao extends ElasticsearchRepository {    List findByProductId(Long productId);    List findByAddressId(Long addressId);    List findByUserId(Long userId);}

2.5. 配置&整合

所有组件都已准备好，现在需要将他们整合在一起。

@Configurationpublic class WideOrderConfiguration extends WideConfigurationSupport {    @Autowired    private WideOrderRepository wideOrderRepository;    @Autowired    private List>> wideItemDataProviders;    @Bean    public WideIndexService createWideIndexService(){        return super.createWideIndexService();    }    @Bean    public WideOrderPatrolService wideOrderPatrolService(){        return new WideOrderPatrolService(createWidePatrolService());    }    @Bean    protected WideService createWideService(            WideIndexService wideIndexService,            WideOrderPatrolService wideOrderPatrolService){        return super.createWideService(wideIndexService, wideOrderPatrolService);    }    @Override    protected WideFactory getWideFactory() {        return WideOrder::new;    }    @Override    protected WideCommandRepository getWideCommandRepository() {        return this.wideOrderRepository;    }    @Override    protected List>> getWideItemProviders() {        return this.wideItemDataProviders;    }}

WideOrderConfiguration 具有如下特点：

继承自 WideConfigurationSupport，父类中存在大量的 createXXX 方法，可以大幅简单代码量；使用 WideOrderRepository 作为宽表的仓库；直接使用 Spring 容器中的所有 WideItemDataProvider 实现；使用定制的 WideOrderPatrolService，为巡检增加延时支持；

其中自定义巡检 WideOrderPatrolService 代码如下：

public class WideOrderPatrolService implements WidePatrolService {    private final WidePatrolService widePatrolService;    public WideOrderPatrolService(WidePatrolService widePatrolService){        this.widePatrolService = widePatrolService;    }    @Override    @DelayBasedRocketMQ(topic = "wide_order_patrol", tag = "SingleIndex", consumerGroup = "order_patrol_group", delayLevel = 2)    public void index(Long aLong){        this.widePatrolService.index(aLong);    }    @Override    public void index(List longs){        WideOrderPatrolService wideOrderPatrolService = ((WideOrderPatrolService)AopContext.currentProxy());        longs.forEach(id -> wideOrderPatrolService.index(id));    }    @Override    public  void updateItem(WideOrderType wideOrderType, KEY key){        ((WideOrderPatrolService)AopContext.currentProxy()).updateItem(wideOrderType, (Long) key);    }    @DelayBasedRocketMQ(topic = "wide_order_patrol", tag = "UpdateByItem", consumerGroup = "order_patrol_group", delayLevel = 2)    public void updateItem(WideOrderType wideOrderType, Long id){        this.widePatrolService.updateItem(wideOrderType, id);    }    @Override    public void setReindexConsumer(Consumer> consumer){        this.widePatrolService.setReindexConsumer(consumer);    }}

WideOrderPatrolService 具体实现如下：

将大部分请求直接转发给内部的 widePatrolService 实例；在索引和更新方法上增加了 @DelayBasedRocketMQ 注解，该注解使的方法拥有延时执行的能力，如果对该注解感兴趣可以翻找下之前的文章；使用 AopContext 在类内获取 Proxy 对象并调用其方法，由于 AOP 实现的限制，在类中直接调用本类中的其他方法，不会触发拦截器；2.6. 实现效果

万事具备只欠东风，写个测试用例测试下功能。

2.6.1. 数据索引

首先，对数据进行索引，示例如下：

// 保存 Userthis.user = new User();this.user.setName("测试");this.userDao.save(this.user);// 保存 Addressthis.address = new Address();this.address.setDetail("详细地址");this.address.setUserId(this.user.getId());this.addressDao.save(this.address);// 保存 Productthis.product = new Product();this.product.setName("商品名称");this.product.setPrice(100);this.productDao.save(this.product);// 保存 Orderthis.order = new Order();this.order.setUserId(this.user.getId());this.order.setAddressId(this.address.getId());this.order.setProductId(this.product.getId());this.order.setDescr("我的订单");this.orderDao.save(this.order);// 进行索引this.wideOrderService.index(this.order.getId());// 比对数据Optional optional = wideOrderDao.findById(this.order.getId());Assertions.assertTrue(optional.isPresent());WideOrder wideOrder = optional.get();Assertions.assertEquals(order.getId(), wideOrder.getId());Assertions.assertEquals(order.getAddressId(), wideOrder.getAddressId());Assertions.assertEquals(order.getProductId(), wideOrder.getProductId());Assertions.assertEquals(order.getUserId(), wideOrder.getUserId());Assertions.assertEquals(order.getDescr(), wideOrder.getOrderDescr());Assertions.assertEquals(user.getName(), wideOrder.getUserName());Assertions.assertEquals(address.getDetail(), wideOrder.getAddressDetail());Assertions.assertEquals(product.getName(), wideOrder.getProductName());Assertions.assertEquals(product.getPrice(), wideOrder.getProductPrice());

单测成功运行后，数据已经成功写入到 ES，具体如下：

2.6.2. 数据更新

更新操作，具体单测如下：

// 更新订单描述this.order.setDescr("订单详情");this.orderDao.save(this.order);// 触发索引更新this.wideOrderService.updateOrder(this.order.getId());// 验证更新结果Optional optional = wideOrderDao.findById(this.order.getId());Assertions.assertTrue(optional.isPresent());WideOrder wideOrder = optional.get();Assertions.assertEquals(order.getId(), wideOrder.getId());Assertions.assertEquals(order.getDescr(), wideOrder.getOrderDescr());

单测成功运行后，数据已经完成更新，ES 数据具体如下：

2.6.3. 数据巡检

仔细观察日志，会发现存在一组 Delay Task 日志，具体如下：

[           main] c.g.l.core.delay.DelayMethodInterceptor  : success to sent Delay Task to RocketMQ for [126][MessageThread_2] c.g.l.c.w.s.SimpleWidePatrolService      : id 126 is same

第一条日志是在提交索引时由主线程打印，向 RocketMQ 提交一个延时任务，用于对 id 为 126 的数据进行校验；

第二条是时间达到后由 Message Consumer 线程打印，表明 DB 与 ES 中的数据是相同的；

如果巡检时发现数据不同，将会自动对 126 进行索引，从而保障两者的一致性；

3. 设计&扩展3.1. 核心设计

整体架构设计如下：

从功能角度，整体可分为如下几部分：

Index 索引部分。内部可以看成是一个基于检索模型的 Pipeline，从众多数据提供器中获取数据，并写入检索模型，最终将填充完数据的检索模型写入的 ES 进行持久化存储；Query 查询部分。直接使用 ES 的 api 对成功索引的数据进行查询。巡检部分。在数据变更时，会自动增加一个延时任务用于数据比较，巡检任务获取变更数据后与ES存储记录进行比较，如果数据不一致则向 Index 模块重新提交索引任务，对问题数据进行再次索引，从而对数据进行恢复；

3.2. 功能扩展

wide 为宽表提供了索引和巡检能力支持，但在实际业务中需要处理多种情况，常见如下：

自动触发，这是系统核心流程之一，数据发生变化后，向 Index 提交新的索引任务。常见的实现策略有：

基于领域事件的索引。监听应用程序发出的领域事件，从而触发新数据的索引；

基于 binlog 的索引。MySQL 的变化全部记录在 binlog 中，可以通过 canal 等框架将 binlog 进行导出，用于触发数据索引；

手工回溯，手工触发索引流程，常见的场景有：

由于业务需要 ES 检索模型发生变更，需要重跑历史数据；

系统故障导致数据不一致，通过手工触发的方式对问题数据进行修复；

天级数据重建。每天凌晨对前一天的数据进行索引重建，主要目的为：

避免错误在 ES 进行累计，也就是在索引和巡检两个机制都不生效的情况下，对问题数据进行修复；

索引优化，在数据完成重建后，可以调用 ES 索引优化接口，对索引进行合并，从而提升系统查询性能；

4. 项目信息