每日一博 | 基于 Kafka 和 Elasticsearch 构建实时站内搜索功能的实践-低调大师

每日一博 | 基于 Kafka 和 Elasticsearch 构建实时站内搜索功能的实践

2021-02-16 568

目前我们在构建一个多租户多产品类网站，为了让用户更好的找到他们所需要的产品，我们需要构建站内搜索功能，并且它应该是实时更新的。本文将会讨论构建这一功能的核心基础设施，以及支持此搜索能力的技术栈。

问题的定义与决策

为了构建一个快速、实时的搜索引擎，我们必须做出某些设计决策。我们使用MySQL作为主数据库存储，因此有以下选择：

直接在MySQL数据库中查询用户在搜索框中输入的每个关键词，就像%#{word1}%#{word2}%...这样。 😐
使用一个高效的搜索数据库，如Elasticsearch。😮

考虑到我们是一个多租户应用程序，同时被搜索的实体可能需要大量的关联操作（如果我们使用的是MySQL一类的关系型数据库），因为不同类型的产品有不同的数据结构，所以我们还可以能需要同时遍历多个数据表来查询用户输入的关键词。所以我们决定不使用直接在MySQL中查询关键词的方案。🤯

因此，我们必须决定一种高效、可靠的方式，将数据实时地从MySQL迁移到Elasticsearch中。接下来需要做出如下的决定：

使用Worker定期查询MySQL数据库，并将所有变化的数据发送到Elasticsearch。😶
在应用程序中使用Elasticsearch客户端，将数据同时写入到MySQL和Elasticsearch中。🤔
使用基于事件的流引擎，将MySQL数据库中的数据更改作为事件，发送到流处理服务器上，经过处理后将其转发到Elasticsearch。🥳

选项1并不是实时的，所以可以直接排除，而且即使我们缩短轮询间隔，也会造成全表扫描给数据库造成查询压力。除了不是实时的之外，选项1无法支持对数据的删除操作，如果对数据进行了删除，那么我们需要额外的表记录之前存在过的数据，这样才能保证用户不会搜索到已经删除了的脏数据。对于其他两种选择，不同的应用场景做出的决定可能会有所不同。在我们的场景中，如果选择选项2，那么我们可以预见一些问题：如过Elasticsearch建立网络连接并确认更新时速度很慢，那么这可能会降低我们应用程序的速度；或者在写入Elasticsearch时发生了未知异常，我们该如何对这一操作进行重试来保证数据完整性；不可否认开发团队中不是所有开发人员都能了解所有的功能，如果有开发人员在开发新的与产品有关的业务逻辑时没有引入Elasticsearch客户端，那么我们将在Elasticsearch中更新这次数据的更改，无法保证MySQL与Elasticsearch间的数据一致性。

接下来我们该考虑如何将MySQL数据库中的数据更改作为事件，发送到流处理服务器上。我们可以在数据库变更后，在应用程序中使用消息管道的客户端同步地将事件发送到消息管道，但是这并没有解决上面提到的使用Elasticsearch客户端带来的问题，只不过是将风险从Elasticsearch转移到了消息管道。最终我们决定通过采集MySQL Binlog，将MySQL Binlog作为事件发送到消息管道中的方式来实现基于事件的流引擎。关于binlog的内容可以点击链接，在这里不再赘述。

服务简介

为了对外提供统一的搜索接口，我们首先需要定义用于搜索的数据结构。对于大部分的搜索系统而言，对用户展示的搜索结果通常包括为标题和内容，这部分内容我们称之可搜索内容（Searchable Content）。在多租户系统中我们还需要在搜索结果中标示出该搜索结果属于哪个租户，或用来过滤当前租户下可搜索的内容，我们还需要额外的信息来帮助用户筛选自己想要搜索的产品类别，我们将这部分通用的但不用来进行搜索的内容称为元数据（Metadata）。最后，在我们展示搜索结果时可能希望根据不同类型的产品提供不同的展示效果，我们需要在搜索结果中返回这些个性化展示所需要的原始内容（Raw Content）。到此为止我们可以定义出了存储到Elasticsearch中的通用数据结构：

{
	"searchable": {
		"title": "string",
		"content": "string"
	},
	"metadata": {
		"tenant_id": "long",
		"type": "long",
		"created_at": "date",
		"created_by": "string",
		"updated_at": "date",
		"updated_by": "string"
	},
	"raw": {}
}

基础设施

Apache Kafka: Apache Kafka是开源的分布式事件流平台。我们使用Apache kafka作为数据库事件（插入、修改和删除）的持久化存储。

mysql-binlog-connector-java: 我们使用mysql-binlog-connector-java从MySQL Binlog中获取数据库事件，并将它发送到Apache Kafka中。我们将单独启动一个服务来完成这个过程。

在接收端我们也将单独启动一个服务来消费Kafka中的事件，并对数据进行处理然后发送到Elasticsearch中。

Q：为什么不使用Elasticsearch connector之类的连接器对数据进行处理并发送到Elasticsearch中？
A：在我们的系统中是不允许将大文本存入到MySQL中的，所以我们使用了额外的对象存储服务来存放我们的产品文档，所以我们无法直接使用连接器将数据发送到Elasticsearch中。
Q：为什么不在发送到Kafka前就将数据进行处理？
A：这样会有大量的数据被持久化到Kafka中，占用Kafka的磁盘空间，而这部分数据实际上也被存储到了Elasticsearch。
Q：为什么要用单独的服务来采集binlog，而不是使用Filebeat之类的agent？
A：当然可以直接在MySQL数据库中安装agent来直接采集binlog并发送到Kafka中。但是在部分情况下开发者使用的是云服务商或其他基础设施部门提供的MySQL服务器，这种情况下我们无法直接进入服务器安装agent，所以使用更加通用的、无侵入性的C/S结构来消费MySQL的binlog。

配置技术栈

我们使用docker和docker-compose来配置和部署服务。为了简单起见，MySQL直接使用了root作为用户名和密码，Kafka和Elasticsearch使用的是单节点集群，且没有设置任何鉴权方式，仅供开发环境使用，请勿直接用于生产环境。

version: "3"
services:
  mysql:
    image: mysql:5.7
    container_name: mysql
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: app
    ports:
      - 3306:3306
    volumes:
      - mysql:/var/lib/mysql
  zookeeper:
    image: bitnami/zookeeper:3.6.2
    container_name: zookeeper
    ports:
      - 2181:2181
    volumes:
      - zookeeper:/bitnami
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: bitnami/kafka:2.7.0
    container_name: kafka
    ports:
      - 9092:9092
    volumes:
      - kafka:/bitnami
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.11.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
    volumes:
      - elasticsearch:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
volumes:
  mysql:
    driver: local
  zookeeper:
    driver: local
  kafka:
    driver: local
  elasticsearch:
    driver: local

在服务启动成功后我们需要为Elasticsearch创建索引，在这里我们直接使用curl调用Elasticsearch的RESTful API，也可以使用busybox基础镜像创建服务来完成这个步骤。

# Elasticsearch
curl "http://localhost:9200/search" -XPUT -d '
{
  "mappings": {
    "properties": {
      "searchable": {
        "type": "nested",
        "properties": {
          "title": {
            "type": "text"
          },
          "content": {
            "type": "text"
          }
        }
      },
      "metadata": {
        "type": "nested",
        "properties": {
          "tenant_id": {
            "type": "long"
          },
          "type": {
            "type": "integer"
          },
          "created_at": {
            "type": "date"
          },
          "created_by": {
            "type": "keyword"
          },
          "updated_at": {
            "type": "date"
          },
          "updated_by": {
            "type": "keyword"
          }
        }
      },
      "raw": {
        "type": "nested"
      }
    }
  }
}'

核心代码实现（SpringBoot + Kotlin）

Binlog采集端：

    override fun run() {
        client.serverId = properties.serverId
        val eventDeserializer = EventDeserializer()
        eventDeserializer.setCompatibilityMode(
            EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
        )
        client.setEventDeserializer(eventDeserializer)
        client.registerEventListener {
            val header = it.getHeader<EventHeader>()
            val data = it.getData<EventData>()
            if (header.eventType == EventType.TABLE_MAP) {
                tableRepository.updateTable(Table.of(data as TableMapEventData))
            } else if (EventType.isRowMutation(header.eventType)) {
                val events = when {
                    EventType.isWrite(header.eventType) -> mapper.map(data as WriteRowsEventData)
                    EventType.isUpdate(header.eventType) -> mapper.map(data as UpdateRowsEventData)
                    EventType.isDelete(header.eventType) -> mapper.map(data as DeleteRowsEventData)
                    else -> emptyList()
                }
                logger.info("Mutation events: {}", events)
                for (event in events) {
                    kafkaTemplate.send("binlog", objectMapper.writeValueAsString(event))
                }
            }
        }
        client.connect()
    }

在这段代码里面，我们首先是对binlog客户端进行了初始化，随后开始监听binlog事件。binlog事件类型有很多，大部分都是我们不需要关心的事件，我们只需要关注TABLE_MAP和WRITE/UPDATE/DELETE就可以。当我们接收到TABLE_MAP事件，我们会对内存中的数据库表结构进行更新，在后续的WRITE/UPDATE/DELETE事件中，我们会使用内存缓存的数据库结构进行映射。整个过程大概如下所示：

Table: ["id", "title", "content",...]
Row: [1, "Foo", "Bar",...]
=>
{
	"id": 1,
	"title": "Foo",
	"content": "Bar"
}

随后我们将收集到的事件发送到Kafka中，并由Event Processor进行消费处理。

事件处理器

@Component
class KafkaBinlogTopicListener(
    val binlogEventHandler: BinlogEventHandler
) {

    companion object {
        private val logger = LoggerFactory.getLogger(KafkaBinlogTopicListener::class.java)
    }

    private val objectMapper = jacksonObjectMapper()

    @KafkaListener(topics = ["binlog"])
    fun process(message: String) {
        val binlogEvent = objectMapper.readValue<BinlogEvent>(message)
        logger.info("Consume binlog event: {}", binlogEvent)
        binlogEventHandler.handle(binlogEvent)
    }
}

首先使用SpringBoot Message Kafka提供的注解对事件进行消费，接下来将事件委托到binlogEventHandler去进行处理。实际上BinlogEventHandler是个自定义的函数式接口，我们自定义事件处理器实现该接口后通过Spring Bean的方式注入到KafkaBinlogTopicListener中。

@Component
class ElasticsearchIndexerBinlogEventHandler(
    val restHighLevelClient: RestHighLevelClient
) : BinlogEventHandler {
    override fun handle(binlogEvent: BinlogEvent) {
        val payload = binlogEvent.payload as Map<*, *>
        val documentId = "${binlogEvent.database}_${binlogEvent.table}_${payload["id"]}"
        // Should delete from Elasticsearch
        if (binlogEvent.eventType == EVENT_TYPE_DELETE) {
            val deleteRequest = DeleteRequest()
            deleteRequest
                .index("search")
                .id(documentId)
            restHighLevelClient.delete(deleteRequest, DEFAULT)
        } else {
            // Not ever WRITE or UPDATE, just reindex
            val indexRequest = IndexRequest()
            indexRequest
                .index("search")
                .id(documentId)
                .source(
                    mapOf<String, Any>(
                        "searchable" to mapOf(
                            "title" to payload["title"],
                            "content" to payload["content"]
                        ),
                        "metadata" to mapOf(
                            "tenantId" to payload["tenantId"],
                            "type" to payload["type"],
                            "createdAt" to payload["createdAt"],
                            "createdBy" to payload["createdBy"],
                            "updatedAt" to payload["updatedAt"],
                            "updatedBy" to payload["updatedBy"]
                        )
                    )
                )
            restHighLevelClient.index(indexRequest, DEFAULT)
        }
    }
}

在这里我们只需要简单地判断是否为删除操作就可以，如果是删除操作需要在Elasticsearch中将数据删除，而如果是非删除操作只需要在Elasticsearch重新按照为文档建立索引即可。这段代码简单地使用了Kotlin中提供的mapOf方法对数据进行映射，如果需要其他复杂的处理只需要按照Java代码的方式编写处理器即可。

总结

在本文是作者工作后编写的第一篇博客，才学疏浅，可能有的地方会出现技术理解不够深入用词不当等问题，感谢大家帮忙指正。本文中提到的完整代码已上传到Gitee，欢迎给个Star～传送门

微信关注我们

原文链接：https://my.oschina.net/u/4929341/blog/4952996

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Go 1.16 发布，支持 macOS ARM64

Go 1.16 正式发布，该版本主要包括支持macOS ARM64、工具链优化、运行时和库的实现优化。并且，该版本保留了Go 1兼容性的承诺，几乎所有 Go 程序都能像以前一样继续编译和运行。部分更新内容 Go 1.16 通过 GOOS=darwin，GOARCH=arm64 在macOS（也称为 Apple Silicon）上添加了对 64 位 ARM 架构的支持。与 darwin/amd64 一样，darwin/arm64 也支持 cgo、内部和外部链接、c-archive、c-shared 和 pie build 模式、竞争检测器 Go 现在在 NetBSD（netbsd/arm64）上支持 64 位 ARM 体系架构 Go 现在在 OpenBSD（openbsd/mips64）上支持 MIPS64 体系架构，但尚不支持 cgo 放弃了对 x87 模式编译（GO386=387）的支持。现在可以使用软浮点模式（GO386=softfloat）来支持非 SSE2 处理器 linux/riscv64 现在支持 cgo 和 -buildmode=pie GO111MODULE 环境变量...

2021-02-17

676

VLC 4.0 最新的每晚构建版本已经可以体验新 UI，以下截图来自nightly builds的20210212-0431版本。从播放器到媒体浏览器 VLC 4.0 新 UI 最明显的变化是启动后的初始界面。在较早之前的版本中，启动后打开的是一个空白的播放器窗口，中间显示 VLC 的标志性交通锥 logo，顶部菜单栏提供了各种功能控件，整体界面是十分传统的播放器样式。新版 VLC 启动后则是打开媒体浏览界面的“Video”选项卡，显示视频文件夹中存在的所有视频缩略图，顶部菜单栏也不再保留此前的各项功能控件。下边是新旧版本启动后初始界面的对比图。除了 Video，新版 UI 的顶部菜单栏还加入了 Music、Browse 和 Discover 选项卡，"Music"选项卡为电脑上的音乐文件夹提供了与“Video”类似的视图，"Discover"提供了一个网络浏览器，以查找用户 LAN 上存在的共享和流，此功能似乎尚未完全实现。上边介绍了新版 VLC 4.0 启动后的初始界面是一个媒体浏览器，因此如果播放视频或其他媒体文件，会产生一个新的独立播放器窗口，并与选择视频的浏览/控制窗...

2021-02-17

656

资源下载

更多资源

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。