Redis高级篇: Redis最佳实践

高级篇: Redis最佳实践"/>

Redis高级篇: Redis最佳实践

学习内容

1. Redis键值对的设计键值对设计的技巧和使用规范
2. 批处理优化在redis中如何高效处理大量命令
3. 服务端配置如何配置内存的大小以及慢查询如何处理
4. Redis集群配置对比主从模式有哪些优缺点,存在什么问题,帮助我们在选择redis架构时做出最佳选择

Redis键值设计

总结:

优雅的key结构

注意点:
如果key是数字,redis底层会直接使用int进行编码;
如果key是字符号,且长度<44byte,redis底层会使用embstr分配连续的空间进行存储
如果key是字符串,且长度>44byte,redis底层会使用raw编码存储,采用指针串联不连续的空间,存储的空间大,且容器产生内存碎片查看编码方式的命令:
type key # 查看key存储数据的类型
object encoding key # 查看key的编码方式

拒绝bigkey

redis提供查看bigkey的命令:memory usage key # 衡量一个key占用的内存大小`不太推荐使用,占用CPU`,占用内存包含三部分: 1.key 2.value 3.存储值的数据结构如何查看key是否是bigkey?
- 见下

bigkey危害解读:

命令解读: 
redis-cli --bigkeys  `有局限性(参考)`
能够统计每一种数据结构中保存的占用内存最大的一个key;
但是这个key不一定是bigkey,并且第二第三个也可能是bigkey第三方工具: github上下载,扫描rdb文件进行分析,时效性差网络监控: 如果使用的是第三方云服务,默认提供scan扫描`自己编程,判断`
不能使用key * ,会阻塞主线程/*** 定义string类型bigkey的标准*/final static int STR_MAX_LEN = 10 * 1024;/*** 定义hash类型bigkey的标准*/final static int HASH_MAX_LEN = 500;@Testvoid testScan() {int maxLen = 0;long len = 0;String cursor = "0";do {// 扫描并获取一部分keyScanResult<String> result = jedis.scan(cursor);// 记录cursorcursor = result.getCursor();List<String> list = result.getResult();if (list == null || list.isEmpty()) {break;}// 遍历for (String key : list) {// 判断key的类型String type = jedis.type(key);switch (type) {case "string":len = jedis.strlen(key);maxLen = STR_MAX_LEN;break;case "hash":len = jedis.hlen(key);maxLen = HASH_MAX_LEN;break;case "list":len = jedis.llen(key);maxLen = HASH_MAX_LEN;break;case "set":len = jedis.scard(key);maxLen = HASH_MAX_LEN;break;case "zset":len = jedis.zcard(key);maxLen = HASH_MAX_LEN;break;default:break;}if (len >= maxLen) {System.out.printf("Found big key : %s, type: %s, length or size: %d %n", key, type, len);}}} while (!cursor.equals("0"));}

解读:
4.0以后提供了unlink,启动异步线程删除元素,不会阻塞主线程
4.0以前想要删除元素,为了不会阻塞主线程,就需要向通过异步命令scan获取所有member,然后再删除

恰当value的数据类型

没有最好的数据结构,只有最合适的数据结构
我们要根据应用场景不同,以及内存占用率网络带宽等各方面因素来考量value的数据结构,达到内存的最大化利用

解读:一般: 因为我们用json比较频繁,并且序列化工具完善,所以我们习惯用stirng类型进行存储,但是数据耦合,更新数据不灵活最佳: hash结合底层使用了ziplist压缩链表,空间占用小,可以灵活修改任意字段的内容,但是需要将java对象转化为map存储,还需要考虑数据类型的转换,实现起来比较麻烦注意点: hash类型底层如果entry>500,会使用hash表而不是ziplist存储,内存占用较多不好: 字段打散可以灵活修改字段内容,但是由于元数据会占用大量存储空间

案例分析:

问题分析:
1. hash类型元素>500,属于bigkey,严重占用内存和网络带宽
2. hash类型entry>500,底层使用hash表存储,而不是ziplist,占用大量内存解决方案:
1. 调整redishash类型hash表存储上限`不好`  config set hash-max-ziplist-entries 1000 
问题: bigkey2. 调整hash为string类型存储,分为100w分key分别存储
问题: string类型底层没有进行优化,内存占用较多;想要批量获取数据比较麻烦;3. 大hash拆分成小hash,将id/100作为key,将id%100作为field,这样每100个元素为一个hash
注意点: 这里的100要根据业务量进行恒定计算,保证一个hash存储元素<500
![在这里插入图片描述](.png)@Testvoid testSetBigKey() {Map<String, String> map = new HashMap<>();for (int i = 1; i <= 650; i++) {map.put("hello_" + i, "world!");}jedis.hmset("m2", map);}@Testvoid testBigHash() {Map<String, String> map = new HashMap<>();for (int i = 1; i <= 100000; i++) {map.put("key_" + i, "value_" + i);}jedis.hmset("test:big:hash", map);}@Testvoid testBigString() {for (int i = 1; i <= 100000; i++) {jedis.set("test:str:key_" + i, "value_" + i);}}@Testvoid testSmallHash() {int hashSize = 100;Map<String, String> map = new HashMap<>(hashSize);for (int i = 1; i <= 100000; i++) {int k = (i - 1) / hashSize;int v = i % hashSize;map.put("key_" + v, "value_" + v);if (v == 0) {jedis.hmset("test:small:hash_" + k, map);}}}

Redis批处理优化

如何优雅的完成海量数据的处理?

案例引入:
查询附近商户,一股脑把商户的地理位置信息全部查出来全部写入到redis中的geo数据结构中,当数据量非常多,如何优化导入操作,如果处理不恰当会造成非常大的危害?

命令处理方式

总结:

解读: 
1. 光纤的传输速度是有限制的,所以即使是本机访问同局域网的机器耗时也在毫秒级别;如果串行执行发送网络请求,执行redis指令.时间=10000X5ms,时间开销是巨大的
2. redis本身执行指令是非常快的,与网络请求速度不成正比,所以我们为了平衡二者的消耗时间,我们考虑让redis一次处理大量的数据,来平衡二者的时间注意点:
毛驴虽然说可以批量处理大量数据,但是由于网络带宽的限制,一次能够发送的网络数据包大小是有限的,为了防止网络出现拥堵,我们应该在能够承受的范围内分批次发送处理命令测试代码:@Testvoid testFor() {for (int i = 1; i <= 100000; i++) {jedis.set("test:key_" + i, "value_" + i);}}@Testvoid testMxx() {String[] arr = new String[2000];int j;long b = System.currentTimeMillis();for (int i = 1; i <= 100000; i++) {j = (i % 1000) << 1;arr[j] = "test:key_" + i;arr[j + 1] = "value_" + i;if (j == 0) {jedis.mset(arr);}}long e = System.currentTimeMillis();System.out.println("time: " + (e - b));}

单个命令的执行流程?

N条命令的依次执行

Pipeline批处理

什么是Pipeline: 大数据量的导入

redis原生的批处理命令

命令: mset hmset sadd等,具有原子性
缺点: 集合类型只能处理单key的多member

pipeline实现批处理命令

好处: 
1. 能够实现复杂数据类型,多key的批处理
2. 能够操作任意类型,通过sync进行批量命令发送缺点:
1. 执行命令具有先后顺序,不具备原子性代码案例:
@Testvoid testPipeline() {// 创建管道Pipeline pipeline = jedis.pipelined();long b = System.currentTimeMillis();for (int i = 1; i <= 100000; i++) {// 放入命令到管道pipeline.set("test:key_" + i, "value_" + i);if (i % 1000 == 0) {// 每放入1000条命令，批量执行pipeline.sync();}}long e = System.currentTimeMillis();System.out.println("time: " + (e - b));}

集群下的批处理

批处理是建立一个连接,然后一次发送上千条指令到redis进行执行,在集群模式下批处理指令必须由一个插槽完成,但是通过hash无法计算得到相同插槽,所以会报错
如果mset或者pipeline批处理需要在一次请求中携带多条命令,而如果此时redis是集群,那么批处理的多个key必须落在同一个插槽内,否则会导致执行失败

解决方案:

解析:串行: 放弃治疗,单命令发送(网络开销大)
串行slot: 计算每一个命令的slot,按照slot分组,然后分组串行执行(串行slot,slot分组)
并行slot`推荐`: 分组后,多线程并行发送分组的命令(并行slot,slot分组)
hash_tag: 所有key设置相同的hash_tag,那么这些数据计算后一定是在相同slot中,就不会出现错误,单节点保存数据过多,但是容易出现数据倾斜的问题

代码案例:

• 错误代码(批处理非相同slot报错) jedis默认没有解决集群批处理问题

public class JedisClusterTest {private JedisCluster jedisCluster;@BeforeEachvoid setUp() {// 配置连接池JedisPoolConfig poolConfig = new JedisPoolConfig();poolConfig.setMaxTotal(8);poolConfig.setMaxIdle(8);poolConfig.setMinIdle(0);poolConfig.setMaxWaitMillis(1000);HashSet<HostAndPort> nodes = new HashSet<>();nodes.add(new HostAndPort("192.168.150.101", 7001));nodes.add(new HostAndPort("192.168.150.101", 7002));nodes.add(new HostAndPort("192.168.150.101", 7003));nodes.add(new HostAndPort("192.168.150.101", 8001));nodes.add(new HostAndPort("192.168.150.101", 8002));nodes.add(new HostAndPort("192.168.150.101", 8003));jedisCluster = new JedisCluster(nodes, poolConfig);}@Testvoid testMSet() {jedisCluster.mset("name", "Jack", "age", "21", "sex", "male");}@Testvoid testMSet2() {Map<String, String> map = new HashMap<>(3);map.put("name", "Jack");map.put("age", "21");map.put("sex", "Male");Map<Integer, List<Map.Entry<String, String>>> result = map.entrySet().stream().collect(Collectors.groupingBy(entry -> ClusterSlotHashUtil.calculateSlot(entry.getKey())));for (List<Map.Entry<String, String>> list : result.values()) {String[] arr = new String[list.size() * 2];int j = 0;for (int i = 0; i < list.size(); i++) {j = i<<2;Map.Entry<String, String> e = list.get(0);arr[j] = e.getKey();arr[j + 1] = e.getValue();}jedisCluster.mset(arr);}}@AfterEachvoid tearDown() {if (jedisCluster != null) {jedisCluster.close();}}

jedis手动解决集群批处理问题代码案例:

package com.heima.jedis.util;import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;/*** @author Christoph Strobl* @since 1.7*/
public final class ClusterSlotHashUtil {private static final int SLOT_COUNT = 16384;private static final byte SUBKEY_START = '{';private static final byte SUBKEY_END = '}';private static final int[] LOOKUP_TABLE = { 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6, 0x70E7, 0x8108,0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273, 0x2252, 0x52B5, 0x4294, 0x72F7,0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF, 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6,0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528, 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611,0x0630, 0x76D7, 0x66F6, 0x5695, 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4,0x58E5, 0x6886, 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF, 0xEB9E, 0x9B79,0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60, 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC,0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5, 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F,0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59, 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E,0xE16F, 0x1080, 0x00A1, 0x30C2, 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D,0xD31C, 0xE37F, 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xA7DB,0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691, 0x16B0, 0x6657, 0x7676, 0x4615,0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A, 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0,0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F, 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37,0x7A16, 0x0AF1, 0x1AD0, 0x2AB3, 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26,0x6C07, 0x5C64, 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0 };private ClusterSlotHashUtil() {}/*** @param keys must not be {@literal null}.* @return* @since 2.0*/public static boolean isSameSlotForAllKeys(Collection<ByteBuffer> keys) {Assert.notNull(keys, "Keys must not be null!");if (keys.size() <= 1) {return true;}return isSameSlotForAllKeys((byte[][]) keys.stream() //.map(ByteBuffer::duplicate) //.map(ByteUtils::getBytes) //.toArray(byte[][]::new));}/*** @param keys must not be {@literal null}.* @return* @since 2.0*/public static boolean isSameSlotForAllKeys(ByteBuffer... keys) {Assert.notNull(keys, "Keys must not be null!");return isSameSlotForAllKeys(Arrays.asList(keys));}/*** @param keys must not be {@literal null}.* @return*/public static boolean isSameSlotForAllKeys(byte[]... keys) {Assert.notNull(keys, "Keys must not be null!");if (keys.length <= 1) {return true;}int slot = calculateSlot(keys[0]);for (int i = 1; i < keys.length; i++) {if (slot != calculateSlot(keys[i])) {return false;}}return true;}/*** Calculate the slot from the given key.** @param key must not be {@literal null} or empty.* @return*/public static int calculateSlot(String key) {Assert.hasText(key, "Key must not be null or empty!");return calculateSlot(key.getBytes());}/*** Calculate the slot from the given key.** @param key must not be {@literal null}.* @return*/public static int calculateSlot(byte[] key) {Assert.notNull(key, "Key must not be null!");byte[] finalKey = key;int start = indexOf(key, SUBKEY_START);if (start != -1) {int end = indexOf(key, start + 1, SUBKEY_END);if (end != -1 && end != start + 1) {finalKey = new byte[end - (start + 1)];System.arraycopy(key, start + 1, finalKey, 0, finalKey.length);}}return crc16(finalKey) % SLOT_COUNT;}private static int indexOf(byte[] haystack, byte needle) {return indexOf(haystack, 0, needle);}private static int indexOf(byte[] haystack, int start, byte needle) {for (int i = start; i < haystack.length; i++) {if (haystack[i] == needle) {return i;}}return -1;}private static int crc16(byte[] bytes) {int crc = 0x0000;for (byte b : bytes) {crc = ((crc << 8) ^ LOOKUP_TABLE[((crc >>> 8) ^ (b & 0xFF)) & 0xFF]);}return crc & 0xFFFF;}
}

• RedisTempalte默认计算插槽,并行slot,解决了集群批处理的问题

Redis服务端优化

持久化配置

TODO: 这里等待复习完毕Redis持久化策略再来看

慢查询优化

Redis在执行命令的时候超过某个阈值的命令,称为慢查询

学习目标:

1. 了解慢查询是什么?
2. 了解慢查询的危害?
3. 如何配置慢查询来记录慢查询日志?
4. 如何查询慢查询日志,找到慢查询指令并优化?

慢查询相关指令

慢查询配置指令

慢查询的阈值可以通过配置指定:

• showlog-log-slower-than: 慢查询阈值,单位是微秒;默认10000,建议1000ws
  慢查询会被放入到慢查询日志中,日志的长度有上限,可以通过配置指定
• showlog-max-len: 慢查询日志(本质是一个队列)的长度;默认128,建议1000
  

慢查询查询指令

• slowlog len: 查询慢查询日志长度
• slowlog get [n]: 读取n条慢查询日志
• slowlog reset: 清空慢查询列表
  

慢查询优化角度分析

耗时命令: 禁用耗时较长的命令/优化指令

求交集并集等耗时长bigkey: 选用合适的数据结构

命令及安全配置

安全是服务端应用最重要的问题

解读:
1. 可以通过修改配置,指定redis持久化文件的存储位置在.ssh目录下面,然后通过set key value保存公钥到服务器,这样就可以免密登录了`前提: 能够连接到redis` 

解读: 
1. 由于redis响应是微秒级别的,所以天生为黑客破解密码提供了遍历,所以redis设置密码一定要复杂
2. 通过rename-command禁用`危险命令`,命令重命名/置空禁用

内存配置这里没有看太懂

=143&vd_source=5481be76fd3d2afb98d3c7091aaaaca6
当redis内存不足的时候,可能导致key频繁被删除,响应时间变长,QPS不稳定等问题,当内存占用率达到90%以上就需要我们警惕,并快速定位到内存占用的问题

解读:
数据内存: 主要是我们在设置key-value的时候产生了bigkey问题/数据结构选择不恰当导致的,属于人为因素,需要开发人员手动去解决;对于内存碎片问题,可以通过重启redis服务器解决`前提是:保证redis服务可用`进程内存: redis服务启动本身占用的内存,不用太关注缓冲区内存: 数据读写会先放入缓冲区

redis内存查询相关指令:

redis内存缓冲区配置

集群最佳实践,集群还是主从?

总结:

集群还是主从:
1. 单体Redis(主从集群)已经能达到万级的QPS,并且具备很强的高可用特性,如果主从能满足业务需求的前提下,能不使用集群就尽量不要用Redis集群

集群使用不当存在的问题?

问题解读:

1. 集群完整性问题默认CP

   如果redis集群任意一个插槽出现问题,redis自动停止整个集群对外提供服务,可以通过修改配置关闭,允许某插槽出现问题,其他插槽还能继续对外提供服务,前提是数据计算出来的插槽落在健康节点上CP还是AP

2. 集群带宽问题

   集群之间心跳检测通过ping完成,每次ping需要携带集群插槽和集群状态信息,集群节点越多每次ping的网络数据包就越大,如果集群中有数以万个节点,容易拉满带宽
   

解读:
集群节点数量过多: 业务拆分,不同业务使用不同的redis集群

3. 数据倾斜

   产生了bigkey问题导致单节点存储的数据远超其他节点

4. 客户端性能问题

   使用客户端操作redis集群的时候要动态的计算插槽以及主从节点的判断来选择存储数据到哪个节点,需要耗时

5. 命令集群兼容性问题

   由于集群模式有些命令要求必须在一个插槽执行,所以会导致像批处理命令无法执行的问题,需要手动去解决实现麻烦,影响性能

6. lua和事务问题集群模式无法使用

   lua和事务中也是执行多条命令,一次请求必须保证所有命令执行在同一插槽内,所以在集群模式下无法使用lua和事务