【数据库优化】记一次引入Elasticsearch的系统架构实战

灰太狼 2024-04-01 09:37 80阅读 0赞

## 前言 ##

我曾经面试，当时面试官问了我一个问题：如果你想使用某个新技术但是领导不愿意，你怎么办？

对于该问题我相信大家就算没有面试被问到过，现实工作中同事之间的合作也会遇到。

因此从我的角度重新去回答这个问题，有以下几点：

1.**师出有名**，在软件工程里是针对问题场景提供解决方案的，如果脱离的实际问题（需求）去做技术选型，无疑是耍流氓。大家可以回顾身边的“架构师”、“技术Leader”是不是拍拍脑袋做决定，问他们为什么这么做，可能连个冠冕堂皇的理由都给不出。

2.**信任度**，只有基于上面的条件，你才有理由建议引入新技术。领导愿不愿意引入新技术有很多原因：领导不了解这技术、领导偏保守、领导不是做技术的等。那么我认为这几种都是信任度，这种**信任度分人和事**，人就是引入技术的提出者，事就是提出引入的技术。

3.**尽人事**，任何问题只是单纯解决**事**都是简单的，以我以往的做法，把基本资料收集全并以通俗易懂的方式归纳与讲解，最好能提供一些能量化的数据，这样更加有说服力。知识普及OK后，就可以尝试写方案与做个Demo，方案最好可以提供多个，可以分短期收益与长期收益的。完成上面几点可以说已经尽人事了，如果领导还不答应那么的确有他的顾虑，就算无法落实，到目前为止的收获也不错。

4.**复杂的是人**，任何人都无法时刻站在理智与客观的角度去看待问题，**事是由人去办的**，所以同一件事由不同的人说出来的效果也不一样。因此得学会向上管理、保持与同事之间合作融洽度，尽早的建立合作信任。本篇文章更多叙述的事，因此人方面不过多深究，有兴趣的我可以介绍一本书《**知行 技术人的管理之路**》。

本篇我的实践做法与上述一样，除了4无法体现。那么下文我分了4大模块：**业务背景介绍**、**基础概念讲解**、**方案的选用**与**技术细节**。

部分源码，我放到了https://github.com/SkyChenSky/Sikiro 的Sikiro.ES.Api里

## 背景 ##

本公司多年以来用SQL Server作为主存储，随着多年的业务发展，已经到了数千万级的数据量。

而部分非核心业务原本应该超亿的量级了，但是因为从物理表的设计优化上进行了数据压缩，导致维持在一个比较稳定的数量。压缩数据虽然能减少存储量，优化提供一定的性能，但是同时带来的损失了业务可扩展性。举个例子：我们平台某个用户拥有**最后访问作品记录**和**总的阅读时长**，但是没有某个**用户的阅读明细，**那么这样的设计就会导致后续新增一个**抽奖业务**，需要在某个时间段内阅读了多长时间或者章节数量的作品，才能参加与抽奖；或者运营想通过阅读记录统计或者分析出，**用户的爱好**和**受欢迎的作品**。现有的设计对以上两种业务情况都是无法满足的。

此外我们平台还有作品搜索功能，like ‘%搜索%’查询是不走索引的而走**全表扫描**，一张表42W全表扫描，数据库服务器配置可以的情况下还是可以的，但是存在**并发请求**时候，资源消耗就特别厉害了，特别是在偶尔被爬虫爬取数据。（我们平台API的并发峰值能达到8w/s，每天的接口在淡季请求次数达到了1亿1千万）

关系型数据库拥有ACID特性，能通过金融级的事务达成数据的一致性，然而它却没有横向扩展性，只要在海量数据场景下，单实例，无论怎么在关系型数据库做优化，都是只是治标。而NoSQL的出现很好的弥补了关系型数据库的短板，在马丁福勒所著的《**NoSQL精粹**》对NoSQL进行了分类：**文档型、图形、列式，键值**，从我的角度其实可以把**搜索引擎**纳入NoSQL范畴，因为它的确满足的NoSQL的4大特性：**易扩展、大数据量高性能、灵活的数据模型、高可用**。我看过一些同行的见解，把Elasticsearch归为**文档型NoSQL**，我个人是没有给他下过于明确的定义，这个上面说法大家见仁见智。

MongoDB作为文档型数据库也属于我的技术选型范围，它的读写性能高且平衡、数据分片与横向扩展等都非常适合我们平台部分场景，最后我还是选择Elasticsearch。原因有三：

1.  我们运维相比于MongoDB更熟悉Elasticsearch。
2.  我们接下来有一些统计报表类的需求，Elastic Stack的各种工具能很好满足我们的需求。
3.  我们目前着手处理的场景以非实时、纯读为主的业务，Elasticsearch近实时搜索已经能满足我们。

## Elasticsearch优缺点 ##

> **百度百科 ：**
> 
> Elasticsearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch由Java语言开发的，是一种流行的企业级搜索引擎。Elasticsearch用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。官方客户端在Java、.NET（C\#）、PHP、Python、Apache Groovy、Ruby和许多其他语言中都是可用的。

对于满足当下的业务需求和未来支持海量数据的搜索，我选择了Elasticsearch，其实原因主要以下几点：

<table> 
 <tbody> 
  <tr> 
   <td><strong>优点</strong></td> 
   <td><strong>描述</strong></td> 
  </tr> 
  <tr> 
   <td>横向可扩展性</td> 
   <td>可单机、可集群，横向扩展非常简单方便，自动整理数据分片</td> 
  </tr> 
  <tr> 
   <td>快</td> 
   <td>索引被分为多个分片(Shard)，利用多台服务器，使用了分而治之的思想提升处理效率</td> 
  </tr> 
  <tr> 
   <td>支持搜索多样化</td> 
   <td>与传统关系型数据库相比，ES提供了全文检索、同义词处理、相关度排名、复杂数据分析、海量数据的近实时处理等功能</td> 
  </tr> 
  <tr> 
   <td>高可用</td> 
   <td>提供副本(Replica)机制，一个分片可以设置多个副本，假如某服务器宕机后，集群仍能正常工作。</td> 
  </tr> 
  <tr> 
   <td>开箱即用</td> 
   <td>简易的运维部署，提供基于Restful API，多种语言的SDK</td> 
  </tr> 
 </tbody> 
</table>

那么我个人认为Elasticsearch比较大的**缺点**只有 **吃内存，**具体原因可以看下文内存读取部分。

## Elasticsearch为什么快？ ##

我个人对于Elasticsearch快的原因主要总结三点：

*  **内存读取**
 *  **多种索引**
    
     *  **倒排索引　　**
     *  **doc values**
 *  **集群分片**

### 内存读取 ###

Elasticsearch是基于Lucene， 而Lucene被设计为可以利用操作系统底层机制来缓存内存数据结构，换句话说Elasticsearch是依赖于操作系统底层的 **Filesystem Cache**，查询时，操作系统会将磁盘文件里的数据自动缓存到 Filesystem Cache 里面去，因此要求Elasticsearch性能足够高，那么就需要服务器的提供的**足够内存给Filesystem Cache 覆盖存储的数据**。

上一段最后一句话什么意思呢？假如：Elasticsearch 节点有 3 台服务器各64G内存，3台总内存就是 64 \* 3 = 192G。每台机器给 Elasticsearch jvm heap 是 32G，那么每服务器留给 **Filesystem Cache** 的就是 32G（50%），而集群里的 **Filesystem Cache** 的就是 32 \* 3 = 96G 内存。此时，在 3 台Elasticsearch服务器共占用了 1T 的磁盘容量，那么每台机器的数据量约等于 341G，意味着每台服务器只有大概**10分之1数据是缓存**在内存的，其余都得走硬盘。

说到这里大家未必会有一个直观得认识，因此我从**《大型网站技术架构：核心原理与案例分析》**第36页抠了一张表格下来：

<table> 
 <tbody> 
  <tr> 
   <td><strong>操作</strong></td> 
   <td><strong>响应时间</strong></td> 
  </tr> 
  <tr> 
   <td>打开一个网站</td> 
   <td>几秒</td> 
  </tr> 
  <tr> 
   <td>在数据库中查询一条记录（有索引）</td> 
   <td>十几毫秒</td> 
  </tr> 
  <tr> 
   <td>机械磁盘一次寻址定位</td> 
   <td>4毫秒</td> 
  </tr> 
  <tr> 
   <td><strong>从机械磁盘顺序读取1MB数据</strong></td> 
   <td><strong>2毫秒</strong></td> 
  </tr> 
  <tr> 
   <td>从SSD磁盘顺序读取1MB数据</td> 
   <td>0.3毫秒</td> 
  </tr> 
  <tr> 
   <td>从远程分布式缓存Redis读取一个数据</td> 
   <td>0.5毫秒</td> 
  </tr> 
  <tr> 
   <td><strong>从内存中读取1MB数据</strong></td> 
   <td><strong>十几微秒</strong></td> 
  </tr> 
  <tr> 
   <td>Java程序本地方法调用</td> 
   <td>几微秒</td> 
  </tr> 
  <tr> 
   <td>网络传输2KB数据</td> 
   <td>1微秒</td> 
  </tr> 
 </tbody> 
</table>

从上图加粗项看出，**内存读取性能**是机械磁盘的200倍，是SSD磁盘约等于30倍，**假如读一次Elasticsearch走内存场景下耗时20毫秒，那么走机械硬盘就得4秒，走SSD磁盘可能约等于0.6秒**。讲到这里我相信大家对是否走内存的性能差异有一个直观的认识。

对于Elasticsearch有很多种索引类型，但是我认为核心主要是倒排索引和doc values

### 倒排索引 ###

Lucene将写入索引的所有信息组织为倒排索引（inverted index）的结构形式。倒排索引是一种将分词映射到文档的数据结构，可以认为倒排索引是面向分词的而不是面向文档的。

假设在测试环境的Elasticsearch存放了有以下三个文档：

*  Elasticsearch Server（文档1）
 *  Masterring Elasticsearch（文档2）
 *  Apache Solr 4 Cookbook（文档3）

以上文档索引建好后，简略显示如下：

<table> 
 <tbody> 
  <tr> 
   <td><strong>词项</strong></td> 
   <td><strong>数量</strong></td> 
   <td><strong>文档</strong></td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>1</td> 
   <td>&lt;3&gt;</td> 
  </tr> 
  <tr> 
   <td>Apache</td> 
   <td>1</td> 
   <td>&lt;3&gt;</td> 
  </tr> 
  <tr> 
   <td>Cookbook</td> 
   <td>1</td> 
   <td>&lt;3&gt;</td> 
  </tr> 
  <tr> 
   <td>Elasticsearch</td> 
   <td>2</td> 
   <td>&lt;1&gt;&lt;2&gt;</td> 
  </tr> 
  <tr> 
   <td>Mastering</td> 
   <td>1</td> 
   <td>&lt;1&gt;</td> 
  </tr> 
  <tr> 
   <td>Server</td> 
   <td>1</td> 
   <td>&lt;1&gt;</td> 
  </tr> 
  <tr> 
   <td>Solr</td> 
   <td>1</td> 
   <td>&lt;3&gt;</td> 
  </tr> 
 </tbody> 
</table>

如上表格所示，每个词项指向该词项所出现过的文档位置，这种索引结构允许快速、有效的搜索出数据。

### doc values　　 ###

对于分组、聚合、排序等某些功能来说，倒排索引的方式并不是最佳选择，这类功能操作的是文档而不是词项，这个时候就得把倒排索引逆转过来成正排索引，这么做会有两个缺点：

1.  构建时间长
2.  内存占用大，易OutOfMemory，且影响垃圾回收

Lucene 4.0之后版本引入了doc values和额外的数据结构来解决上面得问题，目前有五种类型的doc values：NUMERIC、BINARY、SORTED、SORTED\_SET、SORTED\_NUMERIC，针对每种类型Lucene都有特定的压缩方法。

doc values是列式存储的正排索引，通过docID可以快速读取到该doc的特定字段的值，列式存储存储对于聚合计算有非常高的性能。

### 集群分片 ###

**Elasticsearch**可以简单、快速利用多节点服务器形成**集群**，以此分摊服务器的**执行压力**。

此外数据可以进行**分片存储**，搜索时**并发**到不同服务器上的主分片进行搜索。

这里可以简单讲述下Elasticsearch查询原理，Elasticsearch的查询分两个阶段：**分散阶段**与**合并阶段**。

任意一个Elasticsearch节点都可以接受客户端的请求。接受到请求后，就是**分散阶段**，并行发送子查询给其他节点；

然后是**合并阶段，**则从众多分片中收集返回结果，然后对他们进行合并、排序、取长等后续操作。最终将结果返回给客户端。

机制如下图：

![6c732fe2ae205977ec9ff9264ce8d925.png][]

#### 分页深度陷阱 ####

基于以上查询的原理，扩展一个分页深度的问题。

现需要查页长为10、第100页的数据，实际上是会把每个 Shard 上存储的前 1000（10\*100） 条数据都查到一个协调节点上。如果有 5 个 Shard，那么就有 5000 条数据，接着协调节点对这 5000 条数据进行一些合并、处理，再获取到最终第 100 页的 10 条数据。也就是实际上查的数据总量为pageSize\*pageIndex\*shard，页数越深则查询的越慢。因此ElasticSearch也会有要求，每次查询出来的**数据总数不会返回超过****10000****条。**

那么从业务上尽可能跟产品沟通避免分页跳转，使用滚动加载。而**Elasticsearch**使用的相关技术是search\_after、scroll\_id。

## ElasticSearch与数据库基本概念对比 ##

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;"> <p><strong>ElasticSearch</strong></p> </td> 
   <td style="vertical-align:top;"> <p><strong>RDBMS</strong></p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;"> <p>Index</p> </td> 
   <td style="vertical-align:top;"> <p>表</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;"> <p>Document</p> </td> 
   <td style="vertical-align:top;"> <p>行</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;"> <p>Field</p> </td> 
   <td style="vertical-align:top;"> <p>列</p> </td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;"> <p>Mapping</p> </td> 
   <td style="vertical-align:top;"> <p>表结构</p> </td> 
  </tr> 
 </tbody> 
</table>

在**Elasticsearch** 7.0版本之前（<7.0），有type的概念，而**Elasticsearch**和**关系型数据库**的关系是，index = database、type = table，但是在**Elasticsearch** 7.0版本后（>=7.0）弱化了type默认为\_doc，而官方会在8.0之后会彻底移除type。

## 服务器选型 ##

在官方文档（https://www.elastic.co/guide/cn/elasticsearch/guide/current/heap-sizing.html）里建议**Elasticsearch**  JVM Heap最大为32G，同时不超过服务器内存的一半，也就是说内存分别为128G和64G的服务器，JVM Heap最大只需要设置32G；而32G服务器，则建议JVM Heap最大16G，剩余的内存将会给到**Filesystem Cache**充分使用。如果不需要对分词字符串做聚合计算（例如，不需要 fielddata ）可以考虑降低JVM Heap。JVM Heap越小，会导致Elasticsearch的GC频率更高，但Lucene就可以的使用更多的内存，这样性能就会更高。

对于我们公司的未来新增业务还会有收集用户的访问记录来统计PV(page view)、UV(user view)，有一定的聚合计算，经过多方便的考虑与讨论，**平衡成本与需求**后选择了腾讯云的三台配置为CPU 16核、内存64G，SSD云硬盘的服务器，并给与**Elasticsearch 配置**JVM Heap = 32G。

## 需求场景选择 ##

**Elasticsearch**在本公司系统的可使用场景非常多，但是作为第一次引入因慎重选择，给与开发与运维一定的时间熟悉与观察。

经过商讨，选择了两个业务场景，**用户阅读作品的记录明细**与**作品搜索，**选择这两个业务场景原因如下：

*  写场景
    
     *  我们平台的用户黏度比较高，阅读作品是一个高频率的调用，因此**用户阅读作品的记录明细**可在短时间内造成海量数据的场景。（现一个月已达到了70G的数据量，共1亿1千万条）

*  读场景
    
     *  阅读记录需提供给未来新增的抽奖业务使用，可从阅读章节数、阅读时长等进行搜索计算。
     *  **作品搜索**原有实现是通过关系型数据库like查询，已是具有潜在的性能问题与资源消耗的业务场景

对于上述两个业务，**用户阅读作品的记录明细**与**抽奖业务**属于新增业务，对于在投入成本相对较少，也无需过多的需要兼容旧业务的压力。

而作品搜索业务属于优化改造，得保证兼容原有的用户搜索习惯前提下，新增拼音搜索。同时最好以扩展的方式，尽可能的减少代码修改范围，如果使用效果不好，随时可以回滚到旧的实现方式。

## 设计方案 ##

### 共性设计 ###

我使用.Net 5 WebApi将Elasticsearch封装成ES业务服务API，这样的做法主要用来隐藏技术细节（时区、分词器、类型转换等），暴露粗粒度的读写接口。这种做法在马丁福勒所著的《**NoSQL精粹**》称把数据库视为“**应用程序数据库**”，简单来说就是只能通过**应用间接**的访问存储，对于这个应用由一个团队负责维护开发，也只有这个团队才知道存储的结构。这样通过封装的API服务解耦了外部API服务与存储，调用方就无需过多关注存储的特性，像Mongodb与Elasticsearch这种**无模式**的存储，无需优先定义结构，换而言之就是对于存储已有结构可随意修改扩展，那么“**应用程序数据库**”的做法也避免了其他团队无意侵入的修改。

考虑到现在业务需求复杂度相对简单，MQ消费端也一起集成到ES业务服务，若后续MQ消费业务持续增多，再考虑把MQ消费业务抽离到一个（或多个的）消费端进程。

目前以**同步读、同步写、异步写**的三种交互方式，进行与其他服务通信。

### 阅读记录明细 ###

本需求是完全新增，因此引入相对简单，只需要在**【平台****API****】**使用【**RabbitMQ**】进行解耦，使用异步方式写入Elasticsearch，使用队列除了用来解耦，还对此用来缓冲高并发写压力的情况。

对于后续新增的业务例如抽奖服务，则只需要通过RPC框架对接**ES业务API，**以同步读取的方式查询数据。

![510833903091d56a67926ff91379fbee.png][]

### 作品搜索 ###

对于该业务，我第一反应采用CQRS的思想，原有的写入逻辑我无需过多的关注与了解，因此我只需要想办法把关系型数据库的**数据同步**到Elasticsearch，然后提供**业务查询API**替换原有**平台API的数据源**即可。

那么数据同步则一般都是分**推**和**拉**两种方式。

### 推 ###

推的实时性无疑是比拉要高，只需**增量的**推送做写入的数据（增、删、改）即可，无论是从性能、资源利用、时效各方面来看都比拉更有效。

实施该方案，可以选择Debezium和SQL Server开启CDC功能。

Debezium由RedHat开源的，同时需要依赖于kafka的，一个将多种数据源实时变更数据捕获，形成数据流输出的开源工具，同类产品有Canal, DataBus, Maxwell。

CDC全称Change Data Capture，直接翻译过来为变更数据捕获，核心为监测服务捕获数据库的写操作（插入，更新，删除），将这些变更按发生的顺序完整记录下来。

**我个人在我博客文章多次强调架构设计的输入核心为两点**：**满足需求**与**组织架构，**在满足需求的前提应优先选择简单、合适的方案。技术选型应需要考虑自己的团队是否可以支撑。在上述无论是额外加入Debezium和kafka，还是需要针对SQL Server开启CDC都超出了我们运维所能承受的极限，引入新的中间件和技术是需要试错的，而试错是需要额外高的成本，在未知的情况下引入更多的未知，只会造成更大的成本和不可控。

### 拉 ###

拉无疑是最简单最合适的实现方式，只需要使用调度任务服务，每隔段时间定时去从数据库拉取数据写入到Elasticsearch就可。

然而拉取数据，分**全量同步**与**增量同步**：

对于**增量同步**，只需要每次查询数据源Select \* From Table\_A Where RowVersion > LastUpdateVersion，则可以过滤出需要同步的数据。但是这个方式有点致命的缺点，数据源已被删除的数据是无法查询出来的，如果把Elasticsearch反向去跟SQL Server数据做对比又是一件比较愚蠢的方式，因此只能放弃该方式。

而**全量同步**，只要每次从SQL Server数据源全量新增到Elasticsearch，并替换旧的Elasticsearch的Index，因此该方案得全删全增。但是这里又引申出新的问题，如果先删后增，那么在删除后再新增的这段真空期怎么办？假如有5分钟的真空期是没有数据，用户就无法使用搜索功能。那么只能先增后删，先新增到一个Index\_Temp，全量新增完后，把原有Index改名成Index\_Delete，然后再把Index\_Temp改成Index，最后把Index\_Delete删除。这么一套操作下来，有没有觉得很繁琐很费劲？Elasticsearch有一个叫**别名（Aliases）**的功能，别名可以一对多的指向多个Index，也可以**以原子性**的进行别名指向Index的切换，具体实现可以看下文。

![8bbf11ad6f3d3cb68c49deff3c79b359.png][]

## 阅读记录实现细节 ##

### 实体定义 ###

优先定义了个抽象类ElasticsearchEntity进行复用，对于实体定义有三个注意的细节点：

1.对于ElasticsearchEntity我定义两个属性\_id与Timestamp，Elasticsearch是**无模式**的（无需预定义结构），如果实体本身没有\_id，写入到Elasticsearch会自动生成一个\_id，为了后续的使用便捷性，我仍然自主定义了一个。

2.基于上述的分页深度的问题，因此在后续涉及的业务尽可能会以**search\_after+滚动加载**的方式落实到我们的业务。原本我们只需要使用DateTime类型的字段用DateTime.Now记录后，再使用search\_after后会自动把DateTime类型字段转换成毫秒级的Timestamp，但是我在实现demo的时候，去制造数据，在程序里以for循环new数据的时候，发现生成的速度会在微秒级之间，那么假设用毫秒级的Timestamp进行search\_after过滤，同一个毫秒有4、5条数据，那么容易在使用滚动加载时候少加载了几条数据，这样就到导致数据返回不准确了。因此我扩展了个\[*DateTime.Now.DateTimeToTimestampOfMicrosecond()*\]生成微秒级的Timestamp，以此尽可能减少出现漏加载数据的情况。

3.对于Elasticsearch的操作实体的日期时间类型均以DateTimeOffset类型声明，因为Elasticsearch存储的是UTC时间，而且会因为Http请求的日期格式不同导致存放的日期时间也会有所偏差，为了避免日期问题使用DateTimeOffset类型是一种保险的做法。而对于WebAPI 接口或者MQ的Message接受的时间类型可以使用DateTime类型，DTO(传输对象)与DO（持久化对象）使用Mapster或者AutoMapper类似的对象映射工具进行转换即可（注意DateTimeOffset转DateTime得定义转换规则 \[*TypeAdapterConfig<DateTimeOffset, DateTime>.NewConfig().MapWith(dateTimeOffset => dateTimeOffset.LocalDateTime*)\]）。

**如此一来，把Elasticsearch操作细节隐藏在WebAPI里，以友好、简单的接口暴露给开发者使用，降低了开发者对技术细节认知负担。**

[ElasticsearchType(RelationName = "user_view_duration")]
        public class UserViewDuration : ElasticsearchEntity
        {
            /// <summary>
            /// 作品ID
            /// </summary>
            [Number(NumberType.Long, Name = "entity_id")]
            public long EntityId { get; set; }
    
            /// <summary>
            /// 作品类型
            /// </summary>
            [Number(NumberType.Long, Name = "entity_type")]
            public long EntityType { get; set; }
    
            /// <summary>
            /// 章节ID
            /// </summary>
            [Number(NumberType.Long, Name = "charpter_id")]
            public long CharpterId { get; set; }
    
            /// <summary>
            /// 用户ID
            /// </summary>
            [Number(NumberType.Long, Name = "user_id")]
            public long UserId { get; set; }
    
            /// <summary>
            /// 创建时间
            /// </summary>
            [Date(Name = "create_datetime")]
            public DateTimeOffset CreateDateTime { get; set; }
    
            /// <summary>
            /// 时长
            /// </summary>
            [Number(NumberType.Long, Name = "duration")]
            public long Duration { get; set; }
    
            /// <summary>
            /// IP
            /// </summary>
            [Ip(Name = "Ip")]
            public string Ip { get; set; }
        }

public abstract class ElasticsearchEntity
        {
            private Guid? _id;
    
            public Guid Id
            {
                get
                {
                    _id ??= Guid.NewGuid();
                    return _id.Value;
                }
                set => _id = value;
            }
    
            private long? _timestamp;
    
            [Number(NumberType.Long, Name = "timestamp")]
            public long Timestamp
            {
                get
                {
                    _timestamp ??= DateTime.Now.DateTimeToTimestampOfMicrosecond();
                    return _timestamp.Value;
                }
                set => _timestamp = value;
            }
        }

### 异步写入 ###

对于异步写入有两个细节点：

1.该数据从RabbtiMQ订阅消费写入到Elasticsearch，从下面代码可以看出，我刻意以月的维度建立Index，格式为 *userviewrecord-2021-12，*这么做的目的是为了方便管理Index和资源利用，有需要的情况下会删除旧的Index。

2.消息订阅与WebAPI暂时集成到同一个进程，这样做主要是开发、部署都方便，如果后续订阅多了，在把消息订阅相关的业务抽离到独立的进程。

**按需演变，避免过度设计**

订阅消费逻辑

public class UserViewDurationConsumer : BaseConsumer<UserViewDurationMessage>
        {
            private readonly ElasticClient _elasticClient;
    
            public UserViewDurationConsumer(ElasticClient elasticClient)
            {
                _elasticClient = elasticClient;
            }
    
            public override void Excute(UserViewDurationMessage msg)
            {
                var document = msg.MapTo<Entity.UserViewDuration>();
    
                var result = _elasticClient.Create(document, a => a.Index(typeof(Entity.UserViewDuration).GetRelationName() + "-" + msg.CreateDateTime.ToString("yyyy-MM"))).GetApiResult();
                if (result.Failed)
                    LoggerHelper.WriteToFile(result.Message);
            }
        }

/// <summary>
        /// 订阅消费
        /// </summary>
        public static class ConsumerExtension
        {
            public static IApplicationBuilder UseSubscribe<T, TConsumer>(this IApplicationBuilder appBuilder, IHostApplicationLifetime lifetime) where T : EasyNetQEntity, new() where TConsumer : BaseConsumer<T>
            {
                var bus = appBuilder.ApplicationServices.GetRequiredService<IBus>();
                var consumer = appBuilder.ApplicationServices.GetRequiredService<TConsumer>();
    
                lifetime.ApplicationStarted.Register(() =>
                {
                    bus.Subscribe<T>(msg => consumer.Excute(msg));
                });
    
                lifetime.ApplicationStopped.Register(() => bus?.Dispose());
    
                return appBuilder;
            }
        }

订阅与注入

public class Startup
        {
            public Startup(IConfiguration configuration)
            {
                Configuration = configuration;
            }
    
            public IConfiguration Configuration { get; }
    
            public void ConfigureServices(IServiceCollection services)
            {
                ......
            }
    
            public void Configure(IApplicationBuilder app, IWebHostEnvironment env, IHostApplicationLifetime lifetime)
            {
                app.UseAllElasticApm(Configuration);
    
                app.UseHealthChecks("/health");
    
                app.UseDeveloperExceptionPage();
                app.UseSwagger();
                app.UseSwaggerUI(c =>
                {
                    c.SwaggerEndpoint("/swagger/v1/swagger.json", "SF.ES.Api v1");
                    c.RoutePrefix = "";
                });
    
                app.UseRouting();
                app.UseEndpoints(endpoints =>
                {
                    endpoints.MapControllers();
                });
    
                app.UseSubscribe<UserViewDurationMessage, UserViewDurationConsumer>(lifetime);
            }
        }

### 查询接口 ###

查询接口此处有两个细节点：

1.如果不确定月份，则使用通配符查询*userviewrecord-\*，*当然有需要的也可以使用别名处理。

2.因为Elasticsearch是记录UTC时间，因此时间查询得指定TimeZone。

[HttpGet]
            [Route("record")]
            public ApiResult<List<UserMarkRecordGetRecordResponse>> GetRecord([FromQuery] UserViewDurationRecordGetRequest request)
            {
                var dataList = new List<UserMarkRecordGetRecordResponse>();
    
                string dateTime;
    
                if (request.BeginDateTime.HasValue && request.EndDateTime.HasValue)
                {
                    var month = request.EndDateTime.Value.DifferMonth(request.BeginDateTime.Value);
    
                    if(month <= 0 )
                        dateTime = request.BeginDateTime.Value.ToString("yyyy-MM");
                    else
                        dateTime = "*";
                }
                else
                    dateTime = "*";
    
                var mustQuerys = new List<Func<QueryContainerDescriptor<UserViewDuration>, QueryContainer>>();
    
                if (request.UserId.HasValue)
                    mustQuerys.Add(a => a.Term(t => t.Field(f => f.UserId).Value(request.UserId.Value)));
    
                if (request.EntityType.HasValue)
                    mustQuerys.Add(a => a.Term(t => t.Field(f => f.EntityType).Value(request.EntityType)));
    
                if (request.EntityId.HasValue)
                    mustQuerys.Add(a => a.Term(t => t.Field(f => f.EntityId).Value(request.EntityId.Value)));
    
                if (request.CharpterId.HasValue)
                    mustQuerys.Add(a => a.Term(t => t.Field(f => f.CharpterId).Value(request.CharpterId.Value)));
    
                if (request.BeginDateTime.HasValue)
                    mustQuerys.Add(a => a.DateRange(dr =>
                        dr.Field(f => f.CreateDateTime).GreaterThanOrEquals(request.BeginDateTime.Value).TimeZone(EsConst.TimeZone)));
    
                if (request.EndDateTime.HasValue)
                    mustQuerys.Add(a =>
                        a.DateRange(dr => dr.Field(f => f.CreateDateTime).LessThanOrEquals(request.EndDateTime.Value).TimeZone(EsConst.TimeZone)));
    
                var searchResult = _elasticClient.Search<UserViewDuration>(a =>
                    a.Index(typeof(UserViewDuration).GetRelationName() + "-" + dateTime)
                        .Size(request.Size)
                        .Query(q => q.Bool(b => b.Must(mustQuerys)))
                        .SearchAfterTimestamp(request.Timestamp)
                        .Sort(s => s.Field(f => f.Timestamp, SortOrder.Descending)));
    
                var apiResult = searchResult.GetApiResult<UserViewDuration, List<UserMarkRecordGetRecordResponse>>();
                if (apiResult.Success)
                    dataList.AddRange(apiResult.Data);
    
                return ApiResult<List<UserMarkRecordGetRecordResponse>>.IsSuccess(dataList);
            }

## 作品搜索实现细节 ##

### 实体定义 ###

SearchKey是原有SQL Server的数据，现需要同步到Elasticsearch，仍是继承抽象类*ElasticsearchEntity*实体定义，同时这里有三个细节点：

1. *public string KeyName，*我定义的是**Text类型**，在Elasticsearch使用Text类型才会分词。

2.在实体定义我没有给*KeyName*指定分词器，因为我会使用两个分词器：拼音和默认分词，而我会在批量写入数据创建Mapping时定义。

3.实体里的 *public List<int> SysTagId* 与SearchKey在SQL Server是两张不同的物理表*，*是一对多的关系，在代码表示如下，但是在关系型数据库是无法与之对应和体现的，这就是咱们所说的“**阻抗失配”**，但是能在以文档型存储系统（MongoDB、Elasticsearch）里很好的解决这个问题，可以以一个**聚合**的方式写入，避免多次查询关联。

[ElasticsearchType(RelationName = "search_key")]
        public class SearchKey : ElasticsearchEntity
        {
            [Number(NumberType.Integer, Name = "key_id")]
            public int KeyId { get; set; }
    
            [Number(NumberType.Integer, Name = "entity_id")]
            public int EntityId { get; set; }
    
            [Number(NumberType.Integer, Name = "entity_type")]
            public int EntityType { get; set; }
    
            [Text(Name = "key_name")]
            public string KeyName { get; set; }
    
            [Number(NumberType.Integer, Name = "weight")]
            public int Weight { get; set; }
    
            [Boolean(Name = "is_subsidiary")]
            public bool IsSubsidiary { get; set; }
    
            [Date(Name = "active_date")]
            public DateTimeOffset? ActiveDate { get; set; }
    
            [Number(NumberType.Integer, Name = "sys_tag_id")]
            public List<int> SysTagId { get; set; }
        }

### 数据同步 ###

数据同步我采用了Quartz.Net定时调度任务框架，因此时效不高，所以每4小时同步一次即可，有42W多的数据，分批进行同步，每次查询1000条数据同时进行一次批量写入。全量同步一次的时间大概2分钟。因此使用RPC调用\[ES业务API服务\]。

因为具体业务逻辑已经封装在\[ES业务API服务\]，因此同步逻辑也相对简单，查询出SQL Server数据源、聚合整理、调用\[ES业务API服务\]的批量写入接口、重新绑定别名到新的Index。

[DisallowConcurrentExecution]
        public class SearchKeySynchronousJob : BaseJob
        {
            public override void Execute()
            {
                var rm = SFNovelReadManager.Instance();
    
                var maxId = 0;
                var size = 1000;
                string indexName = "";
    
                while (true)
                {
                    //避免一次性全部查询出来，每1000条一次写入。
                    var searchKey = sm.searchKey.GetList(size, maxId);
    
                    if (!searchKey.Any())
                        break;
    
                    var entityIds = searchKey.Select(a => a.EntityID).Distinct().ToList();
    
                    var sysTagRecord = rm.Novel.GetSysTagRecord(entityIds);
    
                    var items = searchKey.Select(a => new SearchKeyPostItem
                    {
                        Weight = a.Weight,
                        EntityType = a.EntityType,
                        EntityId = a.EntityID,
                        IsSubsidiary = a.IsSubsidiary ?? false,
                        KeyName = a.KeyName,
                        ActiveDate = a.ActiveDate,
                        SysTagId = sysTagRecord.Where(c => c.EntityID == a.EntityID).Select(c => c.SysTagID).ToList(),
                        KeyID = a.KeyID
                    }).ToList();
    
                    //以一个聚合写入到ES
                    var postResult = new SearchKeyPostRequest
                    {
                        IndexName = indexName,
                        Items = items
                    }.Excute();
    
                    if (postResult.Success)
                    {
                        indexName = (string)postResult.Data;
                        maxId = searchKey.Max(a => a.KeyID);
                    }
                }
    
                //别名从旧Index指向新的Index，最后删除旧Index
                var renameResult = new SearchKeyRenameRequest
                {
                    IndexName = indexName
                }.Excute();
            }
        }
    }

### 业务API接口 ###

**批量新增接口**这里有2个细节点：

1.在第一次有数据进来的时候需要创建Mapping，因为得对KeyName字段定义分词器，其余字段都可以使用AutoMap即可。

2.新创建的Index名称是精确到秒的 SearchKey-202112261121

/// <summary>
            /// 批量新增作品搜索列表(返回创建的indexName)
            /// </summary>
            /// <param name="request"></param>
            /// <returns></returns>
            [HttpPost]
            public ApiResult Post(SearchKeyPostRequest request)
            {
                if (!request.Items.Any())
                    return ApiResult.IsFailed("无传入数据");
    
                var date = DateTime.Now;
                var relationName = typeof(SearchKey).GetRelationName();
                var indexName = request.IndexName.IsNullOrWhiteSpace() ? (relationName + "-" + date.ToString("yyyyMMddHHmmss")) : request.IndexName;
    
                if (request.IndexName.IsNullOrWhiteSpace())
                {
                    var createResult = _elasticClient.Indices.Create(indexName,
                        a =>
                            a.Map<SearchKey>(m => m.AutoMap().Properties(p =>
                                p.Custom(new TextProperty
                                {
                                    Name = "key_name",
                                    Analyzer = "standard",
                                    Fields = new Properties(new Dictionary<PropertyName, IProperty>
                                    {
                                        { new PropertyName("pinyin"),new TextProperty{ Analyzer = "pinyin"} },
                                        { new PropertyName("standard"),new TextProperty{ Analyzer = "standard"} }
                                    })
                                }))));
    
                    if (!createResult.IsValid && request.IndexName.IsNullOrWhiteSpace())
                        return ApiResult.IsFailed("创建索引失败");
                }
                
                var document = request.Items.MapTo<List<SearchKey>>();
    
                var result = _elasticClient.BulkAll(indexName, document);
    
                return result ? ApiResult.IsSuccess(data: indexName) : ApiResult.IsFailed();
            }

**重新绑定别名接口**这里有4个细节点：

1.别名使用searchkey，只会有一个Index\[searchkey-yyyyMMddHHmmss\]会跟searchkey绑定.

2.优先把已绑定的Index查询出来，方便解绑与删除。

3.别名绑定在Elasticsearch虽然是原子性的，但是不是数据一致性的，因此得先Add后Remove。

4.删除旧得Index免得占用过多资源。

/// <summary>
            /// 重新绑定别名
            /// </summary>
            /// <returns></returns>
            [HttpPut]
            public ApiResult Rename(SearchKeyRanameRequest request)
            {
                var aliasName = typeof(SearchKey).GetRelationName();
                var getAliasResult = _elasticClient.Indices.GetAlias(aliasName);
    
                //给新index指定别名
                var bulkAliasRequest = new BulkAliasRequest
                {
                    Actions = new List<IAliasAction>
                    {
                        new AliasAddDescriptor().Index(request.IndexName).Alias(aliasName)
                    }
                };
    
                //移除别名里旧的索引
                if (getAliasResult.IsValid)
                {
                    var indeNameList = getAliasResult.Indices.Keys;
                    foreach (var indexName in indeNameList)
                    {
                        bulkAliasRequest.Actions.Add(new AliasRemoveDescriptor().Index(indexName.Name).Alias(aliasName));
                    }
                }
    
                var result = _elasticClient.Indices.BulkAlias(bulkAliasRequest);
    
                //删除旧的index
                if (getAliasResult.IsValid)
                {
                    var indeNameList = getAliasResult.Indices.Keys;
                    foreach (var indexName in indeNameList)
                    {
                        _elasticClient.Indices.Delete(indexName);
                    }
                }
    
                return result != null && result.ApiCall.Success ? ApiResult.IsSuccess() : ApiResult.IsFailed();
            }

**查询接口**这里跟前面细节得差不多：

但是这里有一个得**特别注意**的点，可以看到这个查询接口同时使用了should和must，这里得设置minimumShouldMatch才能正常像SQL过滤。

should可以理解成SQL的Or，Must可以理解成SQL的And。

默认情况下minimumShouldMatch是等于0的，等于0的意思是，**should不命中任何的数据仍然会返回must命中的数据**，也就是你们可能想搜索（*keyname.pinyin=’chengong‘ or keyname.standard=’chengong‘) and id > 0*，但是es里没有存keyname='chengong'的数据，会把id> 0 而且 keyname != 'chengong' 数据给查询出来。

因此我们得对minimumShouldMatch=1，就是**should**条件**必须得任意命中一个**才能返回结果。

**在should和must混用的情况下必须得注意minimumShouldMatch的设置！！！**

/// <summary>
            /// 作品搜索列表
            /// </summary>
            /// <param name="request"></param>
            /// <returns></returns>
            [HttpPost]
            [Route("search")]
            public ApiResult<List<SearchKeyGetResponse>> Get(SearchKeyGetRequest request)
            {
                var shouldQuerys = new List<Func<QueryContainerDescriptor<SearchKey>, QueryContainer>>();
                int minimumShouldMatch = 0;
                if (!request.KeyName.IsNullOrWhiteSpace())
                {
                    shouldQuerys.Add(a => a.MatchPhrase(m => m.Field("key_name.pinyin").Query(request.KeyName)));
                    shouldQuerys.Add(a => a.MatchPhrase(m => m.Field("key_name.standard").Query(request.KeyName)));
                    minimumShouldMatch = 1;
                }
    
                var mustQuerys = new List<Func<QueryContainerDescriptor<SearchKey>, QueryContainer>>
                {
                    a => a.Range(t => t.Field(f => f.Weight).GreaterThanOrEquals(0))
                };
    
                if (request.IsSubsidiary.HasValue)
                    mustQuerys.Add(a => a.Term(t => t.Field(f => f.IsSubsidiary).Value(request.IsSubsidiary.Value)));
    
                if (request.SysTagIds != null && request.SysTagIds.Any())
                    mustQuerys.Add(a => a.Terms(t => t.Field(f => f.SysTagId).Terms(request.SysTagIds)));
    
                if (request.EntityType.HasValue)
                {
                    if (request.EntityType.Value == ESearchKey.EntityType.AllNovel)
                    {
                        mustQuerys.Add(a => a.Terms(t => t.Field(f => f.EntityType).Terms(ESearchKey.EntityType.Novel, ESearchKey.EntityType.ChatNovel, ESearchKey.EntityType.FanNovel)));
                    }
                    else
                        mustQuerys.Add(a => a.Term(t => t.Field(f => f.EntityType).Value((int)request.EntityType.Value)));
                }
    
                var sortDescriptor = new SortDescriptor<SearchKey>();
                sortDescriptor = request.Sort == ESearchKey.Sort.Weight
                    ? sortDescriptor.Field(f => f.Weight, SortOrder.Descending)
                    : sortDescriptor.Field(f => f.ActiveDate, SortOrder.Descending);
    
                var searchResult = _elasticClient.Search<SearchKey>(a =>
                    a.Index(typeof(SearchKey).GetRelationName())
                        .From(request.Size * request.Page)
                        .Size(request.Size)
                        .Query(q => q.Bool(b => b.Should(shouldQuerys).Must(mustQuerys).MinimumShouldMatch(minimumShouldMatch)))
                        .Sort(s => sortDescriptor));
    
                var apiResult = searchResult.GetApiResult<SearchKey, List<SearchKeyGetResponse>>();
    
                if (apiResult.Success)
                    return apiResult;
    
                return ApiResult<List<SearchKeyGetResponse>>.IsSuccess("空集合数据");
            }

## APM监控 ##

在之前公司做微服务的时候的APM选型我们使用了Skywalking，但是现在这家公司的运维没有接触过，但是对于Elastic Stack他相对比较熟悉，如同上文所说**架构设计的输入核心为两点**：**满足需求**与**组织架构**，秉着我的技术选型原则是基于团队架构，我们采用了Elastic APM + Kibana（7.4版本），如下图所示：

![4e7df132a68f5e58377a7e458802f28a.png][]

## 结尾 ##

![7635e5c388ebaf56f669dfd1926e4aa7.png][]

**　　**最后上生产的时候也是平滑无损的切换到Elasticsearch，总体情况都十分满意。从上图可以见Sql Server的负载从40%-90%降低到了20%左右。

[6c732fe2ae205977ec9ff9264ce8d925.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/376b38e5ef484dd8a9793af4ab6a613c.png
[510833903091d56a67926ff91379fbee.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/b9b7b175e9504ded84a9038be1aacb2f.png
[8bbf11ad6f3d3cb68c49deff3c79b359.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/8a44998f89ff43e3a17a24c4245fcc43.png
[4e7df132a68f5e58377a7e458802f28a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/985f00af89a5495f88a9a76f27740a72.png
[7635e5c388ebaf56f669dfd1926e4aa7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/9e022269d89143dcab033ca645664a2d.png