客户端缓存（Client side caching） Redis 客户端缓存在某些方面进行了重新设计。 可以执行Client相关的命令，开启客户端缓存。开启后，服务端后记住客户端连接查询的KEY。同时客户端需要订阅内部有主题， 如果一旦客户端查询的KEY发生变化，客户端会收到发生变化的信息。 由于客户端能感知到KEY的变化，因而客户端可以直接缓存KEY的值在本地内存上，而不需要担心KEY失效的问题。 多线程 IO（Threaded I/O）   Redis 6 引入多线程IO，但多线程部分只是用来处理网络数据的读写和协议解析，执行命令仍然是单线程。但带来了更加的整体性能。

消费端最常见的问题就是消费堆积，最常造成堆积的原因是： 消费速度跟不上生产速度，此时应该提高消费速度，详见下一节《提高消费速度》。 消费端产生了阻塞。 消费端拿到消息后，执行消费逻辑，通常会执行一些远程调用，如果这个时候同步等待结果，则有可能造成一直等待，消费进程无法向前推进。 消费端应该竭力避免堵塞消费线程，如果存在等待调用结果的情况，建议设置等待的超时时间，超时后作消费失败处理。

Kafka 是按分区一条一条消息顺序向前推进消费的，如果消费端拿到某条消息后执行消费逻辑失败，比如应用服务器出现了脏数据，导致某条消息处理失败，等待人工干预，那么有以下两种处理方式： 失败后一直尝试再次执行消费逻辑。这种方式有可能造成消费线程阻塞在当前消息，无法向前推进，造成消息堆积。 由于 Kafka 自身没有处理失败消息的设计，实践中通常会打印失败的消息、或者存储到某个服务（比如创建一个 Topic 专门用来放失败的消息），然后定时 check 失败消息的情况，分析失败原因，根据情况处理。

Kafka的消费机制是由客户端主动去服务端拉取消息进行消费的。因此，一般来说，如果客户端能够及时消费，则不会产生较大延迟。如果产生了较大延迟，请先关注是否有堆积，并注意提高消费速度。

项目 说明 测试实例规格 基础读写分离8G(2副本) 测试实例引擎版本 6.0 测试实例地域和可用区 上海36 可用区1 压测机器的规格 c7.2xlarge.2 16核 32G 压测机器的操作系统 CTyunOS 2.0.121.06.4 64位 压测机器地域和可用区 上海36 可用区1 压测机器网络 与Redis实例为相同VPC区，与Redis实例可通过VPC连接 压测工具 redisbenchmark 读写分离 关闭

项目 说明 测试实例规格 基础读写分离8G(2副本) 测试实例引擎版本 7.0 测试实例地域和可用区 上海36 可用区1 压测机器的规格 c7.2xlarge.2 16核 32G 压测机器的操作系统 CTyunOS 2.0.121.06.4 64位 压测机器地域和可用区 上海36 可用区1 压测机器网络 与Redis实例为相同VPC区，与Redis实例可通过VPC连接 压测工具 redisbenchmark 读写分离 关闭

性能指标 说明 QPS Query Per Second，表示每秒处理的请求数，单位是次/秒。 延迟 操作的平均延迟时间，单位为毫秒（ms）。 %延迟 比如99%操作延迟，指99%操作的最大延迟时间，单位为毫秒（ms）。例如该指标的值为15毫秒，表示99%的请求可以在15毫秒内被处理。

Kafka 消费者有两个相关参数： enable.auto.commit：默认值为 true。 auto.commit.interval.ms： 默认值为 1000，也即 1s。 这两个参数组合的结果就是，每次 poll 数据前会先检查上次提交位点的时间，如果距离当前时间已经超过参数auto.commit.interval.ms规定的时长，则客户端会启动位点提交动作。 因此，如果将enable.auto.commit设置为 true，则需要在每次 poll 数据时，确保前一次 poll 出来的数据已经消费完毕，否则可能导致位点跳跃。 如果想自己控制位点提交，请把 enable.auto.commit 设为 false，并调用 commit(offsets)函数自行控制位点提交。

实例规格 CPU类型 数据大小 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 基础Proxy集群双副本24G(3分片) X86 32 1000 507884 1.175 1.783 2.775 基础Proxy集群双副本24G(3分片) X86 32 3000 499463 3.150 4.863 6.687 基础Proxy集群双副本24G(3分片) X86 32 10000 476099 10.611 15.175 18.463 基础Proxy集群双副本24G(3分片) X86 1024 1000 478262 1.434 2.415 4.423 基础Proxy集群双副本24G(3分片) X86 1024 3000 487626 3.406 5.519 8.751 基础Proxy集群双副本24G(3分片) X86 1024 10000 438596 11.701 16.359 20.095 增强Proxy集群双副本24G(3分片) X86 32 1000 841928 1.121 2.167 3.175 增强Proxy集群双副本24G(3分片) X86 32 3000 955338 2.833 6.967 10.167 增强Proxy集群双副本24G(3分片) X86 32 10000 892498 7.719 23.119 34.175

一个 Consumer Group 可以订阅多个 Topic。一个 Topic 也可以被多个 Consumer Group 订阅，且各个 Consumer Group 独立消费 Topic 下的所有消息。 举例：Consumer Group A 订阅了 Topic A，Consumer Group B 也订阅了 Topic A，则发送到 Topic A 的每条消息，不仅会传一份给 Consumer Group A 的消费实例，也会传一份给 Consumer Group B 的消费实例，且这两个过程相互独立，相互没有任何影响。

测试环境说明 项目 说明 测试实例规格 基础Cluster版双副本24G(3分片)。 测试实例引擎版本 6.0。 测试实例地域和可用区 上海36 可用区1。 压测机器的规格 c7.4xlarge.2。 压测机器的操作系统 CTyunOS 2.0.121.06.4 64位。 压测机器地域和可用区 上海36 可用区1。 压测机器网络 与Redis实例为相同VPC区，与Redis实例可通过VPC连接。 压测工具 redisbenchmark。 测试命令 redisbenchmark h ${host} p ${port} cluster a ${password} c ${connectionnum} n 20000000 d 32 r 2000000 t set,get threads 8 说明 参数说明，根据具体情况替换： ${host}: Redis连接地址 ， ${port}：Redis服务端口， ${password} : 密码 ， ${connectionnum}： 连接数。 测试结果 以下测试结果仅供参考，不同的机器环境和网络波动等客观条件可能产生性能差异。 性能指标说明： 性能指标 说明 QPS Query Per Second，表示每秒处理的请求数，单位是次/秒。 延迟 操作的平均延迟时间，单位为毫秒（ms）。 %延迟 比如99%操作延迟，指99%操作的最大延迟时间，单位为毫秒（ms）。例如该指标的值为15毫秒，表示99%的请求可以在15毫秒内被处理。 表1 SET 命令的测试结果 实例规格 CPU类型 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 基础Cluster版双副本24G(3分片) X86 1000 420300.53 2.299 3.159 3.687 基础Cluster版双副本24G(3分片) X86 3000 426721.28 6.760 9.415 12.143 基础Cluster版双副本24G(3分片) X86 10000 358937.53 25.451 39.487 211.583 增强Cluster版双副本24G(3分片) X86 1000 638814.38 1.347 1.855 2.279 增强Cluster版双副本24G(3分片) X86 3000 637999.25 4.053 6.167 7.535 增强Cluster版双副本24G(3分片) X86 10000 571037.06 14.612 23.407 26.831 表2 GET 命令的测试结果 实例规格 CPU类型 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 基础Cluster版双副本24G(3分片) X86 1000 456173.19 2.107 2.47 2.879 基础Cluster版双副本24G(3分片) X86 3000 445950.78 6.493 7.343 209.535 基础Cluster版双副本24G(3分片) X86 10000 402909.00 24.075 35.007 238.719 增强Cluster版双副本24G(3分片) X86 1000 746212.94 1.162 1.671 2.399 增强Cluster版双副本24G(3分片) X86 3000 738525.19 3.367 4.855 6.439 增强Cluster版双副本24G(3分片) X86 10000 652209.31 11.310 16.975 209.279

Kafka的消费者示例代码片段如下： 每个 Consumer Group 可以包含多个消费实例，即可以启动多个 Kafka Consumer，并把参数 group.id 设置成相同的值。属于同一个 Consumer Group 的消费实例会负载消费订阅的 Topic。 举例：Consumer Group A 订阅了 Topic A，并开启三个消费实例 C1、C2、C3，则发送到 Topic A 的每条消息最终只会传给 C1、C2、C3 的某一个。Kafka 默认会均匀地把消息传给各个消息实例，以做到消费负载均衡。 Kafka 负载消费的内部原理是，把订阅的 Topic 的分区，平均分配给各个消费实例。因此，消费实例的个数不要大于分区的数量，否则会有实例分配不到任何分区而处于空跑状态。这个负载均衡发生的时间，除了第一次启动上线之外，后续消费实例发生重启、增加、减少等变更时，都会触发一次负载均衡。 消息队列 Kafka 的每个 Topic 的分区数量默认是 16 个，已经足够满足大部分场景的需求，且云上服务会根据容量调整分区数。

消费消息 发送以下命令消费消息。 go run modvendor consumer/consumer.go 消费消息示例代码如下： package main import ( "encoding/json" "fmt" "github.com/confluentinc/confluentkafkago/kafka" "os" "path/filepath" ) type KafkaConfig struct { Topic string json:"topic" Topic2 string json:"topic2" GroupId string json:"group.id" BootstrapServers string json:"bootstrap.servers" SecurityProtocol string json:"security.protocol" } // config should be a pointer to structure, if not, panic func loadJsonConfig() KafkaConfig { workPath, err : os.Getwd() if err ! nil { panic(err) } configPath : filepath.Join(workPath, "conf") fullPath : filepath.Join(configPath, "kafka.json") file, err : os.Open(fullPath); if (err ! nil) { msg : fmt.Sprintf("Can not load config at %s. Error: %v", fullPath, err) panic(msg) } defer file.Close() decoder : json.NewDecoder(file) var config &KafkaConfig{} err decoder.Decode(config); if (err ! nil) { msg : fmt.Sprintf("Decode json fail for config file at %s. Error: %v", fullPath, err) panic(msg) } json.Marshal(config) return config } func doInitConsumer(cfg KafkaConfig) kafka.Consumer { fmt.Print("init kafka consumer, it may take a few seconds to init the connectionn") //common arguments var kafkaconf &kafka.ConfigMap{ "api.version.request": "true", "auto.offset.reset": "latest", "heartbeat.interval.ms": 3000, "session.timeout.ms": 30000, "max.poll.interval.ms": 120000, "fetch.max.bytes": 1024000, "max.partition.fetch.bytes": 256000} kafkaconf.SetKey("bootstrap.servers", cfg.BootstrapServers); kafkaconf.SetKey("group.id", cfg.GroupId) switch cfg.SecurityProtocol { case "PLAINTEXT" : kafkaconf.SetKey("security.protocol", "plaintext"); case "SSL": kafkaconf.SetKey("security.protocol", "ssl"); kafkaconf.SetKey("ssl.ca.location", "/XXX/caRoot.pem") case "SASLSSL": kafkaconf.SetKey("security.protocol", "saslssl"); kafkaconf.SetKey("ssl.ca.location", "/XXX/caRoot.pem"); kafkaconf.SetKey("sasl.username", cfg.SaslUsername); kafkaconf.SetKey("sasl.password", cfg.SaslPassword); kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism) case "SASLPLAINTEXT": kafkaconf.SetKey("security.protocol", "saslplaintext"); kafkaconf.SetKey("sasl.username", cfg.SaslUsername); kafkaconf.SetKey("sasl.password", cfg.SaslPassword); kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism) default: panic(kafka.NewError(kafka.ErrUnknownProtocol, "unknown protocol", true)) } consumer, err : kafka.NewConsumer(kafkaconf) if err ! nil { panic(err) } fmt.Print("init kafka consumer successn") return consumer; } func main() { // Choose the correct protocol cfg : loadJsonConfig(); consumer : doInitConsumer(cfg) consumer.SubscribeTopics([]string{cfg.Topic, cfg.Topic2}, nil) for { msg, err : consumer.ReadMessage(1) if err nil { fmt.Printf("Message on %s: %sn", msg.TopicPartition, string(msg.Value)) } else { // The client will //automatically try to recover from all errors. fmt.Printf("Consumer error: %v (%v)n", err, msg) } } consumer.Close() }

场景 说明 网络故障或连接中断 网络连接中断或不稳定，导致无法正常与 Redis 服务器通信 故障触发了主从切换 因Redis底层硬件或其他原因导致主节点故障后，会触发主从切换，保障实例仍可用，主从切换会有以下影响：秒级的连接闪断最多30秒的只读 Redis 服务器负载过高 Redis 服务器负载过高，无法及时响应请求 Redis 命令执行失败 Redis 命令执行失败，可能由于写入冲突或 Redis 服务器资源耗尽等原因导致 慢查询引起了请求堵塞 执行时间复杂度为O(N)的操作，引发慢查询和请求的堵塞，此时，客户端发起的其他请求可能出现暂时性失败。 Redis 服务器重启或升级 Redis 服务器进行重启或升级，期间无法正常处理请求 Redis 配置错误 Redis 配置错误或使用了不兼容的命令，导致操作失败

对实例的一些操作，如实例开通、扩容等，会启动一个后台任务，您可以在Redis管理控制台的任务中心页面，查看该任务的状态等信息，同时支持删除任务信息。 操作步骤 1. 登录 Redis管理控制台。 2. 在管理控制台左上角选择实例所在的区域。 3. 左边菜单点击任务中心，进入任务中心页面。 4. 通过筛选时间、任务类型、实例ID等可查询任务信息。 5. 点击删除可删除任务信息。

Redis数据迁移完成后（即迁移结束），可以通过redisfullcheck工具检查源Redis与目标Redis的数据一致性（源Redis数据为目标Redis数据的子集，即表示数据一致）。 前提条件 数据迁移已完成（即迁移结束）。 准备工作 开通一台ECS云主机，要求能在ECS上访问通源和目标Redis实例。 下载redisfullcheck执行文件（建议：1.4.8版本，示例：redisfullcheck1.4.8.tar.gz），用以接下来校验源和目标Redis数据。 操作步骤 1. 登录ECS云主机，进入redisfullcheck解压目录。 2. 解压redisfullcheck1.4.8.tar.gz（tar xvf redisfullcheck1.4.8.tar.gz），解压后的redisfullcheck即为可执行文件。 3. 命令说明： 参考命令： ./redisfullcheck s {源Redis连接地址} p {源Redis密码} t {目标Redis连接地址} a {目标Redis密码} 实际执行： ./redisfullcheck s sourceaddress p sourcepassword t targetadress a targetpassword 说明：源/目标连接地址为集群时，多个连接地址以英文分号（;）分隔 4. 命令参数（可选）说明： 选项 说明 示例 s 或者 source 源Redis连接地址，如果是集群时，多个地址间以英文分号（;）分隔。 s sourceaddress1;sourceaddress2 p 或者 sourcepassword 源Redis连接密码（不支持指定username）。 p password sourcedbtype 源Redis架构系列；0：标准主备/单机（默认），1：集群，2: aliyun proxy, 3: tencent proxy。 sourcedbtype 0 sourcedbfilterlist 源Redis需要对比的db，1：表示对比所有db（默认），db1;db2：表示对比db1和db2。 sourcedbfilterlist 0;1 t 或者 target 目标Redis连接地址，如果是集群时，多个地址间以英文分号（;）分隔。 t targetaddress1;targetaddress2 a 或者 targetpassword 目标Redis连接密码（不支持指定username）。 a password targetdbtype 目标Redis架构系列；0：标准主备/单机（默认），1：集群，2: aliyun proxy, 3: tencent proxy。 targetdbtype 0 targetdbfilterlist 目标Redis需要对比的db，1：表示对比所有db（默认），db1;db2：表示对比db1和db2。 targetdbfilterlist 1 comparetimes 对比轮数，默认：3，最小：1。 comparetimes 3 m 或者 comparemode 校验模式，1：全量比较，2：只对比value的长度（默认），3：只校验key是否存在，4：全量比较，但是忽略大key的比较。 m 2 q 或者 qps 最大的qps限制（默认：15000）。 q 15000 interval 相邻两轮校验的间隔时间（秒）（默认：5）。 interval 5 loglevel 日志级别，'debug', 'info', 'warn', 'error'（默认：info）。 loglevel info f 或者 filterlist 需要比较的key列表，以竖线分割。 f abc v 或者 version 输出redisfullcheck版本。 v 5. 运行结果 出现以下字样时，表示校验完成。 all finish successfully, …… 校验明细。 sqlite3 result.db.3（3表示对比轮数） select from key; select from field; 说明：表key，field记录了数据冲突情况

批量修改权限 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“用户管理”后进入应用用户管理页面。 （5）点击“导入用户”右侧的下拉倒三角的“批量修改权限”后，出现如下上传文件界面，文件格式件批量下载说明。 （6）点击“上传”完成批量创建。 模板如下。 表：参数说明。 参数 说明 :: type (produce/consume) 权限类型，produce或者consume username 用户名 topicName topic名称

参数 说明 权限 ACL 策略的操作权限分为两类：允许和拒绝。 若只设置允许规则，则除允许的规则外的其他IP或网段都无法连接实例。 若只设置拒绝规则，则除拒绝的规则外的其他IP或网段需要设置了允许规则后才可以连接实例。 若同时设置允许规则和拒绝规则，则只有允许规则中的IP或网段可以连接实例，其他IP或网段都无法连接实例。 用户 选择需要设置权限的用户。 IP或网段 填写需要设置权限的 IP 或网段，用 ; 隔开，若 IP 为空，则默认为全部 IP 添加权限。 操作 选择策略生效的动作，即向 Topic 生产或消费消息。 自动应用后续所有新增topic 开启后，后续页面上创建的Topic会自动关联此规则，多个策略只允许一条策略开启。

迁移后验证 迁移完成后，请使用Rediscli连接源Redis和目标Redis，验证数据的完整性。 查看非cluster原生集群版 数据情况 ./rediscli h $host p $port a $psw dbsize 查看cluster原生集群版 数据情况 ./rediscli a $psw cluster call $host:$port dbsize 说明 如果无增量同步，则迁移过程中源Redis写操作不会同步到目标Redis。

可选方式 天翼云缓存实例目前支持[实例克隆] 与[从对象存储导入]两种方式，离线迁移数据至目标实例： 1. 如果源实例与目标实例处于同VPC同子网时，离线迁移请参考：[实例克隆] 2. 如果源实例与目标实例不处于同VPC同子网时，离线迁移请参考：[从对象存储导入] 约束 1. 如果源库或者目标库开启了SSL功能，可能会导致迁移或者迁移发起失败，迁移前请检查SSL是否处于关闭状态 2. 如果目标库实例没有足够的内存空间存储备份文件数据，可能导致离线迁移失 3. 目标实例的内核版本需不低于源库实例的内核版本，否则可能因为内核兼容性问题而导致迁移或者迁移发起失败 4. 备份文件需未进行加密 以下将分别按照 [实例克隆] 与 [从对象存储导入]两类离线迁移方式进行介绍 实例克隆 1. 前提条件 源Redis实例已执行数据备份并备份成功。 2. 操作 实例克隆 从对象存储导入 1. 前提条件 开通天翼云对象存储，对象存储所在资源池需与目标实例所在资源池相同 源Redis实例已执行数据备份并备份成功 下载备份文件至本地，然后从本地上传备份文件至对象存储（如果源实例和对象存储处于同一资源池，可以直接备份数据至对象存储中，无需进行显示上传）

迁移场景 工具 方案 迁移说明 自建Redis迁移至DCS DCS控制台数据迁移功能 使用在线迁移自建Redis 不同Region，可开通 云间高速 ，打通网络 同Region不同VPC，可开通 对等连接 ，打通网络 自建Redis迁移至DCS Rediscli 使用Rediscli迁移自建Redis（AOF文件） 自建Redis迁移至DCS Rediscli 使用Rediscli迁移自建Redis（RDB文件） 自建Redis迁移至DCS RedisShake 使用RedisShake工具迁移自建Redis Cluster集群 其他云厂商Redis服务迁移至DCS DCS控制台数据迁移功能 使用在线迁移其他云厂商Redis 其他云厂商Redis服务迁移至DCS RedisShake 使用RedisShake工具离线迁移其他云厂商Redis Cluster集群 DCS实例间迁移 DCS控制台数据迁移功能 低版本Redis迁移至高版本Redis实例。 如网络互通，建议使用在线迁移功能迁移Redis实例数据 如网络不通，建议打通网络后，再使用在线迁移功能迁移Redis实例数据 1、网络连通 不同Region，可开通 云间高速 ，打通网络 同Region不同VPC，可开通 对等连接 ，打通网络 2、高版本Redis迁移至低版本Redis实例 不建议，否则极易出现命令兼容问题，导致迁移中断 DCS实例间迁移 DCS控制台数据迁移功能 不同Region的Redis实例迁移，建议打通网络后，再使用在线迁移功能迁移Redis实例数据 1、网络连通 不同Region，可开通 云间高速 ，打通网络 同Region不同VPC，可开通 对等连接 ，打通网络 2、高版本Redis迁移至低版本Redis实例 不建议，否则极易出现命令兼容问题，导致迁移中断 DCS实例间迁移 DCS控制台数据迁移功能 不同账号的Redis实例迁移，建议使用在线迁移功能迁移Redis实例数据 1、网络连通 不同Region，可开通 云间高速 ，打通网络 同Region不同VPC，可开通 对等连接 ，打通网络 2、高版本Redis迁移至低版本Redis实例 不建议，否则极易出现命令兼容问题，导致迁移中断 DCS实例间迁移 DCS控制台数据迁移功能 同Region不同VPC下的Redis实例迁移，建议打通网络后，再使用在线迁移功能迁移Redis实例 1、网络连通 不同Region，可开通 云间高速 ，打通网络 同Region不同VPC，可开通 对等连接 ，打通网络 2、高版本Redis迁移至低版本Redis实例 不建议，否则极易出现命令兼容问题，导致迁移中断 DCS实例迁移下云 DCS控制台数据迁移功能 DCS实例迁移下云 建议打通网络后，使用线迁移功能进行数据迁移

本节介绍如何使用的安全接入点接入Kafka的方法，文档以Java代码为例。 前提条件 已配置正确的安全组。 已获取连接Kafka实例的地址。 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。 已创建弹性云服务器，如果使用内网同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。 需要用户先在用户管理页面创建用户，然后给对应的topic授予生产消费权限。 使用内网同一个VPC访问，实例端口为8092，实例连接地址从控制台实例详情菜单处获取，如下图所示。 Maven中引入Kafka客户端 java Kafka实例基于社区版本2.8.2/3.6.2，推荐客户端保持一致。 org.apache.kafka kafkaclients 2.8.2/3.6.2 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLPLAINTEXT"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"Kafka@Test";"); 生产者代码示例 java package com.justin.kafka.service.gw.sasl; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class Producer { private final KafkaProducer producer; public final static String TOPIC "testtopic2"; public final static String BROKERADDR "192.168.0.11:8092,192.168.0.9:8092,192.168.0.10:8092"; public Producer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLPLAINTEXT"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"Kafka@Test";"); props.put(ProducerConfig.ACKSCONFIG, "all"); props.put("retries",3); producer new KafkaProducer<>(props); } public void produce() { try { for (int i 0; i (TOPIC, data), new Callback() { public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception ! null) { // TODO: 异常处理 exception.printStackTrace(); return; } System.out.println("produce msg completed, partition id " + metadata.partition()); } }); } } catch (Exception e) { // TODO: 异常处理 e.printStackTrace(); } producer.flush(); producer.close(); } public static void main(String[] args) { Producer producer new Producer(); producer.produce(); } }

本节介绍如何使用的公共接入点接入Kafka的方法，文档以Java代码为例。 前提条件 已配置正确的安全组。 已获取连接Kafka实例的地址。 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。 已创建弹性云服务器，如果使用内网同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。 使用内网同一个VPC访问，实例端口为8090，实例连接地址从控制台实例详情菜单处获取，如下图所示。 Maven中引入Kafka客户端 java Kafka实例基于社区版本2.8.2/3.6.2，推荐客户端保持一致。 org.apache.kafka kafkaclients 2.8.2/3.6.2 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); 生产者代码示例 java package com.justin.kafka.service.gw.plain; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class Producer { private final KafkaProducer producer; public final static String TOPIC "testtopic"; public final static String BROKERADDR "192.168.0.11:8090,192.168.0.9:8090,192.168.0.10:8090"; public Producer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.ACKSCONFIG, "all"); props.put("retries",3); producer new KafkaProducer<>(props); } public void produce() { try { for (int i 0; i (TOPIC, data), new Callback() { public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception ! null) { // TODO: 异常处理 exception.printStackTrace(); return; } System.out.println("produce msg completed, partition id " + metadata.partition()); } }); } } catch (Exception e) { // TODO: 异常处理 e.printStackTrace(); } producer.flush(); producer.close(); } public static void main(String[] args) { Producer producer new Producer(); producer.produce(); } }

为IAM用户授权 步骤 1 管理员使用已注册的天翼云帐号登录天翼云网门户。 步骤 2 鼠标移动至天翼云首页右上角用户头像，在下拉列表中单击“个人中心”。 步骤 3 个人中心左侧菜单中，单击“主子账号及授权管理”。 步骤 4 在主子账号及授权管理页，单击左侧导航菜单中的“用户组”。

支持的迁移类型 常见场景 从单可用区迁移至单可用区 将实例迁移至ECS所属的可用区，同一可用区ECS和实例通过内网连接拥有更低的网络延迟。 从双可用区迁移至双可用区 将实例迁移至ECS所属的可用区，同一可用区ECS和实例通过内网连接拥有更低的网络延迟。 从单可用区迁移至双可用区 提高实例的容灾能力，实现跨机房容灾。 单可用区实例可以承受服务器和机架级别的故障，而多可用区实例通过跨机房部署可抵御机房级别的故障，从而提升容灾能力。 从双可用区迁移至单可用区 满足特定功能的要求。

本节主要介绍如何删除实例数据 Redis管理控制台支持删除实例数据。删除实例数据操作无法撤销，且实例数据被删除后将无法恢复，请谨慎操作。 前提条件 只有当Redis实例处于“运行中”状态，才能执行此操作。 操作步骤 1. 登录 Redis管理控制台。 2. 在管理控制台左上角选择实例所在的区域。 3. 在实例列表页，单击目标实例名称进入实例详情管理。 4. 左侧菜单点击CloudDBA>Redis节点管理，进入Redis节点管理界面。 5. 在Redis节点管理界面，点击上方按钮清空实例数据，可删除实例数据。

实例概览页面可以帮助用户直观的查看实例资源的使用情况。 包括租户的实例数和实例状态，并提供便利的统计信息，包括当前资源池的实例的内存使用量、使用率、KEY总数、QPS等。 操作步骤 1. 登录 Redis管理控制台。 2. 在管理控制台左上角选择需要概览所在的区域 3. 点击左侧菜单的“实例概览” 4. 页面分为两个部分: 包含全局的“我的资源” 以及只有当前资源池实例的信息。

实例ID 每个实例的标识 实例密码 auth的认证密码。集群版实例密码为：实例名实例密码；标准版实例：实例密码。 DB Redis中的Database，范围[0,255]。 Redis连接数 接入机与每个redis节点之间的连接数。 空闲Redis连接数 接入机与每个redis节点的空闲连接数。 客户端最大连接数 客户端可创建的最大连接数。

命令类型 命令 Strings MSETNX Set SINTER Set SINTERSTORE Set SUNION Set SUNIONSTORE Set SDIFF Set SDIFFSTORE Set SMOVE Sortedset ZUNIONSTORE Sortedset ZINTERSTORE HyperLogLog PFCOUNT HyperLogLog PFMERGE Keys RENAME Keys RENAMENX Keys BITOP Keys RPOPLPUSH

对比项 单机/主备 Proxy集群 Cluster集群 读写分离 兼容Redis版本 基础版兼容开源Redis 5.0、6.0、7.0版本。经典版兼容开源Redis 2.8、4.0、5.0版本。 经典版兼容开源Redis2.8、4.0、5.0版本，有部分命令限制，请参考命令限制。 兼容开源RedisCluster 6.0、7.0版本。 兼容开源Redis 6.0、7.0版本。 特性支持与限制 完全支持开源单机主备使用特性 Proxy集群实现架构转换，基本支持开源单机主备使用特性，但在特殊命令使用上有限制，请参考 Cluster集群完全支持开源社区RedisCluster的使用特性。业务使用RedisCluster的key需要在统一slot槽段范围内，且在使用event notify，scan、keys等命令时需要与每一分片建立连接获取请求结果。 通过将读请求分散到多个从节点上，可以显著提高系统的整体读取吞吐量。主从复制存在一定的延迟，不适用于强一致性的场景。 客户端协议 使用传统Jedis客户端即可。 使用传统Jedis客户端即可，不需要支持Redis Cluster协议。 需要客户端支持Redis Cluster协议。 使用传统Jedis客户端即可。 命令限制 不支持的Redis命令，请参考 不支持的Redis命令，请参考 Cluster集群实例100%支持开源社区集群的命令。 不支持的Redis命令，请参考 副本数 单机实例为单副本，只有一个节点。单副本表示实例只有主节点，无法保障数据高可靠。主备默认为双副本，默认为一主一从的架构。双副本支持高可用切换。 Proxy集群单机实例为单副本，只有一个节点。Proxy集群主备默认为双副本，默认为一主一从的架构。 Cluster集群单机实例为单副本，只有一个节点。Cluster集群主备默认为双副本，默认为一主一从的架构。 多个副本，一主多从架构。支持高可用切换。

Kafka服务端支持版本是多少？ Kafka 1.1.0和2.3.0版本。 创建的Kafka实例是集群模式么？ 创建一个Kafka实例即为一个集群实例。 Kafka实例是否支持修改访问端口？ Kafka实例的访问端口固定，不支持修改。 如果是访问未开启SASL的Kafka专享实例，访问端口为9092。 如果是访问开启SASL的Kafka专享实例，访问端口为9093。 在访问Kafka实例之前，需要确保安全组是否配置正确。 Kafka实例的SSL证书有效期多长？ Kafka实例开启SASL时，需进行单向认证，证书有效期足够长（超过15年），客户端不需要关注证书过期风险。 如何将Kafka实例中的数据同步到另一个Kafka实例中？ Kafka实例之间没有好的实时同步方案，如果需要做实例迁移，可以同时向两个实例生产消息，源实例中的消息可继续消费，待源实例的消息数据全部被消费完或老化后，业务可迁移到新的Kafka实例。 Kafka实例的SASLSSL开关如何修改？ Kafka SASLSSL开关不支持创建实例后修改，在创建时，请慎重选择，如果创建后需要修改，需要重新创建实例。 购买实例时选择的单AZ，怎样可以扩展为多AZ？ 已购买的实例无法扩展AZ，请重新购买多AZ的实例。 Kafka扩容会影响业务吗？ Kafka扩容带宽/存储空间，都不会影响业务的使用。 Kafka实例创建后，能修改VPC和子网吗？ 不能修改VPC和子网。 Kafka实例版本可以升级吗？ Kafka实例创建成功后，实例版本不支持升级。您可以重新创建Kafka实例，实现升级Kafka实例的版本。

对于包周期订购后，支持通过控制台进行退订操作。 具体操作如下： 步骤1：登录Kafka专享版控制台，在实例列表中，右侧的“更多”菜单下拉选择“退订”按钮； 步骤2：进入退订管理界面后，按页面提示选择“退订原因”，并勾选“我已同意本次退订金额及相关费用详情”，点击“退订”按钮完成后续流程。