实例规格 CPU类型 数据大小 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 最大延迟(ms) 基础标准版双副本8G X86 32 100 136349.38 0.720 0.751 0.903 209.919 基础标准版双副本8G X86 32 500 141596.92 3.500 4.311 5.007 418.815 基础标准版双副本8G X86 32 1000 143225.44 6.944 8.455 10.703 822.271 基础标准版双副本8G X86 1024 100 82166.57 1.203 1.343 5.775 721.919 基础标准版双副本8G X86 1024 500 82282.86 5.960 9.263 41.663 3000.319 基础标准版双副本8G X86 1024 1000 81445.16 11.943 18.575 212.991 3000.319 增强标准版双副本8G X86 32 100 220726.19 0.437 0.567 0.639 13.279 增强标准版双副本8G X86 32 500 233426.70 2.008 2.327 2.511 8.959 增强标准版双副本8G X86 32 1000 230606.05 3.957 4.775 5.487 217.471 增强标准版双副本8G X86 1024 100 90125.00 1.021 1.031 2.743 3000.319 增强标准版双副本8G X86 1024 500 96142.75 4.724 5.687 20.511 3000.319 增强标准版双副本8G X86 1024 1000 88639.91 10.971 16.623 73.599 2390.015

本节介绍如何处理消息去重。 方案概述 在RocketMQ的业务处理过程中，如果消息重发了多次，消费者端对该重复消息消费多次与消费一次的结果是相同的，多次消费并没有对业务产生负面影响，那么这个消息处理过程是幂等的。消息幂等保证了无论消息被重复投递多少次，最终的处理结果都是一致的，避免了因消息重复而对业务产生影响。 例如在支付场景下，用户购买商品后进行支付，由于网络不稳定导致用户收到多次扣款请求，导致重复扣款。但实际上扣款业务只应进行一次，商家也只应产生一条订单流水。这时候使用消息幂等就可以避免这个问题。 在实际应用中，导致消息重复的原因有网络闪断、客户端故障等，且可能发生在消息生产阶段，也可能发生在消息消费阶段。因此，可以将消息重复的场景分为以下两类： 生产者发送消息时发生消息重复： 生产者发送消息时，消息成功发送至服务端。如果此时发生网络闪断，导致生产者未收到服务端的响应，此时生产者会认为消息发送失败，因此尝试重新发送消息至服务端。当消息重新发送成功后，在服务端中就会存在两条内容相同的消息，最终消费者会消费到两条内容一样的重复消息。 消费者消费消息时发生消息重复： 消费者消费消息时，服务端将消息投递至消费者并完成业务处理。如果此时发生网络闪断，导致服务端未收到消费者的响应，此时服务端会认为消息投递失败。为了保证消息至少被消费一次，服务端会尝试投递之前已被处理过的消息，最终消费者会消费到两条内容一样的重复消息。

本章节介绍了如何通过云审计服务来查看云审计日志。 查看DMS for RocketMQ云审计日志，请参考查看追踪事件。

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

本文介绍如何使用telnet命令检测Redis端口连通性 ECS实例中已经安装了Telnet（Linux）或开启了Telnet客户端（Windows）。如果Redis服务出现了连接问题，并且参见上述问题中如何检测弹性云主机与Redis之间的网络连接。发现连接成功时，您需要进一步使用telnet命令检测服务端口是否可用。 查看Redis实例的连接地址，详情请参见查看连接地址。 登录ECS实例，执行telnet命令。 telnet {IP} {Redis PORT} 说明 windows系统和Linux系统中都可以使用该命令。 返回信息分析： 如果Redis连接存在问题，但可以在ECS上使用telnet连接到Redis实例，则ECS本身与Redis之间的连接无异常，请排查其它因素，例如客户端、业务代码，以及业务环境导致的Redis服务阻塞等问题。 如果telnet连接失败，但使用ping命令检测ECS与Redis之间的连接成功，可能是由于ECS存在异常行为（例如受恶意程序影响而攻击其它Redis的33016端口等）被系统禁止了部分服务，此时建议您监控ECS的数据找到异常流量并加以处理。

本文介绍Redis key丢失原因 通常情况下Redis不会主动丢失数据，若出现key丢失的情况，请按照以下步骤排查 1. 查看key是否过期。过期key自动清理。 2. 查看监控，确认是否会触发键逐出机制。 3. 去服务端分析info查看是否有删除key的操作。

本节介绍Kafka实例的连接地址数量 Kafka实例的连接地址个数和实例的节点数有关，连接地址个数等于节点数。 各规格节点数信息如下，详细请参考产品规格。

对比项 Cluster集群 Proxy集群 原生兼容性 高 中 客户端兼容性 中 高 性价比 高 中 时延 低时延 中等时延 性能 高 中

发生消费位点重置的两种情况 当服务端不存在曾经提交过的位点时（比如客户端第一次上线）； 当从非法位点拉取消息时（比如某个分区最大位点是10，但客户端却从11开始拉取消息）。 配置重置策略 Java 客户端可以通过auto.offset.reset来配置重置策略，主要策略有： latest：从最大位点开始消费； earliest：从最小位点开始消费； none：不做任何操作，也即不重置。 策略配置建议 强烈建议设置成“latest”，而不要设置成“earliest”，避免因位点非法时从头开始消费，从而造成大量重复 如果是客户自己管理位点，可以设置成”none”； 拉取大消息。 拉取消息注意事项 消费过程是由客户端主动去服务端拉取消息的，在拉取大消息时，需要注意控制拉取速度，注意修改配置。 max.poll.records：如果单条消息超过1MB，建议这里设置为1； fetch.max.bytes：设置比单条消息的大小略大一点； max.partition.fetch.bytes：设置比单条消息的大小略大一点。拉取大消息的核心是一条一条拉。

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

实例规格 CPU类型 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 基础Cluster版双副本24G(3分片) X86 1000 453905 1.833 2.615 3.439 基础Cluster版双副本24G(3分片) X86 3000 450359 5.568 8.647 9.999 基础Cluster版双副本24G(3分片) X86 10000 340147 27.503 38.303 236.159

实例规格 CPU类型 数据大小 连接数 QPS 平均延迟(ms) 95%延迟(ms) 99%延迟(ms) 最大延迟(ms) 基础标准版双副本8G X86 32 100 134622.12 0.729 0.735 0.871 417.535 基础标准版双副本8G X86 32 500 141240.94 3.520 4.319 5.047 826.367 基础标准版双副本8G X86 32 1000 143733.92 6.931 8.191 10.159 223.871 基础标准版双副本8G X86 1024 100 79699 1.235 1.391 4.991 3000.319 基础标准版双副本8G X86 1024 500 81616.66 5.960 8.943 39.967 3000.319 基础标准版双副本8G X86 1024 1000 84454.47 11.807 17.375 65.983 2815.999

参数 说明 实例名称 DCS中的实例名称，下拉列表选择（源Redis、目标Redis必须有一个是DCS实例)。 Cluster集群 当前集群是否是Cluster原生集群（不填写实例名称时填写）。 实例连接地址 IP:PORT（不填写实例名称时填写）。 实例账号/密码 Redis连接账号/密码，需有DBA管理权限。 key冲突模式 跳过目标key，继续执行/覆盖目标key，继续执行/中断迁移。 同步模式 全量同步+增量同步/全量同步（源Redis未放通PSYNC命令时，不支持增量同步）。

本节介绍消费端挂载NFS是否会影响消费速度 消费端在消费消息的主线程里同步将拉取的消息存储在NFS，导致消费端处理消息的速度变慢，阻塞消息处理。 可能原因 NFS本身速度就不太理想。 NFS是网络共享存储，虽然有多机器共享访问的能力优势，但多台机器访问是争抢的，消费者个数增多，性能反而下降。 解决方案 建议将消费端拉取消息和存储消息分别放在两个独立且不同的线程里操作。拉取消息的线程只管消费消息，把消息转给缓存处理线程后就继续消费消息，这样可以保证消费速度的稳定。 也可以考虑采用云盘，给每台消费端处理机挂载自己的云盘，各自独立存储，这样消费端不会再因为争抢NFS而降低性能。如果需要把最终的处理结果集中到同一个NFS上保存，仍然可以通过一个异步的工具或者线程，把云盘上的结果再转发到NFS上，而不要让同步存储NFS阻塞消息处理。总之，对于资源访问造成的处理低效，总是可以用异步处理的方式解决。

对比项 单机/主备 Proxy集群 Cluster集群 兼容Redis版本 100%兼容开源Redis 2.8、4.0、5.0、6.0、7.0。可在购买实例时选择版本号。 兼容开源Redis2.8、4.0、5.0版本，有部分命令限制，请参考命令限制。 100%兼容开源RedisCluster 6.0/7.0版本。可在购买实例时选择版本号。 特性支持与限制 完全支持开源单机主备使用特性 Proxy集群实现架构转换，基本支持开源单机主备使用特性，但在特殊命令使用上有限制，请参考 Cluster集群完全支持开源社区RedisCluster的使用特性。业务使用RedisCluster的key需要在统一slot槽段范围内，且在使用event notify，scan、keys等命令时需要与每一分片建立连接获取请求结果。 客户端协议 使用传统Jedis客户端即可。 使用传统Jedis客户端即可，不需要支持Redis Cluster协议。 需要客户端支持Redis Cluster协议。 命令限制 不支持的Redis命令，请参考 不支持的Redis命令，请参考 Cluster集群实例100%支持开源社区集群的命令。 副本数 单机实例为单副本，只有一个节点。单副本表示实例只有主节点，无法保障数据高可靠。主备默认为双副本，默认为一主一从的架构。双副本支持高可用切换 Proxy集群单机实例为单副本，只有一个节点。Proxy集群主备默认为双副本，默认为一主一从的架构。 Cluster集群单机实例为单副本，只有一个节点。Cluster集群主备默认为双副本，默认为一主一从的架构。

功能说明 当实例磁盘使用率达到指定阈值后，可选择磁盘自动扩容或动态消息保留策略来调整磁盘空间。 磁盘自动扩容 当磁盘使用率达到指定阈值后，系统会自动根据扩容策略增加磁盘容量，避免出现实例因磁盘容量不足产生的故障。 动态消息保留策略 当磁盘使用率达到指定阈值后，系统会自动根据消息保留策略删除一定比例的主题最早消息数据，释放磁盘空间。 约束与变更影响 动态消息保留策略和磁盘自动扩容同时最多只能开启一个。 磁盘自动扩容不会影响业务，如果短时间内磁盘使用量迅速增加，无法保证在耗尽磁盘空间前完成磁盘自动扩容。建议提前做好容量规划，先根据磁盘使用量做好扩容，再开启磁盘自动扩容。 动态消息保留策略触发后会删除最早的主题消息，可能导致消费者未及时消费的消息被删除，且调整效果存在一定延迟。建议及时消费数据，减少数据堆积。 触发动态消息保留策略后，主题的保留时长参数会被按比例减小且不会自动恢复，如后续需要调整请到主题管理页面进行操作。 前提条件 实例磁盘字段扩容会产生相应的费用，请确保账户余额充足。

本文介绍缓存实例的数据被删除找回问题 如果缓存实例执行了备份操作，实例的数据被删除后，可通过备份文件对数据进行恢复，但是恢复会覆盖备份时间到恢复这段时间的写入数据。 实例通过控制台的“备份与恢复”功能将已备份的数据恢复到缓存实例中，参考备份还原 另外，如果缓存实例被删除，实例中原有的数据将被删除，实例的备份数据也会删除，请谨慎操作。

本文介绍创建Redis自定义策略。 如果系统预置的Redis权限，不满足您的授权要求，可以创建自定义策略。自定义策略中可以添加的授权项请参考主子账号和IAM权限管理。 目前天翼云支持可视化视图和JSON视图两种方式设置策略内容，详细介绍请查看统一身份认证用户指南权限管理自定义策略 创建自定义策略。 可视化视图：通过可视化视图创建自定义策略，无需了解JSON语法，按可视化视图导航栏选择云服务、操作、资源、条件等策略内容，可自动生成策略。 JSON视图：通过JSON视图创建自定义策略，可以在选择策略模板后，根据具体需求编辑策略内容；也可以直接在编辑框内编写JSON格式的策略内容。 Redis自定义策略样例 以下策略样例表示：仅允许IAM用户使用redis启动、停止功能。 { "Version": "1.1", "Statement": [ { "Effect": "Allow", "Action": [ "DCS2:inst:redisstart", "DCS2:inst:redisstop" ] } ] }

业务场景 在一个社交媒体平台中，我们需要实时统计用户的点赞数和在线人数。为了满足实时性和高并发的需求，我们决定使用 Redis 来实现一个实时计数器。 业务需求 实时统计用户的点赞数，以便在用户界面上显示点赞数的实时更新。 实时统计在线用户人数，以便在管理后台实时监控在线用户的数量。 需求分析 计数器需要能够实时更新，并能够处理高并发的请求。 计数器的数据需要持久化，以防止数据丢失。 实现方案 使用 Redis 的计数器功能来存储和更新点赞数和在线用户数。 使用 Redis 的键过期功能来自动清除超时的在线用户。 使用 Redis 的持久化功能，如 AOF 或 RDB，来保证数据的持久性。 使用 Redis 的事务功能来确保计数器的原子性更新。 示例代码 以下是一个简化的示例代码，展示了如何使用 Redis 实现实时计数器的关键部分： // 用户点赞操作 public void likePost(String postId, String userId) { Jedis jedis new Jedis("localhost"); jedis.incr("postlikes:" + postId); // 增加点赞数 jedis.sadd("postlikedusers:" + postId, userId); // 添加点赞用户 jedis.close(); } // 获取帖子的点赞数 public long getPostLikes(String postId) { Jedis jedis new Jedis("localhost"); long likes Long.parseLong(jedis.get("postlikes:" + postId)); jedis.close(); return likes; } // 用户上线操作 public void userOnline(String userId) { Jedis jedis new Jedis("localhost"); jedis.setex("useronline:" + userId, 300, "true"); // 设置用户在线标志，并设置过期时间为 300 秒 jedis.close(); } // 获取在线用户数 public long getOnlineUserCount() { Jedis jedis new Jedis("localhost"); long count jedis.keys("useronline:").size(); jedis.close(); return count; } 在上述示例代码中，我们使用 Redis 的计数器功能来存储点赞数，并使用 Redis 的集合数据类型来存储点赞的用户。对于在线用户数的统计，我们使用 Redis 的键过期功能来设置用户在线标志的过期时间，并使用 Redis 的键查询功能来获取在线用户数。

本文介绍如何使用Redisdesktopmanager访问Redis实例 通过VPC子网访问 Redis。 填写连接信息： 打开 Redis Desktop Manager 并选择连接类型为 Redis。 在连接配置中，填写 DCS 实例的子网地址、端口，以及相应的密码。 测试连接： 在填写完连接信息后，单击界面左下角的“测试连接”按钮。 确认连接正常： 如果提示连接成功，说明连接已经建立成功。 操作完成。

环境安装 1. 安装Python。（Python版本为2.7或3.X。） 2. 安装依赖库。（使用公网连接需要安装confluentkafka 1.9.2及以下版本的依赖库） pip install confluentkafka1.9.2 3. 下载Demo包kafkaconfluentpythondemo.zip。 配置修改 1. 如果是ssl连接，需要在控制台下载证书。并且解压压缩包得到ssl.client.truststore.jks，执行以下命令生成caRoot.pem文件。 keytool importkeystore srckeystore ssl.client.truststore.jks destkeystore caRoot.p12 deststoretype pkcs12 openssl pkcs12 in caRoot.p12 out caRoot.pem 2. 修改setting.py文件。（calocation仅在ssl连接时需要配置） kafkasetting { 'bootstrapservers': 'XXX', 'topicname': 'XXX', 'groupname': 'XXX' } 生产消息 发送以下命令发送消息。 python kafkaproducer.py 生产消息示例代码如下： from confluentkafka import Producer import setting conf setting.kafkasetting """初始化一个 Producer 对象""" p Producer({'bootstrap.servers': conf['bootstrapservers']}) def deliveryreport(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush(). """ if err is not None: print('Message delivery failed: {}'.format(err)) else: print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition())) """异步发送消息""" p.produce(conf['topicname'], "Hello".encode('utf8'), callbackdeliveryreport) p.poll(0) """在程序结束时，调用flush""" p.flush() 消费消息 发送以下命令消费消息。 python kafkaconsumer.py 消费消息示例代码如下： from confluentkafka import Consumer, KafkaError import setting conf setting.kafkasetting c Consumer({ 'bootstrap.servers': conf['bootstrapservers'], 'group.id': conf['groupname'], 'auto.offset.reset': 'latest' }) c.subscribe([conf['topicname']]) while True: msg c.poll(1.0) if msg is None: continue if msg.error(): if msg.error().code() KafkaError.PARTITIONEOF: continue else: print("Consumer error: {}".format(msg.error())) continue print('Received message: {}'.format(msg.value().decode('utf8'))) c.close()

非SSL认证方式 import sys from kombu import Connection, Exchange, Queue, Consumer def Main(argv): rabbitmqurl 'amqp://USERNAME:PASSWORD@XXX:xxx/' conn Connection(rabbitmqurl, loginmethod'PLAIN') exchange Exchange("exampleexchange", type"direct") queue Queue(name"examplequeue", exchangeexchange, routingkey"testkey") def processmessage(body, message): print("receive message: %s" % format(body)) message.ack() with Consumer(conn, queuesqueue, callbacks[processmessage], accept["text/plain"]): conn.drainevents(timeout2) if name 'main': Main(sys.argv)

SSL消费消息 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.nio.charset.StandardCharsets; import java.security.; import java.security.cert.CertificateException; import java.util.concurrent.TimeoutException; public class RabbitmqConsumerSsl { //队列名称 private final static String QUEUENAME "helloMQ"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接工厂 ConnectionFactory factory new ConnectionFactory(); //设置主机ip factory.setHost("127.0.0.1"); // 设置amqp的ssl端口号 factory.setPort(5671); String ksFile "/sslpath/ssl/clientrabbitmqkey.p12"; String tksFile "/sslpath/ssl/truststore"; SSLContext c null; try { char[] keyPassphrase "YOUR PASSPHRASE".toCharArray(); KeyStore ks KeyStore.getInstance("PKCS12"); ks.load(new FileInputStream(ksFile), keyPassphrase); KeyManagerFactory kmf KeyManagerFactory.getInstance("SunX509"); kmf.init(ks, keyPassphrase); char[] trustPassphrase "YOUR PASSPHRASE".toCharArray(); KeyStore tks KeyStore.getInstance("JKS"); tks.load(new FileInputStream(tksFile), trustPassphrase); TrustManagerFactory tmf TrustManagerFactory.getInstance("SunX509"); tmf.init(tks); c SSLContext.getInstance("tlsv1.2"); c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); } catch (KeyStoreException e) { throw new RuntimeException(e); } catch (NoSuchAlgorithmException e) { throw new RuntimeException(e); } catch (CertificateException e) { throw new RuntimeException(e); } catch (UnrecoverableKeyException e) { throw new RuntimeException(e); } catch (KeyManagementException e) { throw new RuntimeException(e); } factory.setSaslConfig(DefaultSaslConfig.EXTERNAL); factory.useSslProtocol(c); //设置Vhost，需要在控制台先创建 factory.setVirtualHost("vhost"); //基于网络环境合理设置超时时间 factory.setConnectionTimeout(30 1000); factory.setHandshakeTimeout(30 1000); factory.setShutdownTimeout(0); Connection connection factory.newConnection(); Channel channel connection.createChannel(); //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。 channel.queueDeclare(QUEUENAME, false, false, false, null); System.out.println(" [] Waiting for messages. To exit press CTRL+C"); Consumer consumer new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message new String(body, StandardCharsets.UTF8); System.out.println("Received message: '" + message + "'"); } }; channel.basicConsume(QUEUENAME, true, consumer); } }

本文介绍Redis实例支持SSL加密传输 Redis 6.0基础版实例SSL默认关闭，如需开启SSL加密传输，请参考SSL设置。 在开启SSL加密传输时，会为每个实例生成唯一的服务证书。客户端可以使用从服务控制台上下载的CA根证书，并在连接实例时提供该证书，对实例服务端进行认证并达到加密传输的目的。 Redis 2.8/4.0/5.0不支持SSL加密传输，仅支持明文传输。

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

本节介绍Kafka业务过载最佳实践 方案概述 Kafka业务过载，一般表现为CPU使用率高、磁盘写满的现象。 当CPU使用率过高时，系统的运行速度会降低，并有加速硬件损坏的风险。 当磁盘写满时，相应磁盘上的Kafka日志目录会出现offline问题。此时，该磁盘上的分区副本不可读写，降低了分区的可用性和容错能力。同时由于Leader分区迁移到其他节点，会增加其他节点的负载。 CPU使用率高的原因 数据操作相关线程数（num.io.threads、num.network.threads、num.replica.fetchers）过多，导致CPU繁忙。 分区设置不合理，所有的生产和消费都集中在某个节点上，导致CPU利用率高。 磁盘写满的原因 业务数据增长较快，已有的磁盘空间不能满足业务数据需要。 节点内磁盘使用率不均衡，生产的消息集中在某个分区，导致分区所在的磁盘写满。 Topic的数据老化时间设置过大，保存了过多的历史数据，容易导致磁盘写满。 实施步骤 CPU使用率高的处理措施： 优化线程参数num.io.threads、num.network.threads和num.replica.fetchers的配置。 num.io.threads和num.network.threads建议配置为磁盘个数的倍数，但不能超过CPU核数。 num.replica.fetchers建议配置不超过5。 合理设置Topic的分区，分区一般设置为节点个数的倍数。 生产消息时，给消息Key加随机后缀，使消息均衡分布到不同分区上。 磁盘写满的处理措施： 扩容磁盘，使磁盘具备更大的存储空间。 迁移分区，将已写满的磁盘中的分区迁移到本节点的其他磁盘中。 合理设置Topic的数据老化时间，减少历史数据的容量大小。 在CPU资源情况可控的情况下，使用压缩算法对数据进行压缩。 常用的压缩算法包括：ZIP，GZIP，SNAPPY，LZ4等。选择压缩算法时，需考虑数据的压缩率和压缩耗时。通常压缩率越高的算法，压缩耗时也越高。对于性能要求高的系统，可以选择压缩速度快的算法，比如LZ4；对于需要更高压缩比的系统，可以选择压缩率高的算法，比如GZIP。 可以在生产者端配置“compression.type”参数来启用指定类型的压缩算法。 Properties props new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 开启GZIP压缩 props.put("compression.type", "gzip"); Producer producer new KafkaProducer<>(props);

本文介绍使用RedisShake工具迁移自建Redis Cluster集群 RedisShake是一个开源的Redis迁移工具，可以用于在线迁移和离线迁移（通过备份文件导入）。当你需要迁移部署在其他云厂商上的 Redis集群数据时，如果无法进行在线迁移，你可以选择离线迁移的方式。 在线迁移 在线迁移主要适用于自建Redis Cluster集群迁移到DCS Redis的场景，且两端集群实例能够网络连通，或者有一台中转服务器能够连通两端集群实例。 在线迁移有多种模式，如果SYNC、PSYNC命令未被禁用，建议采用syncreader模式，否则可使用scanreader模式，具体见RedisShake官方社区。 1、在DCS控制台创建Redis实例。 注意 新创建的Redis实例容量不能小于源端Redis实例的实际使用容量。 2、准备一台云服务器，并安装RedisShake。 RedisShake需既能访问源端缓存实例，也能访问目标端DCS 缓存，需要绑定弹性公网IP 3、获取源集群和目标集群的ip地址。 如果源实例或者目标实例是proxy模式架构，则获取proxy ip即可。 4、编辑RedisShake配置文件。 编辑redisshake工具配置文件shake.toml，补充迁移双方信息，及迁移模式。 [syncreader] cluster false set to true if source is a redis cluster address "ip:port" when cluster is true, set address to one of the cluster node username "" keep empty if not using ACL password "" keep empty if no authentication is required tls false syncrdb true set to false if you don't want to sync rdb syncaof true

本文介绍Redis集群版客户端常见错误码 50010001：路由队列堵塞。 50010002：添加路由队列异常。 50010003： 路由处理异常。 50010004：注册客户端读失败。 50010005：超过允许的最大客户端连接数。 50010006：注册目标路由读事件异常。 50010007：单个客户端连接请求数大于规定值。 50010008：协议异常，一般为客户端发送了不支持的命令或者断开连接。 50010009：不完整的报文。 50010010：读取redis超时。表示已经发送给redis，redis处理超时；检查请求报文是否过大，如果请求报文过大，建议拆小。 50010011：网络io异常。 50010012：目标服务器繁忙,稍后再试。 50010013：获取目标路由主机失败，检查redis机器是否io繁忙。 50010014：路由主机配置端口不合法。 50010015：连接目标主机失败。 50010016：目标主机网络繁忙。 50010017：路由发送给目标主机时网络错误。 50010018：读取解析客户端请求失败。 50010019：注册redis服务端返回失败。 50010020：redis read io exception。 50010021：接入机内部异常。 50010022：内存池不够。 50010023：处理客户端报文异常。 50010024：route 程序bug。 50010025：没有该分组权限。 50010026：实例密码错误。 50010027：集群版鉴权实例名错误。 50010028：客户端IP不在IP白名单内。 50010029： 用户时间失效。 50010030：解析报文失败。 50010031：鉴权信息不完整。 50010032：尚未通过鉴权，未获取到鉴权通过信息。 50010033：尚未通过鉴权，未获取到鉴权分组信息。 50010034：尚未选择分组，未获取到选择分组信息。 50010036：dborder越界。 50010037：获取接入机对应用户名失败。 50010038：订单服务冻结，获取接入机服务失败。 50010039：该分组不存在。 50010040：操作的集合个数不能超过10个。 50010041：每个集合的大小不能超过1000个。 50010042：接入机改造命令限制30k。 50010043：该账户禁止写数据。 50010044：读取redis返回报文串包。 50010047：订阅连接超时错误码。 50010048：接入机不支持的命令。 50010049：接入机read only错误。 50010050：redis内存满。

本文介绍执行命令时出现20010004错误码原因及解决方法 20010004错误码的含义是执行命令超时 可能的原因： （1）发起执行耗时的操作，比如删大对象，批量获取数据、存取大报文数据等，此时在管理平台监控界面查看超时操作类型、分组号、ip等信息、以及报文大小是否超过32K； （2）系统过载繁忙，通过监控界面看到的请求量巨大，伴随少量超时错误率返回，一般小于万分之一； （3）应用程序出现full gc。 解决方法: （1）优化应用程序逻辑，增加客户端超时时间，如请求报文过大，建议拆小； （2）根据监控系统错误码来自接入层还是存储层，对瓶颈节点进行系统扩容。

可以在进程内直接增加（需要保证每个实例对应一个线程，否则没有太大意义），也可以部署多个消费实例进程；需要注意的是，实例个数超过分区数量后就不再能提高速度，将会有消费实例不工作。

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

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