文档中心

云原生消息引擎

连接器

创建连接器

在 MQTT Dashboard 页面，支持创建连接器。连接器可以作为 Sink 或 Source 的底层连接通道，用于连接到外部数据系统。本文档介绍创建 Kafka 生产者连接器和 MySQL 连接器的操作步骤。

背景信息

Kafka 生产者连接器：连接到 KafKa 实例或 BMQ 实例，使 KafKa 实例或 BMQ 实例 Topic 中的数据可以被 MQTT 消费。

MySQL 连接器：连接到 MySQL 数据库，将 MQTT 消息数据高效地存储至 MySQL 数据库中。

前提条件

在创建连接器前，您需要先创建待连接的外部数据系统。

Kafka 生产者连接器

已创建 Kafka 生产者（实例），且实例处于正常运行状态。创建实例的具体操作步骤，请参见：

创建 Kafka 实例

创建 BMQ 实例

MySQL 连接器

已创建 MySQL 实例、数据库和账号，具体操作步骤，请参见：

创建 MySQL 实例

创建数据库和账号

注意事项

连接的数据系统需与 MQTT 实例处于相同 VPC 内。

创建 Kafka 生产者连接器

说明

Kafka 生产者连接器用于将 MQTT 数据传输到 Kafka 或 BMQ，此场景下，Kafka 或 BMQ 作为消费者，MQTT 作为生产者。目前支持连接到 Kafka 实例或 BMQ 实例。

在 Dashboard 导航栏选择集成 > 连接器，然后单击页面右上角的创建。

选择连接器类型为 Kafka生产者，然后单击下一步。

配置连接器信息。

参数	说明
连接器名称	自定义设置连接器名称。名称输入规则：由字母、数字、中划线（-）、下划线（_）组成，只能以字母或数字开头。长度为 1~255 个字符。在当前 MQTT 实例中保持唯一。
描述	输入连接器的描述语句。
主机列表	输入需要连接的 Kafka 实例或 BMQ 实例的接入点地址。支持通过以下接入点连接目标 Kafka 实例或 BMQ 实例： Kafka 实例 PLAINTEXT 接入点、SASL_PLAINTEXT 接入点以及 SASL_SSL 接入点。 BMQ 实例用户 VPC 接入点。
认证	无连接到以下接入点时选择： Kafka PLAINTEXT 接入点 BMQ 用户 VPC 接入点基础认证连接到以下接入点时选择： Kafka SASL_PLAINTEXT 接入点 Kafka SASL_SSL 接入点
认证方法	选择基础认证方式时，需要配置认证方法。 plain 连接到 Kafka SASL_PLAINTEXT 接入点时使用。 scram_sha_256 连接到 Kafka SASL_PLAINTEXT 接入点时使用。
用户名	选择基础认证方式时需要配置，输入可登录目标接入点的 SASL 用户名。
密码	选择基础认证方式时需要配置，输入指定用户的密码。
Kerberos Principal	选择基础认证方式时，需要配置 Kerberos Principal。 SASL GSSAPI 认证方法的 Kerberos principal。例如 client_name@MY.KERBEROS.REALM.MYDOMAIN.COM。注意这里使用的 realm 需要配置在 EMQX 服务器的 /etc/krb5.conf 中。
Kerberos Keytab 文件	选择基础认证方式时，需要配置 Kerberos Keytab 文件。 SASL GSSAPI 认证方法的 Kerberos keytab 文件。注意该文件需要上传到 EMQX 服务器中，且运行 EMQX 服务的系统账户需要有读取权限。
启用 TLS	连接到 Kafka SASL_PLAINTEXT 接入点时开启，其他情况保持关闭即可。
验证服务器证书	在启用 TLS 后，默认开启验证服务器证书，请关闭该功能。

单击测试连接。

返回连接成功提示后，单击创建。

在弹出的提示框中，选择返回连接器列表或创建规则。

返回连接器列表：创建当前连接器并返回连接器页面。

创建规则：创建当前连接器并进入创建规则页面，创建规则的具体操作请参见创建规则。

创建 MySQL 连接器

说明

MySQL 连接器用于将 MQTT 消息高效地存储至 MySQL 数据库中。

在 Dashboard 导航栏选择集成 > 连接器，然后单击页面右上角的创建。

选择连接器类型为 MySQL，然后单击下一步。

配置连接器信息。

参数	说明
连接器名称	自定义设置连接器名称。名称输入规则：由字母、数字、中划线（-）、下划线（_）组成，只能以字母或数字开头。长度为 1~255 个字符。在当前 MQTT 实例中保持唯一。
描述	输入连接器的描述语句。
服务器地址	连接器的连接地址。输入需要连接 MySQL 实例的连接地址，格式为私网域名:端口。如果未指定端口，则使用 MySQL 默认端口 3306。
数据库名称	填写需要连接的数据库名称。
用户名	填写可访问 MySQL 实例的用户名。
密码	填写可访问 MySQL 实例的用户密码。
启用 TLS	配置是否启用 TLS 校验，以及是否校验服务器证书。如需校验，请分别配置 Cert 和 Key。
高级设置说明通常情况下，高级设置参数保持默认配置即可，您也可根据实际需求修改。
连接池大小	配置 MQTT 到指定 MySQL 数据库的并发连接数。默认值为 8，您可根据业务需求配置为其他整数值。
启动超时时间	配置等待 MySQL 数据库达到健康状态的最长时间间隔。默认为 5 秒。连接器在响应创建数据库连接请求之前，将在配置的时间间隔内持续检查目标数据库状态，等待数据库达到健康状态后，再执行连接和数据操作，例如插入、删除和更新数据，避免在数据库尚未完全启动时执行操作而引起的错误或数据不一致的问题。
健康检查间隔	配置 MQTT 对 MySQL 的连接执行自动健康检查的时间间隔，默认为 15 秒。

单击测试连接。

返回连接成功提示后，单击创建。

在弹出的提示框中，选择返回连接器列表或创建规则。

返回连接器列表：创建当前连接器并返回连接器页面。

创建规则：创建当前连接器并进入创建规则页面，创建规则的具体操作请参见创建规则。

最近更新时间：2025.08.04 17:30:02

这个页面对您有帮助吗？

有用

无用