1. 概覽
相信負(fù)責(zé)過(guò)“搜索服務(wù)”的伙伴，最害怕的一句話就是：“數(shù)據(jù)怎么又搜索不出來(lái)了?。?！”。每當(dāng)收到這句話，都會(huì)心中一顫，因?yàn)槊鎸?duì)幾千萬(wàn)甚至幾億的索引數(shù)據(jù)，我真的無(wú)從下手，不知道業(yè)務(wù)要搜索什么，也不知道是哪些數(shù)據(jù)出了問(wèn)題….
1.1. 背景
目前，“搜索”已經(jīng)成為后端管理平臺(tái)的必備功能，在這個(gè)業(yè)務(wù)場(chǎng)景中，很多人都會(huì)基于 elasticsearch 強(qiáng)大的檢索能力構(gòu)建自己的搜索服務(wù)。但實(shí)際開(kāi)發(fā)中，elasticsearch 的引入是非常小的一部分，往往大頭是索引模型的數(shù)據(jù)管理，在整個(gè)過(guò)程中，我們
- 需要根據(jù)業(yè)務(wù)需求構(gòu)建檢索模型和ES存儲(chǔ)模型；
- 需要從多個(gè)數(shù)據(jù)源中獲取數(shù)據(jù)，并填充到檢索模型；
- 需要關(guān)注所有數(shù)據(jù)源的數(shù)據(jù)變化，并對(duì)變更數(shù)據(jù)進(jìn)行索引重建；
- 需要對(duì)不一致的數(shù)據(jù)進(jìn)行識(shí)別和處理…
如此繁瑣的事情，哪一環(huán)出現(xiàn)問(wèn)題都會(huì)收到業(yè)務(wù)的投訴。
1.2. 目標(biāo)
對(duì)搜索場(chǎng)景中的最佳實(shí)踐進(jìn)行封裝，從而：
- 降低開(kāi)發(fā)成本，開(kāi)發(fā)人員將精力放在模型構(gòu)建上，拋開(kāi)繁瑣的技術(shù)細(xì)節(jié)；
- 對(duì)數(shù)據(jù)索引、關(guān)聯(lián)數(shù)據(jù)更新有很好的支持；
- 引入數(shù)據(jù)實(shí)時(shí)巡檢能力，對(duì)于數(shù)據(jù)不一致的情況進(jìn)行自動(dòng)修復(fù)；
- 引入天級(jí)對(duì)賬機(jī)制，保障數(shù)據(jù)的一致性；
2. 快速入門(mén)
2.1. 準(zhǔn)備環(huán)境
首先，增加對(duì) spring data elasticsearch 的支持，具體 maven 坐標(biāo)如下：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
在 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，具體如下：
<groupId>com.geekhalo.lego</groupId>
<artifactId>lego-starter</artifactId>
<version>0.1.14-wide-SNAPSHOT</version>
至此，就完成了項(xiàng)目的準(zhǔn)備工具，可以著手構(gòu)建索引模型。
2.2. 構(gòu)建模型
構(gòu)造模型之前，需要構(gòu)建一個(gè) Enum 用以管理模型中所有關(guān)聯(lián)數(shù)據(jù)，具體如下：
public enum WideOrderType implements WideItemType<WideOrderType> {
ORDER, // 訂單主數(shù)據(jù)
USER, // 用戶數(shù)據(jù)
ADDRESS, // 用戶地址數(shù)據(jù)
PRODUCT // 購(gòu)買(mǎi)商品數(shù)據(jù)
}WideOrderType 枚舉實(shí)現(xiàn) WideItemType 接口，用于與框架進(jìn)行集成。
接下來(lái)，構(gòu)建待索引的寬表模型，具體如下：
@Data
@NoArgsConstructor
@AllArgsConstructor
@Document(indexName = "wide_order")
public class WideOrder extends BindFromBasedWide<Long, WideOrderType> {
@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<WideItemKey> 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();
}
}
該模型有如下幾個(gè)特點(diǎn)：
- 存在很多屬性，是由多個(gè)表數(shù)據(jù)共同組成的“寬表”；
- 除 id 屬性外，其他屬性上都有 @BindFrom 注解，用于標(biāo)明該字段的數(shù)據(jù)是來(lái)自于哪個(gè)實(shí)體的那個(gè)字段；
- 繼承自 BindFromBasedWide<Long, WideOrderType>，其中 Long 為模型主鍵，WideOrderType 為剛建的枚舉，BindFromBasedWide 將根據(jù)字段上的 @BindFrom 注解自動(dòng)完成 數(shù)據(jù)更新 和 數(shù)據(jù)比對(duì)；
- Long getId() 方法返回模型的主鍵信息;
- boolean isValidate() 用于對(duì)數(shù)據(jù)的有效性進(jìn)行驗(yàn)證，無(wú)效數(shù)據(jù)將不會(huì)進(jìn)行持久化處理
- List<WideItemKey> getItemsKeyByType(WideOrderType wideOrderType) 根據(jù)關(guān)聯(lián)數(shù)據(jù)類型（WideOrderType）返回不同鍵信息，以進(jìn)行數(shù)據(jù)組裝；
至此，模型就建立完畢。
2.3. 數(shù)據(jù)提供器
有了模型后，我們需要構(gòu)建一些組件用于為“寬表”提供數(shù)據(jù)，這就是 WideItemDataProvider 體系。
我們以 OrderProvider 為例，具體如下：
@Component
@org.springframework.core.annotation.Order(value = Ordered.HIGHEST_PRECEDENCE)
public class OrderProvider implements WideItemDataProvider<WideOrderType, Long, Order> {
@Autowired
private OrderDao orderDao;
@Override
public List<Order> apply(List<Long> key){
return orderDao.findAllById(key);
}
@Override
public WideOrderType getSupportType(){
return WideOrderType.ORDER;
}
}
該類有如下特點(diǎn)：
- 實(shí)現(xiàn) WideItemDataProvider 接口，其中 WideOrderType 為剛剛定義的枚舉，Long 為 Order 模型的關(guān)聯(lián)鍵類型，Order 為要提供的數(shù)據(jù)；
- List<Order> apply(List<Long> key)，根據(jù) key 獲得對(duì)應(yīng)的數(shù)據(jù)；
- WideOrderType getSupportType()，獲取該組件所支持的 關(guān)聯(lián)類型；
- @Component 標(biāo)記該類為 Spring 的托管 Bean；
- @Order(value = Ordered.HIGHEST_PRECEDENCE) 指定組件的順序，由于為 WideOrder 提供主數(shù)據(jù)，優(yōu)先級(jí)調(diào)到最高；
每一類關(guān)聯(lián)數(shù)據(jù)都會(huì)提供自己的數(shù)據(jù)提供器，簡(jiǎn)單看下 UserProvider 實(shí)現(xiàn)，具體如下：
@Component
public class UserProvider implements WideItemDataProvider<WideOrderType, Long, User> {
@Autowired
private UserDao userDao;
@Override
public List<User> apply(List<Long> key){
return userDao.findAllById(key);
}
@Override
public WideOrderType getSupportType(){
return WideOrderType.USER;
}
}
和 OrderProvider 沒(méi)有本質(zhì)區(qū)別，當(dāng)然，demo 中還提供了多種實(shí)現(xiàn)，如：
- OrderProvider，提供訂單主數(shù)據(jù)；
- UserProvider，提供用戶信息；
- AddressProvider，提供用戶地址信息；
- ProductProvider，提供商品信息；
2.4. 構(gòu)建寬表倉(cāng)庫(kù)
數(shù)據(jù)都準(zhǔn)備好了，需要將 “寬表” 進(jìn)行持久化，將其放入最合適的存儲(chǔ)引擎，以便更好的處理查詢請(qǐng)求。
基于 ElasticsearchRepository 的 WideOrderRepository 具體如下：
@Repository
public class WideOrderRepository implements WideCommandRepository<Long, WideOrderType, WideOrder> {
@Autowired
private WideOrderESDao wideOrderDao;
@Override
public void save(List<WideOrder> wides){
wideOrderDao.saveAll(wides);
}
@Override
public List<WideOrder> findByIds(List<Long> masterIds){
return Lists.newArrayList(wideOrderDao.findAllById(masterIds));
}
@Override
public <KEY> void consumeByItem(WideOrderType wideOrderType, KEY key, Consumer<WideOrder> 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 <KEY> void updateByItem(WideOrderType wideOrderType, KEY key, Consumer<WideOrder> wideConsumer){
Consumer<WideOrder> 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 <KEY> void updateByItem(WideOrderType wideOrderType, KEY key, WideItemData<WideOrderType, ?> item){
}
}
倉(cāng)庫(kù)具有如下特征：
- 實(shí)現(xiàn) WideCommandRepository<Long, WideOrderType, WideOrder> 接口，其中 Long 是模型主鍵（也是寬表主鍵），WideOrderType 是之前定義的枚舉，WideOrder 是寬表；
- void save(List<WideOrder> wides) 提供批量保存方法；
- List<WideOrder> findByIds(List<Long> masterIds) 提供根據(jù)主鍵批量查詢方法；
- void consumeByItem(WideOrderType wideOrderType, KEY key, Consumer<WideOrder> wideConsumer)，該方法主要用于數(shù)據(jù)巡檢，根據(jù)類型 和 鍵信息 從底層引擎中獲取數(shù)據(jù)，并進(jìn)行部分比對(duì)，用于發(fā)現(xiàn)數(shù)據(jù)不一致情況；
- boolean supportUpdateFor(WideOrderType wideOrderType)，該實(shí)現(xiàn)用于判斷是否支持特定類型的批量更新，及依賴引擎能力批量對(duì)數(shù)據(jù)進(jìn)行更新操作；
- void updateByItem(WideOrderType wideOrderType, KEY key, WideItemData<WideOrderType, ?> item)，supportUpdateFor 返回為 true 時(shí)，調(diào)用該方法，使用引擎的更新能力批量對(duì)數(shù)據(jù)進(jìn)行更新；
- void updateByItem(WideOrderType wideOrderType, KEY key, Consumer<WideOrder> wideConsumer)，supportUpdateFor 返回為 false 時(shí)，調(diào)用該方法，根據(jù) 類型 和 鍵信息 依次查詢所有數(shù)據(jù)，在內(nèi)存中完成更新，并寫(xiě)回存儲(chǔ)引擎；
所依賴的 WideOrderESDao 基于 ElasticsearchRepository 實(shí)現(xiàn)，具體如下：
public interface WideOrderESDao extends ElasticsearchRepository<WideOrder, Long> {
List<WideOrder> findByProductId(Long productId);
List<WideOrder> findByAddressId(Long addressId);
List<WideOrder> findByUserId(Long userId);
}2.5. 配置&整合
所有組件都已準(zhǔn)備好，現(xiàn)在需要將他們整合在一起。
@Configuration
public class WideOrderConfiguration extends WideConfigurationSupport<Long, WideOrderType, WideOrder> {
@Autowired
private WideOrderRepository wideOrderRepository;
@Autowired
private List<WideItemDataProvider<WideOrderType, ?, ? extends WideItemData<WideOrderType, ?>>> wideItemDataProviders;
@Bean
public WideIndexService<Long, WideOrderType> createWideIndexService(){
return super.createWideIndexService();
}
@Bean
public WideOrderPatrolService wideOrderPatrolService(){
return new WideOrderPatrolService(createWidePatrolService());
}
@Bean
protected WideService<Long, WideOrderType> createWideService(
WideIndexService<Long, WideOrderType> wideIndexService,
WideOrderPatrolService wideOrderPatrolService){
return super.createWideService(wideIndexService, wideOrderPatrolService);
}
@Override
protected WideFactory<Long, WideOrder> getWideFactory() {
return WideOrder::new;
}
@Override
protected WideCommandRepository<Long, WideOrderType, WideOrder> getWideCommandRepository() {
return this.wideOrderRepository;
}
@Override
protected List<WideItemDataProvider<WideOrderType, ?, ? extends WideItemData<WideOrderType, ?>>> getWideItemProviders() {
return this.wideItemDataProviders;
}
}
WideOrderConfiguration 具有如下特點(diǎn)：
- 繼承自 WideConfigurationSupport<Long, WideOrderType, WideOrder>，父類中存在大量的 createXXX 方法，可以大幅簡(jiǎn)單代碼量；
- 使用 WideOrderRepository 作為寬表的倉(cāng)庫(kù)；
- 直接使用 Spring 容器中的所有 WideItemDataProvider 實(shí)現(xiàn)；
- 使用定制的 WideOrderPatrolService，為巡檢增加延時(shí)支持；
其中自定義巡檢 WideOrderPatrolService 代碼如下：
public class WideOrderPatrolService implements WidePatrolService<Long, WideOrderType> {
private final WidePatrolService<Long, WideOrderType> widePatrolService;
public WideOrderPatrolService(WidePatrolService<Long, WideOrderType> 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<Long> longs){
WideOrderPatrolService wideOrderPatrolService = ((WideOrderPatrolService)AopContext.currentProxy());
longs.forEach(id -> wideOrderPatrolService.index(id));
}
@Override
public <KEY> 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<List<Long>> consumer){
this.widePatrolService.setReindexConsumer(consumer);
}
}WideOrderPatrolService 具體實(shí)現(xiàn)如下：
- 將大部分請(qǐng)求直接轉(zhuǎn)發(fā)給內(nèi)部的 widePatrolService 實(shí)例；
- 在索引和更新方法上增加了 @DelayBasedRocketMQ 注解，該注解使的方法擁有延時(shí)執(zhí)行的能力，如果對(duì)該注解感興趣可以翻找下之前的文章；
- 使用 AopContext 在類內(nèi)獲取 Proxy 對(duì)象并調(diào)用其方法，由于 AOP 實(shí)現(xiàn)的限制，在類中直接調(diào)用本類中的其他方法，不會(huì)觸發(fā)攔截器；
2.6. 實(shí)現(xiàn)效果
萬(wàn)事具備只欠東風(fēng)，寫(xiě)個(gè)測(cè)試用例測(cè)試下功能。
2.6.1. 數(shù)據(jù)索引
首先，對(duì)數(shù)據(jù)進(jìn)行索引，示例如下：
// 保存 User
this.user = new User();
this.user.setName("測(cè)試");
this.userDao.save(this.user);
// 保存 Address
this.address = new Address();
this.address.setDetail("詳細(xì)地址");
this.address.setUserId(this.user.getId());
this.addressDao.save(this.address);
// 保存 Product
this.product = new Product();
this.product.setName("商品名稱");
this.product.setPrice(100);
this.productDao.save(this.product);
// 保存 Order
this.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);
// 進(jìn)行索引
this.wideOrderService.index(this.order.getId());
// 比對(duì)數(shù)據(jù)
Optional<WideOrder> 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());
單測(cè)成功運(yùn)行后，數(shù)據(jù)已經(jīng)成功寫(xiě)入到 ES，具體如下：
2.6.2. 數(shù)據(jù)更新
更新操作，具體單測(cè)如下：
// 更新訂單描述
this.order.setDescr("訂單詳情");
this.orderDao.save(this.order);
// 觸發(fā)索引更新
this.wideOrderService.updateOrder(this.order.getId());
// 驗(yàn)證更新結(jié)果
Optional<WideOrder> 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());
單測(cè)成功運(yùn)行后，數(shù)據(jù)已經(jīng)完成更新，ES 數(shù)據(jù)具體如下：
2.6.3. 數(shù)據(jù)巡檢
仔細(xì)觀察日志，會(huì)發(fā)現(xiàn)存在一組 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
第一條日志是在提交索引時(shí)由主線程打印，向 RocketMQ 提交一個(gè)延時(shí)任務(wù)，用于對(duì) id 為 126 的數(shù)據(jù)進(jìn)行校驗(yàn)；
第二條是時(shí)間達(dá)到后由 Message Consumer 線程打印，表明 DB 與 ES 中的數(shù)據(jù)是相同的；
如果巡檢時(shí)發(fā)現(xiàn)數(shù)據(jù)不同，將會(huì)自動(dòng)對(duì) 126 進(jìn)行索引，從而保障兩者的一致性；
3. 設(shè)計(jì)&擴(kuò)展
3.1. 核心設(shè)計(jì)
整體架構(gòu)設(shè)計(jì)如下：
從功能角度，整體可分為如下幾部分：
- Index 索引部分。內(nèi)部可以看成是一個(gè)基于 檢索模型 的 Pipeline，從眾多數(shù)據(jù)提供器中獲取數(shù)據(jù)，并寫(xiě)入 檢索模型，最終將填充完數(shù)據(jù)的檢索模型寫(xiě)入的 ES 進(jìn)行持久化存儲(chǔ)；
- Query 查詢部分。直接使用 ES 的 api 對(duì)成功索引的數(shù)據(jù)進(jìn)行查詢。
- 巡檢部分。在數(shù)據(jù)變更時(shí)，會(huì)自動(dòng)增加一個(gè)延時(shí)任務(wù)用于數(shù)據(jù)比較，巡檢任務(wù)獲取變更數(shù)據(jù)后與ES存儲(chǔ)記錄進(jìn)行比較，如果數(shù)據(jù)不一致則向 Index 模塊重新提交索引任務(wù)，對(duì)問(wèn)題數(shù)據(jù)進(jìn)行再次索引，從而對(duì)數(shù)據(jù)進(jìn)行恢復(fù)；
3.2. 功能擴(kuò)展
wide 為寬表提供了索引和巡檢能力支持，但在實(shí)際業(yè)務(wù)中需要處理多種情況，常見(jiàn)如下：
- 自動(dòng)觸發(fā)，這是系統(tǒng)核心流程之一，數(shù)據(jù)發(fā)生變化后，向 Index 提交新的索引任務(wù)。常見(jiàn)的實(shí)現(xiàn)策略有：
基于領(lǐng)域事件的索引。監(jiān)聽(tīng)?wèi)?yīng)用程序發(fā)出的領(lǐng)域事件，從而觸發(fā)新數(shù)據(jù)的索引；
基于 binlog 的索引。MySQL 的變化全部記錄在 binlog 中，可以通過(guò) canal 等框架將 binlog 進(jìn)行導(dǎo)出，用于觸發(fā)數(shù)據(jù)索引；
- 手工回溯，手工觸發(fā)索引流程，常見(jiàn)的場(chǎng)景有：
由于業(yè)務(wù)需要 ES 檢索模型發(fā)生變更，需要重跑歷史數(shù)據(jù)；
系統(tǒng)故障導(dǎo)致數(shù)據(jù)不一致，通過(guò)手工觸發(fā)的方式對(duì)問(wèn)題數(shù)據(jù)進(jìn)行修復(fù)；
- 天級(jí)數(shù)據(jù)重建。每天凌晨對(duì)前一天的數(shù)據(jù)進(jìn)行索引重建，主要目的為：
避免錯(cuò)誤在 ES 進(jìn)行累計(jì)，也就是在索引和巡檢兩個(gè)機(jī)制都不生效的情況下，對(duì)問(wèn)題數(shù)據(jù)進(jìn)行修復(fù)；
索引優(yōu)化，在數(shù)據(jù)完成重建后，可以調(diào)用 ES 索引優(yōu)化接口，對(duì)索引進(jìn)行合并，從而提升系統(tǒng)查詢性能；
4. 項(xiàng)目信息
項(xiàng)目倉(cāng)庫(kù)地址：https://gitee.com/litao851025/lego
項(xiàng)目文檔地址：https://gitee.com/litao851025/lego/wikis/support/Wide%20%E5%AE%BD%E8%A1%A8
