掃二維碼與項(xiàng)目經(jīng)理溝通
我們?cè)谖⑿派?4小時(shí)期待你的聲音
解答本文疑問/技術(shù)咨詢/運(yùn)營(yíng)咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流
相信負(fù)責(zé)過“搜索服務(wù)”的伙伴,最害怕的一句話就是:“數(shù)據(jù)怎么又搜索不出來了?。?!”。每當(dāng)收到這句話,都會(huì)心中一顫,因?yàn)槊鎸?duì)幾千萬甚至幾億的索引數(shù)據(jù),我真的無從下手,不知道業(yè)務(wù)要搜索什么,也不知道是哪些數(shù)據(jù)出了問題….

目前,“搜索”已經(jīng)成為后端管理平臺(tái)的必備功能,在這個(gè)業(yè)務(wù)場(chǎng)景中,很多人都會(huì)基于 elasticsearch 強(qiáng)大的檢索能力構(gòu)建自己的搜索服務(wù)。但實(shí)際開發(fā)中,elasticsearch 的引入是非常小的一部分,往往大頭是索引模型的數(shù)據(jù)管理,在整個(gè)過程中,我們
如此繁瑣的事情,哪一環(huán)出現(xiàn)問題都會(huì)收到業(yè)務(wù)的投訴。
對(duì)搜索場(chǎng)景中的最佳實(shí)踐進(jìn)行封裝,從而:
首先,增加對(duì) spring data elasticsearch 的支持,具體 maven 坐標(biāo)如下:
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.lego lego-starter 0.1.14-wide-SNAPSHOT
至此,就完成了項(xiàng)目的準(zhǔn)備工具,可以著手構(gòu)建索引模型。
構(gòu)造模型之前,需要構(gòu)建一個(gè) Enum 用以管理模型中所有關(guān)聯(lián)數(shù)據(jù),具體如下:
public enum WideOrderType implements WideItemType{
ORDER, // 訂單主數(shù)據(jù)
USER, // 用戶數(shù)據(jù)
ADDRESS, // 用戶地址數(shù)據(jù)
PRODUCT // 購買商品數(shù)據(jù)
}
WideOrderType 枚舉實(shí)現(xiàn) WideItemType 接口,用于與框架進(jìn)行集成。
接下來,構(gòu)建待索引的寬表模型,具體如下:
@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 ListgetItemsKeyByType(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òu)建一些組件用于為“寬表”提供數(shù)據(jù),這就是 WideItemDataProvider 體系。
我們以 OrderProvider 為例,具體如下:
@Component
@org.springframework.core.annotation.Order(value = Ordered.HIGHEST_PRECEDENCE)
public class OrderProvider implements WideItemDataProvider{
@Autowired
private OrderDao orderDao;
@Override
public Listapply(List key){
return orderDao.findAllById(key);
}
@Override
public WideOrderType getSupportType(){
return WideOrderType.ORDER;
}
}
該類有如下特點(diǎn):
每一類關(guān)聯(lián)數(shù)據(jù)都會(huì)提供自己的數(shù)據(jù)提供器,簡(jiǎn)單看下 UserProvider 實(shí)現(xiàn),具體如下:
@Component
public class UserProvider implements WideItemDataProvider{
@Autowired
private UserDao userDao;
@Override
public Listapply(List key){
return userDao.findAllById(key);
}
@Override
public WideOrderType getSupportType(){
return WideOrderType.USER;
}
}
和 OrderProvider 沒有本質(zhì)區(qū)別,當(dāng)然,demo 中還提供了多種實(shí)現(xiàn),如:
數(shù)據(jù)都準(zhǔn)備好了,需要將 “寬表” 進(jìn)行持久化,將其放入最合適的存儲(chǔ)引擎,以便更好的處理查詢請(qǐng)求。
基于 ElasticsearchRepository 的 WideOrderRepository 具體如下:
@Repository
public class WideOrderRepository implements WideCommandRepository{
@Autowired
private WideOrderESDao wideOrderDao;
@Override
public void save(Listwides){
wideOrderDao.saveAll(wides);
}
@Override
public ListfindByIds(List masterIds){
return Lists.newArrayList(wideOrderDao.findAllById(masterIds));
}
@Override
publicvoid 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
publicvoid updateByItem(WideOrderType wideOrderType, KEY key, Consumer wideConsumer){
ConsumerupdateAndSave = 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
publicvoid updateByItem(WideOrderType wideOrderType, KEY key, WideItemData item){
}
}
倉庫具有如下特征:
所依賴的 WideOrderESDao 基于 ElasticsearchRepository 實(shí)現(xiàn),具體如下:
public interface WideOrderESDao extends ElasticsearchRepository{
ListfindByProductId(Long productId);
ListfindByAddressId(Long addressId);
ListfindByUserId(Long userId);
}
所有組件都已準(zhǔn)備好,現(xiàn)在需要將他們整合在一起。
@Configuration
public class WideOrderConfiguration extends WideConfigurationSupport{
@Autowired
private WideOrderRepository wideOrderRepository;
@Autowired
private List>> wideItemDataProviders;
@Bean
public WideIndexServicecreateWideIndexService(){
return super.createWideIndexService();
}
@Bean
public WideOrderPatrolService wideOrderPatrolService(){
return new WideOrderPatrolService(createWidePatrolService());
}
@Bean
protected WideServicecreateWideService(
WideIndexServicewideIndexService,
WideOrderPatrolService wideOrderPatrolService){
return super.createWideService(wideIndexService, wideOrderPatrolService);
}
@Override
protected WideFactorygetWideFactory() {
return WideOrder::new;
}
@Override
protected WideCommandRepositorygetWideCommandRepository() {
return this.wideOrderRepository;
}
@Override
protected List>> getWideItemProviders() {
return this.wideItemDataProviders;
}
}
WideOrderConfiguration 具有如下特點(diǎn):
其中自定義巡檢 WideOrderPatrolService 代碼如下:
public class WideOrderPatrolService implements WidePatrolService{
private final WidePatrolServicewidePatrolService;
public WideOrderPatrolService(WidePatrolServicewidePatrolService){
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(Listlongs){
WideOrderPatrolService wideOrderPatrolService = ((WideOrderPatrolService)AopContext.currentProxy());
longs.forEach(id -> wideOrderPatrolService.index(id));
}
@Override
publicvoid 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 具體實(shí)現(xiàn)如下:
萬事具備只欠東風(fēng),寫個(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ù)
Optionaloptional = 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)成功寫入到 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é)果
Optionaloptional = 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)行索引,從而保障兩者的一致性;
整體架構(gòu)設(shè)計(jì)如下:
從功能角度,整體可分為如下幾部分:
3.2. 功能擴(kuò)展
wide 為寬表提供了索引和巡檢能力支持,但在實(shí)際業(yè)務(wù)中需要處理多種情況,常見如下:
基于領(lǐng)域事件的索引。監(jiān)聽?wèi)?yīng)用程序發(fā)出的領(lǐng)域事件,從而觸發(fā)新數(shù)據(jù)的索引;
基于 binlog 的索引。MySQL 的變化全部記錄在 binlog 中,可以通過 canal 等框架將 binlog 進(jìn)行導(dǎo)出,用于觸發(fā)數(shù)據(jù)索引;
由于業(yè)務(wù)需要 ES 檢索模型發(fā)生變更,需要重跑歷史數(shù)據(jù);
系統(tǒng)故障導(dǎo)致數(shù)據(jù)不一致,通過手工觸發(fā)的方式對(duì)問題數(shù)據(jù)進(jìn)行修復(fù);
避免錯(cuò)誤在 ES 進(jìn)行累計(jì),也就是在索引和巡檢兩個(gè)機(jī)制都不生效的情況下,對(duì)問題數(shù)據(jù)進(jìn)行修復(fù);
索引優(yōu)化,在數(shù)據(jù)完成重建后,可以調(diào)用 ES 索引優(yōu)化接口,對(duì)索引進(jìn)行合并,從而提升系統(tǒng)查詢性能;
項(xiàng)目倉庫地址
項(xiàng)目文檔地址

我們?cè)谖⑿派?4小時(shí)期待你的聲音
解答本文疑問/技術(shù)咨詢/運(yùn)營(yíng)咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流