专属云Kafka的产品服务与公有云Kafka的产品规格保持一致，由两种服务资源组成，队列规格、队列存储。 其中队列计算规格按消息队列基准带宽分为4种分别是：100MB/s、300MB/s、600MB/s、1200MB/s； 存储类型分2种，分别如下：高I/O、超高I/O。

专属云Kafka包括2个计费项：专属计费资源、专属存储（可选）。

对于包周期订购后，支持通过控制台进行续订操作。 具体操作如下： 步骤1：登录Kafka专享版控制台，在实例列表中，右侧的“更多”菜单下拉选择“续费”按钮； 步骤2：进入续订管理界面后，选择续订周期后，点击“确认提交”，按提示完成后续支付流程。

Kafka作为一款热门的消息队列中间件，具备高效可靠的消息异步传递机制，主要用于不同系统间的数据交流和传递，在企业解决方案、金融支付、电信、电子商务、社交、即时通信、视频、物联网、车联网等众多领域都有广泛应用。 异步通信 将业务中属于非核心或不重要的流程部分，使用消息异步通知的方式发给目标系统，这样主业务流程无需同步等待其他系统的处理结果，从而达到系统快速响应的目的。 如网站的用户注册场景，在用户注册成功后，还需要发送注册邮件与注册短信，这两个流程使用Kafka消息服务通知邮件发送系统与短信发送系统，从而提升注册流程的响应速度。 图 串行发送注册邮件与短信流程 图 借助消息队列异步发送注册邮件与短信流程 错峰流控与流量削峰 在电子商务系统或大型网站中，上下游系统处理能力存在差异，处理能力高的上游系统的突发流量可能会对处理能力低的某些下游系统造成冲击，需要提高系统的可用性的同时降低系统实现的复杂性。电商大促销等流量洪流突然来袭时，可以通过队列服务堆积缓存订单等信息，在下游系统有能力处理消息的时候再处理，避免下游订阅系统因突发流量崩溃。消息队列提供亿级消息堆积能力，3天的默认保留时长，消息消费系统可以错峰进行消息处理。 另外，在商品秒杀、抢购等流量短时间内暴增场景中，为了防止后端应用被压垮，可在前后端系统间使用Kafka消息队列传递请求。 图 消息队列应对秒杀大流量场景 日志同步 在大型业务系统设计中，为了快速定位问题，全链路追踪日志，以及故障及时预警监控，通常需要将各系统应用的日志集中分析处理。 Kafka设计初衷就是为了应对大量日志传输场景，应用通过可靠异步方式将日志消息同步到消息服务，再通过其他组件对日志做实时或离线分析，也可用于关键日志信息收集进行应用监控。 日志同步主要有三个关键部分：日志采集客户端，Kafka消息队列以及后端的日志处理应用。 1. 日志采集客户端，负责用户各类应用服务的日志数据采集，以消息方式将日志“批量”“异步”发送Kafka客户端。 Kafka客户端批量提交和压缩消息，对应用服务的性能影响非常小。 2. Kafka将日志存储在消息文件中，提供持久化。 3. 日志处理应用，如Logstash，订阅并消费Kafka中的日志消息，最终供文件搜索服务检索日志，或者由Kafka将消息传递给Hadoop等其他大数据应用系统化存储与分析。 图 日志同步示意图 上图中Logstash、ElasticSearch分别为日志分析和检索的开源工具，Hadoop表示大数据分析系统。

接口功能介绍 查询证书列表。 接口约束 无 URI GET /v3/api/ssl/cert/list 路径参数 无 Query参数 参数 是否必填 参数类型 说明 示例 下级对象 certId 否 String 证书ID 123 certName 否 String 证书名称 myCert type 否 String 证书类型 SERVER pageNum 否 Integer 分页参数，页索引，默认为1 1 pageSize 否 Integer 分页参数，页大小，默认为10 10 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 实例的资源池ID。 bb9fdb42056f11eda1610242ac110002 请求体body参数 无 响应参数 参数 参数类型 说明 示例 下级对象 statusCode Integer 调用结果状态 error String 错误码 message String 结果描述 returnObj Object 返回数据 returnObj 表 returnObj 参数 参数类型 说明 示例 下级对象 pageNum Integer 分页参数，页索引 1 pageSize Integer 分页参数，页大小 10 totalCount Integer 返回数据条目总数 100 result Array of Objects 返回数据列表 result 表 result 参数 参数类型 说明 示例 下级对象 certId String 证书id 123 certName String 证书名称 myCert remark String 证书备注 test validFrom String 证书有效期起始日期 20250101 00:00:00 validEnd String 证书有效期结束日期 20280101 00:00:00 uploadTime String 证书上传时间 20240101 00:00:00 sn String 证书SN序列号 123 certType String 证书类型 SERVER encryptor String 私钥加密类型 RSA subject String 证书主体 lollipop.inc

为什么实例中存在默认名为trace和consumeroffsets的Topic？ 问题现象： Kafka Manager中存在默认名为trace和consumeroffsets的Topic。 处理方法： trace和consumeroffsets是Kafka实例内部预留的Topic，不建议删除这两个Topic，删除后可能导致实例无法使用。

请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body { "prodInstId": "445d30ce3d5d40a2a79dc444a8c83859", "data": [ { "topics": "t/a,t/b,t/c", "clientid": "mqtta", "qos": 0, "payload": "test messsage" } ] } 响应示例 { "statusCode":800 } 状态码 请参考 状态码 错误码 请参考 错误码

状态码 请参考 状态码 错误码 请参考 错误码

响应示例 { "statusCode": "800", "returnObj": { "prodInstId": "445d30ce3d5d40a2a79dc444a8c83859", "tpm": 10 } } 状态码 请参考 状态码 错误码 请参考 错误码

接口功能介绍 创建证书。 接口约束 无 URI POST /v3/api/ssl/cert/create 路径参数 无 Query参数 无 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 实例的资源池ID。 bb9fdb42056f11eda1610242ac110002 请求体body参数 参数 是否必填 参数类型 说明 示例 下级对象 certName 是 String 证书名称 myCert type 是 String 证书类型 CLIENTCA certFile 是 String 证书内容 123 certKey 否 String 证书配套私钥（仅服务端证书需要） 123 remark 否 String 证书备注 testcert encryptor 是 String 私钥加密方式 RSA 响应参数 参数 参数类型 说明 示例 下级对象 statusCode Integer 调用结果状态 error String 错误码 message String 结果描述 returnObj Object 返回数据 returnObj 表 returnObj 参数 参数类型 说明 示例 下级对象 certId String 新创建证书id 123 枚举参数 参数名 type 枚举值 类型 说明 CLIENTCA String 客户端CA证书 SERVER String 服务端证书 参数名 encryptor 枚举值 类型 说明 RSA String RSA

枚举参数 参数名 certSrc 枚举值 类型 说明 CUSTOMIZE String 自定义证书 DEFAULT String 服务端默认证书 参数名 authType 枚举值 类型 说明 DI String 单向认证 BI String 双向认证 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body { "certSrc": "CUSTOMIZE", "authType": "BI", "serverCert": { "certId": "xxx" }, "clientCA": { "certId": "yyy" } } 响应示例 { "statusCode":800 } 状态码 请参考 状态码 错误码 请参考 错误码

本文主要介绍应用场景。 Kafka作为一款热门的消息队列中间件，具备高效可靠的消息异步传递机制，主要用于不同系统间的数据交流和传递，在企业解决方案、金融支付、电信、电子商务、社交、即时通信、视频、物联网、车联网等众多领域都有广泛应用。 异步通信 将业务中属于非核心或不重要的流程部分，使用消息异步通知的方式发给目标系统，这样主业务流程无需同步等待其他系统的处理结果，从而达到系统快速响应的目的。 如网站的用户注册场景，在用户注册成功后，还需要发送注册邮件与注册短信，这两个流程使用Kafka消息服务通知邮件发送系统与短信发送系统，从而提升注册流程的响应速度。 图 串行发送注册邮件与短信流程 图 借助消息队列异步发送注册邮件与短信流程 错峰流控与流量削峰 在电子商务系统或大型网站中，上下游系统处理能力存在差异，处理能力高的上游系统的突发流量可能会对处理能力低的某些下游系统造成冲击，需要提高系统的可用性的同时降低系统实现的复杂性。电商大促销等流量洪流突然来袭时，可以通过队列服务堆积缓存订单等信息，在下游系统有能力处理消息的时候再处理，避免下游订阅系统因突发流量崩溃。消息队列提供亿级消息堆积能力，3天的默认保留时长，消息消费系统可以错峰进行消息处理。 另外，在商品秒杀、抢购等流量短时间内暴增场景中，为了防止后端应用被压垮，可在前后端系统间使用Kafka消息队列传递请求。 图 消息队列应对秒杀大流量场景

方向 协议 端口 源地址 说明 入方向 TCP 9094 0.0.0.0/0 通过公网访问Kafka（关闭SSL加密）。 入方向 TCP 9095 0.0.0.0/0 通过公网访问Kafka（开启SSL加密）。

参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID targetNodeNum 是 Integer 目标实例的代理数，取值为4到9，9节点不可以再进行扩容

参数 参数类型 说明 示例 下级对象 message String 描述状态 statusCode String 800成功 其他失败 error String 错误码 returnObj Object 返回对象

扩容过程中涉及数据迁移吗？ 扩容过程中不会迁移数据。 扩容/缩容过程中生产消息失败 可能原因： 在扩容/缩容代理规格的过程中，代理采用滚动重启的方式进行实例变更。在重启过程中，分区Leader会进行切换，此时生产客户端的元数据缓存中保存的分区Leader为旧分区Leader ID，仍然会向旧分区Leader发送消息，导致生产消息失败。 解决方法： 在生产客户端配置重试机制。建议设置retriesInteger.MAXVALUE 扩容失败提示资源不足 问题现象： 扩容失败，提示底层ECS/EVS资源不足，但在ECS控制台可以购买对应规格的ECS 可能原因： 底层资源配额与控制台界面显示的可用规格配额存在差异 解决方法： 联系客服增加配额数

参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池id

接口功能介绍 关闭设备轨迹 接口约束 无 URI POST /v3/api/stopClientTrack 路径参数 无 Query参数 无 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池编码 bb9fdb42056f11eda1610242ac110002 华东1 请求体body参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID 445d30ce3d5d40a2a79dc444a8c83859 响应参数 参数 参数类型 说明 示例 下级对象 message String 描述状态 statusCode String 800成功 其他失败 error String 错误码 returnObj Object 返回对象 枚举参数 无 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body { "prodInstId":"445d30ce3d5d40a2a79dc444a8c83859" } 响应示例 { "statusCode": 800 } 状态码 请参考 状态码 错误码 请参考 错误码

响应示例 { "statusCode": 800 } 状态码 请参考 状态码 错误码 请参考 错误码

参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 实例的资源池ID。 bb9fdb42056f11eda1610242ac110002

本章节主要介绍如何连接未开启SSL的RabbitMQ实例。 RabbitMQ实例兼容开源协议，请参考RabbitMQ官网提供的不同语言的连接和使用向导：< 介绍VPC内访问与使用RabbitMQ方法，假设RabbitMQ客户端部署在弹性云主机上。 前提条件 参考创建实例章节创建RabbitMQ实例，并记录创建时输入的用户名和密码。 建完成后，单击实例名称，查看并记录实例详情中的“连接地址”。 已创建弹性云主机，并且弹性云主机的VPC、子网、安全组与RabbitMQ实例的VPC、子网、安全组保持一致。 已完成JDK安装及环境变量配置。 命令行模式连接实例 以下操作命令以Linux系统为例进行说明： 1、下载RabbitMQTutorial.zip示例工程代码。 2、解压RabbitMQTutorial.zip压缩包。 $ unzip RabbitMQTutorial.zip 3、进入RabbitMQTutorial目录，该目录下包含预编译好的jar文件。 $ cd RabbitMQTutorial 4、运行生产消息示例。 $ java cp .:rabbitmqtutorial.jar Send host port user password 其中，host表示RabbitMQ实例的连接地址，port为RabbitMQ实例的监听端口（默认为5672），user表示RabbitMQ用户名，password表示用户名对应的密码。 图1 运行生产消息示例 使用Ctrl+C命令退出。 5、运行消费消息示例。 $ java cp .:rabbitmqtutorial.jar Recv host port user password 其中，host表示RabbitMQ实例的连接地址，port为RabbitMQ实例的监听端口（默认为5672），user表示RabbitMQ用户名，password表示用户名对应的密码。 图2 运行消费消息示例 如需停止消费使用Ctrl+C命令退出。

本文主要 Kafka客户端参数配置建议。 Kafka客户端的配置参数很多，以下提供producer和consumer几个常用参数配置。其他参数配置，请参考Kafka配置。 表 Producer参数 参数 默认值 推荐值 说明 acks 1 高可靠：all或者1高吞吐：1 收到Server端确认信号个数，表示producer需要收到多少个这样的确认信号，算消息发送成功。 acks参数代表了数据备份的可用性。常用选项：acks0：表示producer不需要等待任何确认收到的信息，副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为1。 acks1：这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。如果follower没有成功备份数据，而此时leader又无法提供服务，则消息会丢失。 acksall或者1：这意味着leader需要等待ISR中所有备份都成功写入日志。只要任何一个备份存活，数据都不会丢失。min.insync.replicas指定必须确认写入才能被认为成功的副本的最小数量。 retries 0 结合实际业务调整 客户端发送消息的重试次数。值大于0时，这些数据发送失败后，客户端会重新发送。 注意，这些重试与客户端接收到发送错误时的重试没有什么不同。允许重试将潜在的改变数据的顺序，如果这两个消息记录都是发送到同一个partition，则第一个消息失败第二个发送成功，则第二条消息会比第一条消息出现要早。针对网络闪断场景，生产者建议配置重试能力，推荐重试次数retries3，重试间隔retry.backoff.ms1000。 request.timeout.ms 30000 结合实际业务调整 设置一个请求最大等待时间，超过这个时间则会抛Timeout异常。超时时间如果设置大一些，如127000（127秒），高并发的场景中，能减少发送失败的情况。 block.on.buffer.full TRUE TRUE TRUE表示当我们内存用尽时，停止接收新消息记录或者抛出错误。默认情况下，这个设置为TRUE。 然而某些阻塞可能不值得期待，因此立即抛出错误更好。如果设置为false，则producer抛出一个异常错误：BufferExhaustedException batch.size 16384 262144 默认的批量处理消息字节数上限。producer将试图批处理消息记录，以减少请求次数。 这将改善client与server之间的性能。不会试图处理大于这个字节数的消息字节数。 发送到brokers的请求将包含多个批量处理，其中会包含对每个partition的一个请求。较小的批量处理数值比较少用，并且可能降低吞吐量（0则会仅用批量处理）。 较大的批量处理数值将会浪费更多内存空间，这样就需要分配特定批量处理数值的内存大小。 buffer.memory 33554432 67108864 producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常，以“block.on.buffer.full”来表明。 这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。 表 Consumer参数 参数 默认值 推荐值 说明 :::: auto.commit.enable TRUE FALSE 如果为真，consumer所fetch的消息的offset将会自动的同步到zookeeper。这项提交的offset将在进程无法提供服务时，由新的consumer使用。约束：设置为false后，需要先成功消费再提交，这样可以避免消息丢失。 auto.offset.reset latest earliest 没有初始化offset或者offset被删除时，可以设置以下值： earliest：自动复位offset为最早 latest：自动复位offset为最新 none：如果没有发现offset则向消费者抛出异常 anything else：向消费者抛出异常。 说明 如果将此配置设置为latest，新增分区时，生产者可能会在消费者重置初始偏移量之前开始向新增加的分区发送消息，从而导致部分消息丢失。 connections.max.idle.ms 600000 30000 空连接的超时时间，设置为30000可以在网络异常场景下减少请求卡顿的时间。

安装依赖 plaintext pip3 install pahomqtt 代码示例 python import ssl import random from paho.mqtt import client as pahoclient 填入您在mqtt控制台创建的ACL账号密码。 USERNAME "yourusername" AUTHPASSWORD "yourpassword" 是否使用tls加密传输 isTls True class MQTTClient: def init(self): 填写mqtt云消息服务的接入点，去掉:{端口号}部分 self.broker "tcp://localhost" 指定连接客户端的id self.clientid "ctgmqttclienttest" self.client None def onconnect(self, client, userdata, flags, rc): 连接建立成功 if rc 0: print("Connected to MQTT Broker!") else: print(f"Failed to connect, return code {rc}") def ondisconnect(self, client, userdata, rc): 连接丢失 print(f"Disconnected with result code {rc}") 可以在这里添加重连逻辑 def onmessage(self, client, userdata, msg): 收到消息的回调 print(f"Received {msg.payload.decode()} from {msg.topic} topic") def onpublish(self, client, userdata, mid): 成功发送消息到服务端 print(f"Message {mid} published successfully") def connectmqtt(self): 创建客户端实例 self.client pahoclient.Client(clientidself.clientid) self.client.usernamepwset(USERNAME, AUTHPASSWORD) 设置回调函数 self.client.onconnect self.onconnect self.client.ondisconnect self.ondisconnect self.client.onmessage self.onmessage self.client.onpublish self.onpublish 设置TLS if isTls: 创建不验证证书的SSL上下文 context ssl.createdefaultcontext() context.checkhostname False context.verifymode ssl.CERTNONE self.client.tlssetcontext(context) 设置其他连接选项 self.client.connect(self.broker.replace("tcp://", "").replace("ssl://", ""), port8883 if isTls else 1883) self.client.loopstart()

messages { height: 300px; border: 1px solid ccc; padding: 10px; overflowy: auto; marginbottom: 10px; } .message { marginbottom: 5px; padding: 5px; background: f5f5f5; } .controls { marginbottom: 20px; } input, button { padding: 8px; marginright: 5px; } 浏览器 MQTT 客户端 连接 断开 订阅 发布 // 全局变量 let client null; // DOM 元素 const connectBtn document.getElementById('connectBtn'); const disconnectBtn document.getElementById('disconnectBtn'); const subscribeBtn document.getElementById('subscribeBtn'); const publishBtn document.getElementById('publishBtn'); const messagesDiv document.getElementById('messages'); // 添加消息到界面 function addMessage(text, isError false) { const messageDiv document.createElement('div'); messageDiv.className 'message'; messageDiv.style.color isError ? 'red' : 'black'; messageDiv.textContent [${new Date().toLocaleTimeString()}] ${text}; messagesDiv.appendChild(messageDiv); messagesDiv.scrollTop messagesDiv.scrollHeight; } // 连接MQTT服务器 connectBtn.addEventListener('click', () > { const brokerUrl document.getElementById('brokerUrl').value; const username document.getElementById('username').value; const password document.getElementById('password').value; if (!brokerUrl) { addMessage('请输入MQTT服务器地址', true); return; } // 连接选项 const options { username: username undefined, password: password undefined // 对于WebSocket连接，路径通常包含在URL中 // 例如: wss://broker.example.com/mqtt // 如果需要单独设置: // wsOptions: { // path: '/mqtt' // }, }; addMessage(正在连接到: ${brokerUrl}); try { client mqtt.connect(brokerUrl, options); // 连接成功 client.on('connect', () > { addMessage('连接成功'); connectBtn.disabled true; disconnectBtn.disabled false; subscribeBtn.disabled false; publishBtn.disabled false; }); // 接收消息 client.on('message', (topic, message) > { addMessage(收到消息 [${topic}]: ${message.toString()}); }); // 错误处理 client.on('error', (err) > { addMessage(错误: ${err.message}, true); }); // 断开连接 client.on('close', () > { addMessage('连接已关闭'); connectBtn.disabled false; disconnectBtn.disabled true; subscribeBtn.disabled true; publishBtn.disabled true; }); } catch (err) { addMessage(连接失败: ${err.message}, true); } }); // 断开连接 disconnectBtn.addEventListener('click', () > { if (client) { addMessage('正在断开连接...'); client.end(); } }); // 订阅主题 subscribeBtn.addEventListener('click', () > { const topic document.getElementById('subscribeTopic').value; if (topic && client) { client.subscribe(topic, (err) > { if (err) { addMessage(订阅失败: ${err.message}, true); } else { addMessage(已订阅主题: ${topic}); } }); } }); // 发布消息 publishBtn.addEventListener('click', () > { const topic document.getElementById('publishTopic').value; const message document.getElementById('publishMessage').value; if (topic && message && client) { client.publish(topic, message); addMessage(已发布消息到 [${topic}]: ${message}); document.getElementById('publishMessage').value ''; } });

响应示例 { "statusCode":800 } 状态码 请参考 状态码 错误码 请参考 错误码

" } } 状态码 请参考 状态码 错误码 请参考 错误码

参数 参数类型 说明 示例 下级对象 statusCode Integer 调用结果状态 error String 错误码 message String 结果描述 returnObj Object 返回数据 returnObj 表 returnObj

参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池id

接口功能介绍 查设备轨迹 接口约束 无 URI GET /v3/api/getClientTrack 路径参数 无 Query参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例Id clientId 是 String 设备ID startTime 是 String 起始时间，时间格式'yyyyMMddHHmmss' endTime 是 String 截止时间，时间格式'yyyyMMddHHmmss' 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池id 请求体body参数 无 响应参数 参数 参数类型 说明 示例 下级对象 returnObj Array of Objects 返回对象 returnObj message String 描述状态 statusCode String 800成功 其他失败 error String 错误码 表 returnObj 参数 参数类型 说明 示例 下级对象 clientId String 设备ID action String 动作connect、disconnect time String 时间 yyyyMMdd HH:mm:ss 枚举参数 参数名 action 枚举值 类型 说明 connected String 成功连接事件 disconnected String 连接断开事件 authenticate String 客户端认证事件 subscribe String 客户端订阅事件 unsubscribe String 客户端取消订阅事件

参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID clientId 是 String 设备ID

接口功能介绍 向消息队列MQTT发送单条消息 接口约束 无 URI POST /v3/api/sendMessage2Mqtt 路径参数 无 Query参数 无 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池id 请求体body参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID mqttTopic 是 String MQTT主题 payload 是 String 消息内容 响应参数 参数 参数类型 说明 示例 下级对象 message String 描述状态 statusCode String 800成功 其他失败 error String 错误码 returnObj Object 返回对象 枚举参数 无 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body { "prodInstId": "123", "mqttTopic": "tt/a/b", "payload": "test" } 响应示例 { "statusCode": 800 } 状态码 请参考 状态码 错误码 请参考 错误码

参数 参数类型 说明 示例 下级对象 returnObj Object 返回数据 message String 描述状态 statusCode String 800成功 error String 错误码