消息队列Kafka「检索组件」上线

简介：本文对消息队列 Kafka「检索组件」进行详细介绍，首先通过对消息队列使用过程中的痛点问题进行介绍，然后针对痛点问题提出相应的解决办法，并对关键技术技术进行解读，旨在帮助大家对消息队列 Kafka「检索组件」的特点及使用方式更加熟悉，以期可以帮助大家更有效的解决在消息排查过程中遇到的痛点问题。

作者：Kafka&Tablestore团队

前言

还在为消息队列使用时，不能高效排查重复和失败的消息而困扰吗？

还在为消息队列使用时，无法准确查找消息内容和定位问题而苦恼吗？

。。。

消息队列 Kafka「检索组件」来帮您～

本文对消息队列 Kafka「检索组件」进行详细介绍，首先通过对消息队列使用过程中的痛点问题进行介绍，然后针对痛点问题提出相应的解决办法，并对关键技术技术进行解读，旨在帮助大家对消息队列 Kafka「检索组件」的特点及使用方式更加熟悉，以期可以帮助大家更有效的解决在消息排查过程中遇到的痛点问题。

痛点问题介绍

在消息队列的使用过程中，业内默认的是假设消息进入消息队列后，消息是可靠的，丢失的概率也是低的。但实际应用中会面临各种各样的问题：

应用时面临的痛点问题

由于分布式系统的特性，消息的失败、重复是不可避免的，对于失败和重复的排查，通常是依靠客户端的日志来推导，但如果规模庞大，客户手动做这个事情的难度也会很大，这就会使消息的可靠性受到挑战；
此外，较大的项目一般由多人或多团队协作完成，消息发送和消费的代码实现方式也各异，这会给消息最终是否成功完成使命带来挑战；
除了对问题结果的排查外，消息会不会在产生时就不符合预期呢？这同样也是困扰客户的难点之一。从目前消息队列的体系来看，还无法提供按照内容查看的方式来排查，导致了业务的正确性排查难度较大。

简单来说，消息领域往往每条消息都能代表具体的含义和动作，一旦出现失败、丢失和错误，在业内现有的消息队列现状下，很难排查具体问题，从而会导致定位整个上下游链路的问题难度较大。

技术侧面临的痛点问题

以上是客户在消息应用的场景中会面临的问题。基于应用场景问题，在技术侧同样会面临不少痛点，在处理消息问题排查时：

首先需要研发的代码投入、部署和运维，同时运维人员需要比较熟悉 Kafka 的使用，需要通过使用 Kafka 客户端进行消费者消费，然后按照遍历的方式进行消息确认，从而确认消息的存在；
除了需要研发的代码投入、部署和运维外，可能还需要引入其他产品，如对接流计算，通过流计算遍历消息等。

更为麻烦的是，目前这种排查往往是非常频繁的，经常以周、甚至以天为单位，会使得研发、部署和运维投入较高的时间成本；同时每次需要确认的元信息都不一样，会导致投入较大，而且灵活性也不高。

总结来说，消息队列在使用过程中对失败和重复等问题排查时，一来在没有较好的工具和方式完成对内容的检索，排查难度较高，准确性和易用性都不足；二来需要投入较高的时间和人力成本，投入大且不灵活。这些问题都会给用户在进行消息问题排查时带来不少困扰。

Kafka 检索组件介绍

通过上述痛点问题的介绍可以看到，目前在消息领域，对消息排查等存在比较多的痛点问题，为了解决以上问题，阿里云消息队列 Kafka 版重磅推出消息检索组件。下面对组件内容进行详细介绍：

检索组件简介

消息队列 Kafka「检索组件」是一个全托管、高弹性、交互式的检索组件，具备万亿级别消息内容检索的秒级响应能力。

主要面向运维人员故障排查和恢复场景，用于消息相关的全链路消息排查，包括消息的发送、重复生产和丢失校验；主要功能包括支持消息按 Topic 分区、位点范围和时间范围检索，同时支持按消息 Key 和 Value 关键字检索等；
主要用来解决业内消息产品不支持检索消息内容的难题。

消息队列 Kafka「消息检索」借助 Kafka Connect 功能及表格存储（Tablestore）实现，通过 Connector 对 Topic 中的消息进行转储，然后发送到表格存储中的数据表中，最后通过表格存储索引功能提供消息检索的能力。

其核心是提供了完备的消息内容检索能力，可以快速定位问题，同时便捷操作、节省人力；当用户使用时，在完成消息队列 Kafka 实例创建后，仅需简单五步即可实现对 Kafka 检索组件的应用：

下面简要对消息队列 Kafka 版消息检索的操作步骤进行介绍。

检索组件操作介绍

1）开通消息检索

首先开通某个实例下 Topic 的消息检索功能，以便根据需要对其 Topic 中的消息进行检索。步骤如下：

登录消息队列 Kafka 版控制台；
在概览页面的资源分布区域，选择地域；
在左侧导航栏，单击消息检索；
在消息检索页面，从选择实例的下拉列表选择需检索 Topic 消息所属的实例，然后单击开通消息检索；
开通消息检索面板，填写开通参数，然后单击确定。

2）测试发送消息

开通消息检索后，可以向消息队列 Kafka 版的数据源 Topic 发送消息，以此来调度任务和测试消息检索是否创建成功。

在消息检索页面，找到需要测试的目标 Topic，根据任务状态在对应位置操作；
在快速体验消息收发面板中发送测试消息。

3）搜索消息

在消息检索页面，找到目标 Topic，在其操作列，单击搜索；
在搜索面板，设置搜索条件，在搜索项下拉列表中选择需要添加的搜索项，单击添加搜索项，添加搜索项并在值列设置搜索信息，然后单击确定即可。

4）查看消息检索任务详情

开通消息检索后，即可查看自动创建的 Topic、Group、表格存储实例名称、表格存储数据表表名等详细信息，也可以在详情中直接进入表格存储数据表；
在消息检索页面，找到目标 Topic，在其操作列，单击详情；
在任务详情页面可以查看到目标 Topic 相关消息检索的详细信息；也可以在基础信息区域的目标服务栏，单击表格存储，即可进入数据表详情页面查看。

5）查看消费详情

支持查看订阅当前 Topic 的在线 Group 在 Topic 各个分区的消费进度，了解消息的消费和堆积情况。

在消息检索页面，找到需要查看消费进度的目标 Topic，在其操作列，单击消费进度；
如下图，在消费详情页面，可以查看 Topic 各分区的消费情况：

除以上功能外，在运行消息检索功能时，还可以实现暂停消息检索任务、启用消息检索任务和删除消息检索任务等操作。

Kafka 检索组件技术解读

之前消息队列 Kafka 版的消息检索方式仅支持根据消费位点或创建时间的两种范围来查找，依靠 Kafka 系统本身无法很好的支持用户对于通过关键字检索消息的需求。

为了更好的解决这个问题，Kafka 与 Tablestore 强强联合，将 Kafka 消息通过 Connector 导入 Tablestore 的数据表中，利用 Tablestore 的能力实现关键字检索。

下面对关键技术进行解读：

Kafka Connect

Kafka Connect 的核心是为解决异构数据的同步问题。解决的思路是在各个数据源之间加一层消息中间件，所有的数据都经过消息中间件进行存储和分发。

这样做的好处有以下两点：

1）通过消息中间件做异步解耦，所有系统只和消息中间件通信；

2）需要开发的解析工具数量，也从原来的 n 平方个，变成线性的 2*n 个；Kafka Connect 则用于连接消息系统和数据源，根据数据的流向不同，连接可以分为 Source Connector 和 Sink Connector。

其原理也很简单，Source Connector 负责解析来源数据，转换成标准格式的消息，通过 Kafka Producer 发送到 Kafka Broker中。同理，Sink Connector 则通过 Kafka Consumer 消费对应的 Topic，然后投递到目标系统中。在整个过程中，Kafka Connect 统一解决了任务调度、与消息系统交互、自动扩缩容、容错以及监控等问题，大大减少了重复劳动。

消息队列 Kafka 版提供了全托管、免运维的 Kafka Connect，用于消息队列 Kafka 版和其他阿里云服务之间的数据同步。如下图所示，可以看到消息队列 Kafka 版支持了表格存储 Tablestore、Mysql Source Connector、OSS Sink Connector、MaxCompute Sink Connector 以及 FC Sink Connector 等主流的 Connector。如果用户想要使用这些 Connector 进行数据同步，只用在消息队列 Kafka 控制台的图形界面上做几个配置，就可以一键拉起 Connector 任务。

表格存储 Tablestore

表格存储 Tablestore 是构建在阿里云飞天分布式系统之上的海量结构化数据存储服务。基于飞天盘古分布式文件系统作为存储底座，采用存储计算分离架构，弹性共享资源池设计，实现了一个云原生的 Serverless 存储产品。内置分布式索引系统，可根据写入流量自动扩展构建索引所需的计算资源，支撑极高的写入流量。同时优化了索引结构，能够支持更快速的模糊查询。存算分离架构、高吞吐实时索引等关键能力让 Tablestore 能够支撑 Kafka 中海量数据的写入与高效搜索，帮助快速有效检索所需信息。

技术领先性

Kafka+Kafka Connect+ 表格存储 Tablestore 的云原生数据应用解决方案，通过 Kafka Connect 作为实时处理任务触发器，能够实时接收到新发送到消息队列集群的数据，然后转发到表格存储 Tablestore。

作为后续数据流转中的一环，Kafka Connect 除了保障数据的实时性以外，还解决了任务调度、与消息系统交互、自动扩缩容、容错以及监控等问题，大大减少了重复劳动。数据到了表格存储 Tablestore 以后，借助表格存储的分布式存储和强大的索引引擎，能够支持 PB 级存储、千万 TPS 和毫秒级延迟的服务能力，同时支持全托管、高弹性、交互式的检索组件，从而可实现秒级响应的万亿级别消息内容检索能力。