集群类型 计费单位 计费方式 按需标准价格（元/vCPU/小时） 按需标准价格（元/vCPU/月） 按需标准价格（元/vCPU/年） 天翼云集群 vCPU 按量计费 0.0556 40.032 487.056 三方云集群 vCPU 按量计费 0.0556 40.032 487.056 本地集群 vCPU 按量计费 0.1668 120.096 1461.168 AnyWhere集群 vCPU 按量计费 0.1668 120.096 1461.168 天翼云集群 GPU 按量计费 0.2223 160.056 1947.348 三方云集群 GPU 按量计费 0.2223 160.056 1947.348 本地集群 GPU 按量计费 0.3335 240.12 2921.46 AnyWhere集群 GPU 按量计费 0.3335 240.12 2921.46 集群联邦 vCPU/GPU 按量计费 0 0 0 AnyWhere集群通道 vCPU/GPU 按量计费 0 0 0

本文介绍使用Rediscli迁移自建Redis(AOF文件) 迁移介绍 Rediscli 是 Redis 自带的命令行客户端工具，它允许用户通过命令行与 Redis 服务器进行交互。 在本章节中，我们将重点介绍如何使用 Rediscli 工具以 AOF 文件的方式，将自建的 Redis 数据迁移到 DCS 缓存实例。 说明 进行迁移操作前，建议暂停相关业务，以避免数据丢失或不完整。 建议业务空闲时间进行迁移操作。 步骤1：生成AOF文件 使用以下命令来开启缓存持久化并生成 AOF 持久化文件： ./rediscli h {redisaddress} p {redisport} a {password} config set appendonly yes 如果 AOF 文件的大小不再变化，说明AOF文件为全量缓存数据。 说明 使用 Rediscli 工具登录 Redis 实例，输入命令“config get dir”可以查找生成的AOF文件保存路径。 如果没有进行特殊指定，该文件的文件名默认为 appendonly.aof。 如果需要关闭同步，可以使用 Rediscli 工具登录 Redis 实例，并输入命令 “config set appendonly no” 来关闭同步。 步骤2：上传AOF文件至天翼云ECS 为节省传输时间，请先压缩AOF文件再传输。 将压缩文件（如以SFTP/SCP等方式）上传到天翼云ECS。 说明 ECS需保证有足够的磁盘空间，供数据文件存储，同时需要与缓存实例网络互通，通常要求相同VPC和相同子网，且安全组规则不限制访问端口。 步骤3：导入数据 ./rediscli h {redisaddress} p {redisport} a {password} pipe < appendonly.aof 步骤4：迁移后验证 数据导入成功后，连接DCS缓存实例，通过dbsize命令，确认数据是否导入成功 如果导入不成功，需要分析原因，修正导入语句，然后使用flushall或者flushdb命令清理实例中的缓存数据，并重新导入。

Topic与Tag释义 1）Topic：消息主题，通过Topic对不同的业务消息进行分类。 2）Tag：消息标签，用来进一步区分某个Topic下的消息分类，消息队列RocketMQ允许消费者按照Tag对消息进行过滤，确保消费者最终只消费到他关注的消息类型。 Topic与Tag都是业务上用来归类的标识，区分在于Topic是一级分类，而Tag可以说是二级分类。 适用场景 什么时候该用Topic，什么时候该用Tag，可以参考下面的一些考虑进行权衡： 考虑消息类型：如普通消息、顺序消息，事务消息、定时（延时）消息，不同消息类型是无法通过Tag区分的，这种情况就需要我们创建不同的Topic。 业务关联性：如果是不同业务之间没有直关联的消息，建议按照Topic进行区分；而同一个业务只是子类型不一样的消息可以用Tag进行区分。 消息优先级：不同的业务场景可能会导致消费端对于消息的优先级需求不同，有的紧急，有的相对来说对于延时的接收程度更大，不同优先级的消息用不同的Topic进行区分。 消息量级：如果量小但延时要求高的消息，跟超大量级（如万亿）的消息使用同一个Topic，则有可能排队时间过长导致延时无法接受，所以不同量级的消息不要使用不同的Tag，需要用不同的Topic。 总结起来就是，在消息分类实践中，有创建多个Topic，以及在同一个Topic下创建多个Tag两种常见做法。一般来说，不同的Topic之间的消息不产生直接业务上的关联，而同一个Topic下相互之间产生联系的消息可以选择用Tag来区分，一般是相同业务下的不同板块不同类型。

RocketMQ无法避免消息重复，原因主要有以下几点： 签收的偏移量是定时（每5秒/次）同步到服务端的。 为保证消息不丢失，SDK每次提交的总是队列未签收的最小偏移量（比如无序消费，offset为1、2、3、4、5 的消息，1、3、4、5消费并已签收，2未签收，签收时最后提交的偏移量将会是2，如果此时客户端重启，会从2这个位置开始消费）。 有网络交互就不能确保每一次的交互数据都是送达的，为保证数据不丢失就要进行重试，有重试就存在重复的可能。 如果业务对消费重复非常敏感，务必要注意，建议可以采用以下两种方式处理： 业务层面可以根据msgId做去重处理，如果key字段为业务唯一字段，也可采用key去重。 业务逻辑实现消费幂等，即多次处理同一消息，对业务的影响是幂等的。

API

本节介绍了在RabbitMQ实例中如何创建、修改和删除虚拟主机限制策略。 背景信息 RabbitMQ虚拟主机限制管理是一种用于管理RabbitMQ虚拟主机的功能，它允许管理员对虚拟主机的资源和权限进行限制和管理。 虚拟主机是RabbitMQ中的逻辑隔离单位，它允许在同一台RabbitMQ服务器上创建多个独立的消息队列环境。每个虚拟主机都有自己的队列、交换机、绑定和权限设置。通过虚拟主机限制管理，管理员可以对RabbitMQ服务器上的虚拟主机进行细粒度的控制和管理。这有助于确保不同的应用程序或用户之间的隔离性、安全性和资源利用率，提高整个消息队列系统的可靠性和性能。 操作步骤 1.登录管理控制台。 2.进入RabbitMQ管理控制台。 3.在实例列表页在操作列，目标实例行点击“管理”。 4.点击“集群管理”后点击“虚拟主机限制”到达虚拟主机页面，点击“新建”按钮。 5.点击“新建”后出现以下界面，选择虚拟主机，添加限制策略内容。 限制项参数 说明 maxconnections 最大TCP连接。 maxqueues 最大队列数。 6.在目标虚拟主机限制行点击“删除”或“修改”，即可删除当前虚拟主机策略。

如何通过仪表盘发现问题 以下指标是最基础也是最重要的指标，用于快速判断RabbitMQ实例健康状况。 指标1：可消费消息数 （Ready Messages） 正常情况：这个数值应该保持在一个相对稳定的水平，或者有轻微波动。 异常情况 ：如果这个数值持续快速增长 ，通常意味着生产速度远大于消费速度 ，消费者处理能力不足、数量不够，或者出现了故障。这是消息堆积的直接体现。 指标2：连接数 （Connections） 正常情况：连接数应与系统设计的客户端数量匹配。 异常情况1 ：连接数过多 ：可能导致服务器资源（内存、文件描述符）耗尽，影响性能甚至崩溃。这通常是由于客户端没有正确关闭连接（连接泄漏）。 异常情况2：连接数骤降：可能表示大量客户端离线或网络中断。 指标3：信道数 （Channels） 正常情况：信道数通常远大于连接数。 异常情况1 ：信道数过多：即使连接数不多，过多的信道也会增加服务器的内存消耗和管理负担。 异常情况2：信道数与连接数比例异常：如果每个连接只使用一个信道，可能没有充分利用 TCP 连接的多路复用优势。 指标4：消费者数（Consumers） 正常情况：消费者数量应足以处理预期的消息流量。 异常情况1 ：消费者数为零：对于需要即时处理的队列来说是严重问题，会导致消息堆积。 异常情况2：消费者数不足：会导致消息处理缓慢，Ready 消息数增长。 异常情况3：消费者数过多：可能导致资源竞争，反而降低整体效率，需要根据实际情况进行压测和调优。

本节介绍了RabbitMQ 接入方式。 安全接入点 RabbitMQ 安全接入点支持 "PLAIN"、"AMQPLAIN" 授权机制。 1、访问控制 RabbitMQ "PLAIN"、"AMQPLAIN"授权机制需要创建用户，从而获得对应虚拟主机的访问权限。 2、接入步骤 （1）新建用户（集群管理>用户>新建用户） （2）运行demo 客户端关键参数设置 "PLAIN"、"AMQPLAIN" 授权机制的客户端关键参数配置 String host "192.168.0.0"; //安全接入点ip Integer port 5672; //安全接入点port String username "xxx"; //集群管理用户列表的用户名 String password "xxx"; String vhost "/"; ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost(vhost); SSL接入点 RabbitMQ 安全接入点支持 "EXTERNAL" 授权机制 1、访问控制 无 2、接入步骤 （1）下载SSL证书（实例概览>导出服务>下载SSL文件） （2）运行demo 客户端关键参数设置 "EXTERNAL" 授权机制的客户端关键参数配置 java String host "192.168.0.0"; //SSL接入点ip int port 5671; //SSL接入点port //以下2个ssl文件可通过控制台获取安装包, 具体的获取方式可以查看2.2.1接入步骤的第二小节 String ksFile "D:tmpsslclientrabbitmqkey.p12"; String tksFile "D:tmpssltruststore"; String vhost "/"; char[] keyPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore ks KeyStore.getInstance("PKCS12"); ks.load(new FileInputStream(ksFile), keyPassphrase); KeyManagerFactory kmf KeyManagerFactory.getInstance("SunX509"); kmf.init(ks, keyPassphrase); char[] trustPassphrase null; trustPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore tks KeyStore.getInstance("JKS"); tks.load(new FileInputStream(tksFile), trustPassphrase); TrustManagerFactory tmf TrustManagerFactory.getInstance("SunX509"); tmf.init(tks); SSLContext c SSLContext.getInstance("tlsv1.2"); c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setVirtualHost(vhost); connectionFactory.setSaslConfig(DefaultSaslConfig.EXTERNAL); connectionFactory.useSslProtocol(c);

本节介绍了RabbitMQ接入的代码示例。 安全接入点(PLAIN、AMQPLAIN授权机制) java import com.rabbitmq.client.; import java.io.IOException; public class RabbitmqAmqpDemo { public static void main(String[] args) throws Exception { String host "192.168.0.0"; //安全接入点ip Integer port 5672; //安全接入点port String username "xxx"; //集群管理用户列表的用户名 String password "xxx"; //集群管理用户列表的密码 String vhost "/"; String exchangeName "extest"; String queueName "qutest"; ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setVirtualHost(vhost); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, "test"); String message "Hello Aop"; for (int i 0; i < 10; i++) { channel.basicPublish(exchangeName, "test", null, message.getBytes()); System.out.println("消息发送成功"); } Channel consumeChannel connection.createChannel(); Consumer consumer new DefaultConsumer(consumeChannel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String messageGet new String(body, "UTF8"); if (messageGet.equals(message)) { System.out.println("消息消费成功"); } } }; consumeChannel.setDefaultConsumer(consumer); consumeChannel.basicConsume(queueName, false, consumer); Thread.sleep(10000); } } SSL接入点(EXTERNAL授权机制) java import com.rabbitmq.client.; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import java.io.FileInputStream; import java.io.IOException; import java.security.KeyStore; public class RabbitmqExternalDemo { public static void main(String[] args) throws Exception { String host "192.168.0.0"; //SSL接入点ip int port 5671; //SSL接入点port //以下2个ssl文件可通过控制台获取安装包, 具体的获取方式可以查看2.2.1接入步骤的第二小节 String ksFile "D:tmpsslclientrabbitmqkey.p12"; String tksFile "D:tmpssltruststore"; String vhost "/"; String exchangeName "extest"; String queueName "qutest"; char[] keyPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore ks KeyStore.getInstance("PKCS12"); ks.load(new FileInputStream(ksFile), keyPassphrase); KeyManagerFactory kmf KeyManagerFactory.getInstance("SunX509"); kmf.init(ks, keyPassphrase); char[] trustPassphrase null; trustPassphrase "W3zT98Zz9Io".toCharArray(); KeyStore tks KeyStore.getInstance("JKS"); tks.load(new FileInputStream(tksFile), trustPassphrase); TrustManagerFactory tmf TrustManagerFactory.getInstance("SunX509"); tmf.init(tks); SSLContext c SSLContext.getInstance("tlsv1.2"); c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(host); connectionFactory.setPort(port); connectionFactory.setVirtualHost(vhost); connectionFactory.setSaslConfig(DefaultSaslConfig.EXTERNAL); connectionFactory.useSslProtocol(c); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, "test"); String message "Hello Aop"; for (int i 0; i < 10; i++) { channel.basicPublish(exchangeName, "test", null, message.getBytes()); System.out.println("消息发送成功"); } Channel consumeChannel connection.createChannel(); Consumer consumer new DefaultConsumer(consumeChannel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String messageGet new String(body, "UTF8"); if (messageGet.equals(message)) { System.out.println("消息消费成功"); } } }; consumeChannel.setDefaultConsumer(consumer); consumeChannel.basicConsume(queueName, false, consumer); Thread.sleep(10000); } }

本节主要介绍如何下载备份数据 Redis 管理控制台支持备份数据下载。 备份完成后，您可根据需求获取公网或内网下载地址进行下载。 前提条件 只有当缓存实例处于“运行中”状态，才能执行此操作。 操作步骤 1. 登录 Redis管理控制台。 2. 在管理控制台右上角选择实例所在的区域。 3. 在实例列表页，单击目标实例名称进入实例详情页面。 4. 左侧菜单点击备份与恢复，进入备份管理操作页面。 5. 点击【下载】按钮，弹框展示公网和内网下载地址，根据需求获取下载地址。 6. 访问下载地址，获取备份文件。 说明 仅当备份成功后，方可下载备份数据。

本节主要介绍如何下载备份数据 Redis 管理控制台支持备份数据下载。 备份完成后，您可根据需求获取公网或内网下载地址进行下载。 前提条件 只有当缓存实例处于“运行中”状态，才能执行此操作。 操作步骤 1. 登录 Redis管理控制台。 2. 在管理控制台右上角选择实例所在的区域。 3. 在实例列表页，单击目标实例名称进入实例详情页面。 4. 左侧菜单点击备份与恢复，进入备份管理操作页面。 5. 点击【下载】按钮，弹框展示公网和内网下载地址，根据需求获取下载地址。 6. 访问下载地址，获取备份文件。 说明 仅当备份成功后，方可下载备份数据。

本节介绍联邦资源调度策略。 调度策略概述 调度策略分为自定义调度策略和集群调度策略，自定义调度策略是命名空间级别的，集群调度策略是集群级别的， 自定义调度策略的优先级高于集群调度策略。 调度策略说明 自定义调度策略和集群调度策略配置数据一致。 基础配置 配置说明 优先级： 策略优先级，值越高，优先级越高。 冲突解决策略：资源冲突处理策略，默认“Abort”表示停止传播避免意外覆盖；可选“Overwrite”表示通过主动覆盖来接管资源。 依赖自动传播: 关联的资源是否自动传递，比如工作负载Deployment依赖配置资源ConfigMap，如果开启该配置，配置资源ConfigMap会自动传播，否则不自动传播。 资源选择 选择需要调度的资源，支持多选。支持指定资源名称，也支持通过标签选择资源。 调度配置 调度方式支持集群权重、自动均衡、复制分发三种方式。 集群权重 以静态权重比例方式分配成员集群Pod群。存在余数时，剩余Pod数将按照权重继续分配，如权重相同，则随机选择集群进行分配。

本节介绍联邦资源差异化策略。 差异化策略概述 差异化策略分为自定义差异化策略和集群差异化策略，自定义差异化策略是命名空间级别的，集群差异化策略是集群级别的， 多种差异化策略会同时生效。 差异化策略说明 基础配置&资源选择 资源选择： 选择需要调度的资源，支持多选。支持指定资源名称，也支持通过标签选择资源。 差异化策略 目标集群选择 可选择规则对所有成员集群生效。 可根据过滤条件指定生产的集群，过滤条件支持：指定集群、排除集群、标签过滤、字段过滤四种方式。 差异化配置 差异化配置说明 差异项 差异化配置说明 镜像差异化(imageOverrider) 指定容器镜像参数差异化配置 命令差异化(commandOverrider) 指定容器运行命令的差异化配置 参数差异化(argsOverrider) 指定容器运行参数的差异化配置，与commandOverrider类似 标签差异化(labelsOverrider) 指标资源标签的差异化配置 注解差异化(annotationsOverrider) 指定资源注解的差异化配置 字段差异化(fieldOverrider) 支持内置yaml、json的差异化 文本差异化(plaintext) 泛指任意资源，与kubectl patch方式类似

本节介绍如何统一管理任意环境下的Kubernetes集群。 概述 CCE One 注册集群可以帮助您快速实现Kubernetes集群注册到云端，使用天翼云容器管理控制台对注册集群进行统一管理，赋予云下集群使用云上资源能力。 场景描述 统一管理天翼云集群：天翼云CCE专有版、托管版、智算版以及Serverless容器引擎等。 统一管理本地集群：由CCE提供的运行在您数据中心基础设施之上的Kubernetes集群，以及满足CNCF标准的IDC自建的Kubernetes集群。 统一管理三方集群：三方公有云上提供的Kubernetes集群产品，如阿里云(ACK)、腾讯云(TKE)、GCP (GKE)、AWS(EKS)等。 产品优势 云上云上集群统一管理，通过天翼云容器管理控制台，赋予云上云下集群可观测、权限管理等能力。 云下集群赋予云上能力，如通过接入云上ECS、虚拟节点等扩展云下集群算力。 产品架构 操作步骤 1. 创建注册集群并接入Kubernetes集群，详细步骤见 本地注册集群、三方云注册集群。 2. 集群接入后，使用天翼云容器管理控制台对接入集群进行管理，如 Kubernetes RBAC授权、运维等。

本节介绍计费模式。 按量计费 适用场景 按需付费是一种灵活的计费模式，适用于需要灵活调整资源、业务不稳定或资金有限的场景。在选择计费模式时，应结合业务需求和实际情况来做出合适的选择。 收费方式 一种后付费模式，即先使用再付费。 计费说明 开通按需计费资源时，您的天翼云账户余额+通用代金券须大于100元，并且大于按需计费资源（当前已开通+即将要开通）1小时费用的总和。 按需计费根据资源的结算周期进行结算。一般按需计费的结算周期有小时/日/月等，在达到结算周期时，系统会生成账单，进行扣费。 如果账户余额不足，资源将进入保留期，需要在保留期完成缴费，超过保留期，所使用资源将被关停并收回资源。

本节介绍资源委托协议。 使用 CCE One 服务需要授予访问以下云资源的权限： 访问计算类服务 注册集群按需弹性云上资源时会关联创建云服务器，需要获取访问弹性云服务器、弹性裸金属服务器的权限。 访问存储类服务 为注册集群关联云上节点和容器挂载存储，需要获取访问云硬盘、弹性文件、对象存储等服务的权限。 访问网络类服务 为注册集群及联邦提供网络代理及服务对外暴露，需要获取访问虚拟私有云、弹性负载均衡、弹性IP、NAT网关等服务的权限。 访问容器与监控类服务 为三方注册集群下容器提供镜像拉取、监控和日志分析等功能，需要获取访问容器镜像、应用运维管理等服务的权限同意授权后，CCE One 将在统一身份认证服务为您创建名为 CtyunAssumeRoleForCCEONE 的 委托，为保证服务正常使用，在使用 CCE One 服务期间请不要删除或修改 CtyunAssumeRoleForCCEONE 委托。

Command Family Command redis 5.0 Bitmap BITCOUNT ✔️ Bitmap BITFIELD ✔️ Bitmap BITFIELDRO ✔️ Bitmap BITOP ✔️ Bitmap BITPOS ✔️ Bitmap GETBIT ✔️ Bitmap SETBIT ✔️ Cluster management ASKING ✔️ Cluster management CLUSTER ADDSLOTS ✔️ Cluster management CLUSTER ADDSLOTSRANGE ➖ Cluster management CLUSTER BUMPEPOCH ✔️ Cluster management CLUSTER COUNTFAILUREREPORTS ✔️ Cluster management CLUSTER COUNTKEYSINSLOT ✔️ Cluster management CLUSTER DELSLOTS ✔️ Cluster management CLUSTER DELSLOTSRANGE ➖ Cluster management CLUSTER FAILOVER ✔️ Cluster management CLUSTER FLUSHSLOTS ✔️ Cluster management CLUSTER FORGET ✔️ Cluster management CLUSTER GETKEYSINSLOT ✔️ Cluster management CLUSTER INFO ✔️ Cluster management CLUSTER KEYSLOT ✔️ Cluster management CLUSTER LINKS ➖ Cluster management CLUSTER MEET ✔️ Cluster management CLUSTER MYID ✔️ Cluster management CLUSTER MYSHARDID ➖ Cluster management CLUSTER NODES ✔️ Cluster management CLUSTER REPLICAS ✔️ Cluster management CLUSTER REPLICATE ✔️ Cluster management CLUSTER RESET ✔️ Cluster management CLUSTER SAVECONFIG ✔️ Cluster management CLUSTER SETCONFIGEPOCH ✔️ Cluster management CLUSTER SETSLOT ✔️ Cluster management CLUSTER SHARDS ➖ Cluster management CLUSTER SLAVES ✔️ Cluster management CLUSTER SLOTS ✔️ Cluster management READONLY ✔️ Cluster management READWRITE ✔️ Connection AUTH ✔️ Connection CLIENT CACHING ➖ Connection CLIENT GETNAME ✔️ Connection CLIENT GETREDIR ➖ Connection CLIENT ID ✔️ Connection CLIENT INFO ➖ Connection CLIENT KILL ✔️ Connection CLIENT LIST ✔️ Connection CLIENT NOEVICT ➖ Connection CLIENT NOTOUCH ➖ Connection CLIENT PAUSE ✔️ Connection CLIENT REPLY ✔️ Connection CLIENT SETINFO ➖ Connection CLIENT SETNAME ✔️ Connection CLIENT TRACKING ➖ Connection CLIENT TRACKINGINFO ➖ Connection CLIENT UNBLOCK ✔️ Connection CLIENT UNPAUSE ➖ Connection ECHO ✔️ Connection HELLO ➖ Connection PING ✔️ Connection QUIT ✔️ Connection RESET ➖ Connection SELECT ✔️ Generic COPY ➖ Generic DEL ✔️ Generic DUMP ✔️ Generic EXISTS ✔️ Generic EXPIRE ✔️ Generic EXPIREAT ✔️ Generic EXPIRETIME ➖ Generic KEYS ✔️ Generic MIGRATE ✔️ Generic MOVE ✔️ Generic OBJECT ENCODING ✔️ Generic OBJECT FREQ ✔️ Generic OBJECT IDLETIME ✔️ Generic OBJECT REFCOUNT ✔️ Generic PRESIST ✔️ Generic PEXPIRE ✔️ Generic PEXPIREAT ✔️ Generic PEXPIRETIME ➖ Generic PTTL ✔️ Generic RANDOMKEY ✔️ Generic RENAME ✔️ Generic RENAMENX ✔️ Generic RESTORE ✔️ Generic SCAN ✔️ Generic SORT ✔️ Generic SORTRO ➖ Generic TOUCH ✔️ Generic TTL ✔️ Generic TYPE ✔️ Generic UNLINK ✔️ Generic WAIT ✔️ Generic WAITAOF ➖ Hash HDEL ✔️ Hash HEXISTS ✔️ Hash HGET ✔️ Hash HGETALL ✔️ Hash HINCRBY ✔️ Hash HINCRBYFLOAT ✔️ Hash HKEYS ✔️ Hash HLEN ✔️ Hash HMGET ✔️ Hash HMSET ✔️ Hash HRANDFIELD ➖ Hash HSCAN ✔️ Hash HSET ✔️ Hash HSETNX ✔️ Hash HSTRLEN ✔️ Hash HVALS ✔️ HyperLogLog PFADD ✔️ HyperLogLog PFMERGE ✔️ HyperLogLog PFCOUNT ✔️ HyperLogLog PFDEBUG ✔️ HyperLogLog PFSELFTEST ✔️ List BLMOVE ➖ List BLMPOP ➖ List BLPOP ✔️ List BRPOP ✔️ List BRPOPLPUSH ✔️ List LINDEX ✔️ List LINSERT ✔️ List LLEN ✔️ List LMOVE ➖ List LMPOP ➖ List LPOP ✔️ List LPOS ➖ List LPUSH ✔️ List LPUSHX ✔️ List LRANGE ✔️ List LREM ✔️ List LSET ✔️ List LTRIM ✔️ List RPOP ✔️ List RPOPLPUSH ✔️ List RPUSH ✔️ List RPUSHX ✔️ PubSub PSUBSCRIBE ✔️ PubSub PUBLISH ✔️ PubSub PUBSUB CHANNELS ✔️ PubSub PUBSUB NUMPAT ✔️ PubSub PUBSUB NUMSUB ✔️ PubSub PUBSUB SHARDCHANNELS ➖ PubSub PUBSUB SHARDNUMSUB ➖ PubSub PUNSUBSCRIBE ✔️ PubSub SPUBLISH ➖ PubSub SSUBSCRIBE ➖ PubSub SUBSCRIBE ✔️ PubSub SUNSUBSCRIBE ➖ PubSub UNSUBSCRIBE ✔️ Scripting EVAL ✔️ Scripting EVALRO ➖ Scripting EVALSHA ✔️ Scripting EVALSHARO ➖ Scripting FCALL ➖ Scripting FCALLRO ➖ Scripting FUNCTION DELETE ➖ Scripting FUNCTION DUMP ➖ Scripting FUNCTION FLUSH ➖ Scripting FUNCTION KILL ➖ Scripting FUNCTION LIST ➖ Scripting FUNCTION LOAD ➖ Scripting FUNCTION RESTORE ➖ Scripting FUNCTION STATS ➖ Scripting SCRIPT DEBUG ✔️ Scripting SCRIPT EXISTS ✔️ Scripting SCRIPT FLUSH ✔️ Scripting SCRIPT KILL ✔️ Scripting SCRIPT LOAD ✔️ Server ACL CAT ➖ Server ACL DELUSER ➖ Server ACL DRYRUN ➖ Server ACL GENPASS ➖ Server ACL GETUSER ➖ Server ACL LIST ➖ Server ACL LOAD ➖ Server ACL LOG ➖ Server ACL SAVE ➖ Server ACL SETUSER ➖ Server ACL USERS ➖ Server ACL WHOAMI ➖ Server BGREWRITEAOF ✔️ Server BGSAVE ✔️ Server COMMAND ✔️ Server COMMAND COUNT ✔️ Server COMMAND DOCS ➖ Server COMMAND GETKEYS ✔️ Server COMMAND GETKEYSANDFLAGS ➖ Server COMMAND INFO ✔️ Server COMMAND LIST ➖ Server CONFIG GET ✔️ Server CONFIG RESETSTAT ✔️ Server CONFIG REWRITE ✔️ Server CONFIG SET ✔️ Server DBSIZE ✔️ Server FAILOVER ➖ Server FLUSHALL ✔️ Server FLUSHDB ✔️ Server INFO ✔️ Server LASTSAVE ✔️ Server LATENCY DOCTOR ✔️ Server LATENCY GRAPH ✔️ Server LATENCY HISTOGRAM ➖ Server LATENCY HISTORY ✔️ Server LATENCY LATEST ✔️ Server LATENCY RESET ✔️ Server LOLWUT ✔️ Server MEMORY DOCTOR ✔️ Server MEMORY MALLOCSTATS ✔️ Server MEMORY PURGE ✔️ Server MEMORY STATS ✔️ Server MEMORY USAGE ✔️ Server MODULE LIST ✔️ Server MODULE LOAD ✔️ Server MODULE LOADEX ➖ Server MODULE UNLOAD ✔️ Server MONITOR ✔️ Server PSYNC ✔️ Server REPLCONF ✔️ Server REPLICAOF ✔️ Server RESTOREASKING ✔️ Server ROLE ✔️ Server SAVE ✔️ Server SHUTDOWN ✔️ Server SLAVEOF ✔️ Server SLOWLOG GET ✔️ Server SLOWLOG LEN ✔️ Server SLOWLOG RESET ✔️ Server SWAPDB ✔️ Server SYNC ✔️ Server TIME ✔️ Set SADD ✔️ Set SCARD ✔️ Set SDIFF ✔️ Set SDIFFSTORE ✔️ Set SINTER ✔️ Set SINTERCARD ➖ Set SINTERSTORE ✔️ Set SISMEMBER ✔️ Set SMEMBERS ✔️ Set SMISMEMBER ➖ Set SMOVE ✔️ Set SPOP ✔️ Set SRANDMEMBER ✔️ Set SREM ✔️ Set SSCAN ✔️ Set SUNION ✔️ Set SUNIONSTORE ✔️ Sorted Set BZMPOP ➖ Sorted Set BZPOPMAX ✔️ Sorted Set BZPOPMIN ✔️ Sorted Set ZADD ✔️ Sorted Set ZCARD ✔️ Sorted Set ZCOUNT ✔️ Sorted Set ZDIFF ➖ Sorted Set ZDIFFSTORE ➖ Sorted Set ZINCRBY ✔️ Sorted Set ZINTER ➖ Sorted Set ZINTERCARD ➖ Sorted Set ZINTERSTORE ✔️ Sorted Set ZLEXCOUNT ✔️ Sorted Set ZMPOP ➖ Sorted Set ZMSCORE ➖ Sorted Set ZPOPMAX ✔️ Sorted Set ZPOPMIN ✔️ Sorted Set ZRANDMEMBER ➖ Sorted Set ZRANGE ✔️ Sorted Set ZRANGEBYLEX ✔️ Sorted Set ZRANGEBYSCORE ✔️ Sorted Set ZRANGESTORE ➖ Sorted Set ZRANK ✔️ Sorted Set ZREM ✔️ Sorted Set ZREMRANGEBYLEX ✔️ Sorted Set ZREMRANGEBYRANK ✔️ Sorted Set ZREMRANGEBYSCORE ✔️ Sorted Set ZREVRANGE ✔️ Sorted Set ZREVRANGEBYLEX ✔️ Sorted Set ZREVRANGEBYSCORE ✔️ Sorted Set ZREVRANK ✔️ Sorted Set ZSCAN ✔️ Sorted Set ZSCORE ✔️ Sorted Set ZUNION ➖ Sorted Set ZUNIONSTORE ✔️ Stream XACK ✔️ Stream XADD ✔️ Stream XAUTOCLAIM ➖ Stream XCLAIM ✔️ Stream XDEL ✔️ Stream XGROUP ✔️ Stream XINFO ✔️ Stream XLEN ✔️ Stream XPENDING ✔️ Stream XRANGE ✔️ Stream XREAD ✔️ Stream XREADGROUP ✔️ Stream XREVRANGE ✔️ Stream XSETID ✔️ Stream XTRIM ✔️ String APPEND ✔️ String DECR ✔️ String DECRBY ✔️ String GET ✔️ String GETDEL ➖ String GETEX ➖ String GETRANGE ✔️ String GETSET ✔️ String INCR ✔️ String INCRBY ✔️ String INCRBYFLOAT ✔️ String LCS ➖ String MGET ✔️ String MSET ✔️ String MSETNX ✔️ String PSETEX ✔️ String SET ✔️ String SETEX ✔️ String SETNX ✔️ String SETRANGE ✔️ String STRLEN ✔️ String SUBSTR ✔️ Transactions DISCARD ✔️ Transactions EXEC ✔️ Transactions MULTI ✔️ Transactions UNWATCH ✔️ Transactions WATCH ✔️

RocketMQ实例是否支持扩容？ 可以扩容。登录RocketMQ控制台，在包周期实例所在行，单击“更多 > 扩容”，即答完成实例的扩容。 消费的最长保留时间是多久？ 一般情况下消息如果未被消费会一直保留，只有被消费后，才会被删除。但是如果设置了过期时间（TTL），则以TTL时间为准。

在业务场景允许的情况下，优先选择无序消息，或者在业务能变通的情况下，将有序消息转化为无序消息。 无序消息的优点： 生产者可以使用多进程、多线程往同一个Topic发送消息，性能更好。 消费者可以使用多进程、多线程同时消费，性能较好。 可以充分使用集群的Failover特点，无须依赖自动主备切换（切换过程服务会中断），包括： 当集群中某一Broker节点故障时，不影响业务消息生产，消息将failover发送到其它节点； 当集群中某一Broker节点故障时，不影响其它节点数据消费，故障恢复后即可消费。 能动态地扩容。 有序消息的缺点： 对于有序消息，当节点故障时，Queue数不会变化，生产与消费都会出现异常，直到故障节点恢复。 对于有序消息，需要将所有消息消费完，并且停止客户端，才能扩容。

场景描述 在RocketMQ中，删除Topic是一个比较重要且敏感的操作，需要谨慎处理。一般来说，删除Topic的场景有以下几种情况： 业务不再需要该Topic：当某个Topic对应的业务已经结束或不再需要时，可以考虑删除该Topic，以释放资源和减少管理工作。 Topic配置错误或不合适：如果创建Topic时配置错误或者配置不合适，可以考虑删除该Topic，并重新创建一个正确的Topic。 数据归档或清理：在某些情况下，可能需要对Topic中的数据进行归档或清理，以释放存储空间。在归档或清理之前，需要先将Topic中的消息备份或迁移到其他地方，确保数据的完整性和可恢复性。 无论是哪种场景，删除Topic都需要注意以下几点： 确保Topic中的消息已经被正确处理和消费：在删除Topic之前，需要确保Topic中的消息已经被正确地处理和消费，以避免数据丢失或处理中断。 停止生产者和消费者对该Topic的操作：在删除Topic之前，需要停止生产者和消费者对该Topic的操作，以避免数据冲突或丢失。 确保删除操作的权限和安全性：删除Topic通常需要管理员或具有相应权限的用户来执行，确保只有授权人员可以进行删除操作。 慎重操作，备份重要数据：在删除Topic之前，建议备份Topic中的重要数据，以防止误操作或数据丢失。 总之，删除Topic是一个敏感操作，需要在慎重考虑和评估后进行，以确保不会对业务和数据产生不可逆的影响。 操作步骤 1、进入Topic管理菜单。 2、在Topic管理菜单选择将要删除的主题，在更多下拉框选择删除，即可完成删除 注意事项： ● 删除主题前必须确保该主题对应的生产消费实例已经全部停止。 ● 删除主题后消息数据会发生丢失。

产品服务协议请查看： 产品服务协议 产品服务等级协议请查看： 产品服务等级协议

本节介绍计费模式。 按量计费 适用场景 按需付费是一种灵活的计费模式，适用于需要灵活调整资源、业务不稳定或资金有限的场景。在选择计费模式时，应结合业务需求和实际情况来做出合适的选择。 收费方式 一种后付费模式，即先使用再付费。 计费说明 开通按需计费资源时，您的天翼云账户余额+通用代金券须大于100元，并且大于按需计费资源（当前已开通+即将要开通）1小时费用的总和。 按需计费根据资源的结算周期进行结算。一般按需计费的结算周期有小时/日/月等，在达到结算周期时，系统会生成账单，进行扣费。 如果账户余额不足，资源将进入保留期，需要在保留期完成缴费，超过保留期，所使用资源将被关停并收回资源。

本章节主要介绍连接类问题 RocketMQ是否支持自建Broker连接云上nameserver？ 不支持。自建Broker连接到云上RocketMQ的nameserver，会影响管理面功能，导致查询不到Topic等问题。

功能 Proxy集群默认只支持DB 0。 通过改写请求的KEY名，实现多DB的功能。 开启多DB后，DB数支持0~255。 打开/关闭方法 1. 登录 Redis管理控制台。 2. 在管理控制台左上角选择实例所在的区域。 3. 在实例列表页，单击目标实例名称进入实例详情管理。 4. 左侧菜单点击实例配置>参数配置，打开参数配置界面，修改multipledb参数后点击保存按钮，即可生效。 注意 注意开启/关闭多DB前，最好先清空数据。 修改后，原来的数据无法正常通过代理访问。 开启多DB后的限制 1. 不支持SWAPDB命令 2. DBSIZE和FLUSHDB命令执行的效率很低，涉及SCAN所有KEY。尽量不要使用 3. SCAN命令和KEYS命令的执行效率也会降低。 4. INFO命令KEYSPACE部分不会展示其他DB的信息。 5. RANDOMKEY命令有限支持。但并不会随机返回所有KEY 6. LUA脚本和FUNCTION函数里面不支持多DB。 脚本和函数里面也不支持使用参数以外的其他KEY访问 7. LUA脚本中不支持使用publish 8. DB支持范围0~255 9. 事务中不支持执行SELECT和MOVE命令 10. 实际存储的KEY是经过改写的。所以从RDB文件或慢语句等其他途径会看到非原始的KEY。但客户端通过代理的场景，则不会有任何影响。

本文介绍节点重启后消费者如何重连，以Java中使用的RabbitMQ客户端amqpclient为例。 amqpclient自带重连机制，但是自带的重连机制只会重试一次，一次连不上后就不会再执行了，这时如果消费者没有做额外的重试机制，那么这个消费者就彻底丧失的消费能力。 amqpclient在节点断连后，根据与通道建立的节点不同，产生不同的错误。 如果通道连接的是队列所在的节点，消费者就会收到一个shutdown信号，这时amqpclient的重连机制就会生效，尝试重新连接服务端。如果连上了，这个通道就会继续连接消费。如果连不上，就会执行channel.close方法，关闭这个通道。 如果通道连接的不是队列所在的节点，消费者不会触发关闭动作，而是由服务端发送的一个取消动作，这个动作对amqpclient来说并不是异常行为，所以日志上不会有明显的报错，但是连接最终还是会关闭。 amqpclient出现上面两种错误时，会分别回调handleShutdownSignal以及handleCancel方法，您可以通过重写这两种方法，在回调时执行重写的重连逻辑，就能在通道关闭后重新创建消费者的新通道继续消费。 以下提供一个简单的代码示例，能够解决上面的两种错误，实现消费者的持续消费。 import com.rabbitmq.client.; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; public class MyRabbitConsumer { public static void main(String... args) throws IOException, TimeoutException { ConnectionFactory factory new ConnectionFactory(); factory.setHost("192.168.x.x"); factory.setPort(5672); factory.setUsername("name"); factory.setPassword("password"); Connection connection factory.newConnection(); createNewConnection(connection); } public static void createNewConnection(Connection connection) { try { Channel channel connection.createChannel(); channel.basicQos(64); channel.basicConsume("queue1", false, new CustomConsumer(channel, connection)); } catch (Exception e) { createNewConnection(connection); } } static class CustomConsumer implements Consumer { private final Channel channel; private final Connection connection; public CustomConsumer(Channel channel, Connection connection) { channel channel; connection connection; } @Override public void handleConsumeOk(String consumerTag) {} @Override public void handleCancelOk(String consumerTag) {} @Override public void handleCancel(String consumerTag) throws IOException { createNewConnection(connection); } @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { createNewConnection(connection); } @Override public void handleRecoverOk(String consumerTag) {} @Override public void handleDelivery(String consumerTag, Envelope env, AMQP.BasicProperties prop, byte[] body) throws IOException { String message new String(body, StandardCharsets.UTF8); System.out.println("收到消息: " + message); channel.basicAck(env.getDeliveryTag(), false); } } }

本节介绍Kafka 发送给Topic的消息在分区中分布不均衡原因 问题现象 发送消息到某个Topic后，该Topic下部分分区消息比较多，部分分区消息少，甚至没有。 可能原因 发送消息时指定了分区，导致未指定的分区没有消息。 发送消息时指定了消息Key，按照对应的Key发送消息至对应的分区，导致分区消息不均衡。 通过代码重新实现了分区分配策略，但策略逻辑有问题，导致分区消息不均衡。

本节介绍Kafka不推荐使用Sarama Go客户端收发消息 问题现象 所有Sarama Go版本客户端存在以下已知问题： 当Topic新增分区时，Sarama Go客户端无法感知并消费新增分区，需要客户端重启后，才能消费到新增分区。 当Sarama Go客户端同时订阅两个以上的Topic时，有可能会导致部分分区无法正常消费消息。 当Sarama Go客户端的消费位点重置策略设置为Oldest(earliest)时，如果客户端宕机或服务端版本升级，由于Sarama Go客户端自行实现OutOfRange机制，有可能会导致客户端从最小位点开始重新消费所有消息。 解决方案 建议尽早将Sarama Go客户端替换为Confluent Go客户端。

本节介绍Kafka消费客户端频繁出现Rebalance原因及解决方案 可能原因 可能是Kafka客户端版本过低或者Consumer没有独立线程维持心跳。 v0.10.2之前版本的客户端：Consumer没有独立线程维持心跳，而是把心跳维持与poll接口耦合在一起。其结果就是，如果用户消费出现卡顿，就会导致Consumer心跳超时，引发Rebalance。 v0.10.2及之后版本的客户端：如果消费时间过慢，超过一定时间（max.poll.interval.ms设置的值，默认5分钟）未进行poll拉取消息，则会导致客户端主动离开队列，而引发Rebalance。 解决方案 首先您需要了解以下几点信息： session.timeout.ms：心跳超时时间（可以由客户端自行设置）。 max.poll.records：每次poll返回的最大消息数量。 v0.10.2之前版本的客户端：心跳是通过poll接口来实现的，没有内置的独立线程。 v0.10.2及之后版本的客户端：为了防止客户端长时间不进行消费，Kafka客户端在v0.10.2及之后的版本中引入了max.poll.interval.ms配置参数。 （1）参考以下说明调整参数值： session.timeout.ms：v0.10.2之前的版本可适当提高该参数值，需要大于消费一批数据的时间，但不要超过30s，建议设置为25s；而v0.10.2及其之后的版本，保持默认值10s即可。 max.poll.records：降低该参数值，建议远远小于 的积。 max.poll.interval.ms：该值要大于 / ( )的值。 （2）尽量提高客户端的消费速度，消费逻辑另起线程进行处理。 （3）减少Group订阅Topic的数量，一个Group订阅的Topic最好不要超过5个，建议一个Group只订阅一个Topic。 （4）将客户端升级至0.10.2以上版本。

本文记录了用户管理功能使用过程中的常见问题 创建用户时，"密码是否Hash"选项怎么选择？ 正常情况下不需要开启该选项，不开启该选项时，使用SASL连接时，直接使用设置的密码即可。如果开启了该选项，使用SASL连接时，需要取设置的密码的小写32位MD5哈希值。 原先创建的用户，没有涉及到"密码是否Hash"参数，该怎么使用？ 在2025年8月24日之前创建的用户，默认都是开启了"密码是否Hash"选项，即使用SASL连接时，需要取设置的密码的小写32位MD5哈希值。具体用户是否开启了"密码是否Hash"选项，可以通过查看控制台用户管理的用户列表页面得知。 创建用户时，为什么没有"密码是否Hash"选项？ 新租户不支持选择该选项，使用SASL连接时，直接使用设置的密码即可。如果需要使用该功能，可以提单联系我们。

本文介绍Kafka如何配置安全组 客户端只能部署在与Kafka实例处于相同虚拟私有云（VPC）和相同子网的弹性云主机（CTECS）上。 除了CTECS、Kafka实例必须处于相同VPC之外，还需要他们的安全组分别配置了正确的规则，客户端才能访问Kafka实例。 如果CTECS、Kafka实例配置相同的安全组，安全组创建后，默认包含组内网络访问不受限制的规则。 如果CTECS、Kafka实例配置了不同安全组，可参考如下配置方式： 说明 假设CTECS、Kafka实例分别配置了安全组：sgsCTECS、sgsKafka。 以Kafka访问端口9098为例，其它实例请以实际情况为准。 以下规则，远端可使用安全组，也可以使用具体的IP地址。 配置CTECS所在安全组。CTECS所在安全组需要增加如下出方向规则，以保证客户端能正常访问Kafka实例。如果出方向规则不受限，则不用添加。 配置Kafka实例所在安全组。Kafka实例所在安全组需要增加如下入方向规则，以保证能被客户端访问。

本文介绍解决Lettuce 6.x版本客户端鉴权错误问题 使用Lettuce 6.x版本客户端，连接redis 2.8/4.0/5.0集群实例时，会报错"NOAUTH Authentication required"。 原因是Lettuce 6.x版本开始，使用RESP3（Redis 6.x引入）的HELLO命令进行版本自适应判断，但是对于不支持HELLO命令的低版本实例，兼容性存在一定问题。所以对于低版本的实例，建议直接在Lettuce中指定使用RESP2协议（兼容Redis 4/5）的版本来使用。 在客户端连接代码添加指定RESP2协议访问Redis即可解决： @Configuration public class SpringConfig implements LettuceClientConfigurationBuilderCustomizer { @Override public void customize(LettuceClientConfiguration.LettuceClientConfigurationBuilder clientConfigurationBuilder) { // manually specifying RESP2 clientConfigurationBuilder.clientOptions(ClientOptions.builder() .protocolVersion(ProtocolVersion.RESP2) .build()); } }

消费消息 import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.config.SslConfigs; import java.util.ArrayList; import java.util.List; import java.util.Properties; public class KafkaConsumerDemo { public static void main(String args[]) { //加载kafka.properties Properties kafkaProperties JavaKafkaConfigurer.getKafkaProperties(); Properties props new Properties(); //设置接入点，请通过控制台获取对应Topic的接入点 props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, kafkaProperties.getProperty("bootstrap.servers")); //设置SSL根证书的路径，请记得将XXX修改为自己的路径 props.put(SslConfigs.SSLTRUSTSTORELOCATIONCONFIG, kafkaProperties.getProperty("ssl.truststore.location")); //根证书store的密码，保持不变 props.put(SslConfigs.SSLTRUSTSTOREPASSWORDCONFIG, "c24f5210"); //接入协议，目前支持使用SSL协议接入 props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SSL"); //两次poll之间的最大允许间隔 //可更加实际拉去数据和客户的版本等设置此值，默认30s props.put(ConsumerConfig.SESSIONTIMEOUTMSCONFIG, 30000); //设置单次拉取的量，走公网访问时，该参数会有较大影响 props.put(ConsumerConfig.MAXPARTITIONFETCHBYTESCONFIG, 32000); props.put(ConsumerConfig.FETCHMAXBYTESCONFIG, 32000); //每次poll的最大数量 //注意该值不要改得太大，如果poll太多数据，而不能在下次poll之前消费完，则会触发一次负载均衡，产生卡顿 props.put(ConsumerConfig.MAXPOLLRECORDSCONFIG, 30); //消息的反序列化方式 props.put(ConsumerConfig.KEYDESERIALIZERCLASSCONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUEDESERIALIZERCLASSCONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); //当前消费实例所属的消费组，请在控制台申请之后填写 //属于同一个组的消费实例，会负载消费消息 props.put(ConsumerConfig.GROUPIDCONFIG, kafkaProperties.getProperty("group.id")); //hostname校验改成空 props.put(SslConfigs.SSLENDPOINTIDENTIFICATIONALGORITHMCONFIG, ""); //构造消息对象，也即生成一个消费实例 KafkaConsumer consumer new org.apache.kafka.clients.consumer.KafkaConsumer (props); //设置消费组订阅的Topic，可以订阅多个 //如果GROUPIDCONFIG是一样，则订阅的Topic也建议设置成一样 List subscribedTopics new ArrayList (); //如果需要订阅多个Topic，则在这里add进去即可 //每个Topic需要先在控制台进行创建 subscribedTopics.add(kafkaProperties.getProperty("topic")); consumer.subscribe(subscribedTopics); //循环消费消息 while (true){ try { ConsumerRecords records consumer.poll(1000); //必须在下次poll之前消费完这些数据, 且总耗时不得超过SESSIONTIMEOUTMSCONFIG //建议开一个单独的线程池来消费消息，然后异步返回结果 for (ConsumerRecord record : records) { System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset())); } } catch (Exception e) { try { Thread.sleep(1000); } catch (Throwable ignore) { } e.printStackTrace(); } } } }