Redis从入门到实战 - 高级篇（下）

一、Redis键值设计

1. 优雅的key结构

Redis的Key虽然可以自定义，但最好遵循下面几个最佳实践约定：

遵循基本格式：[业务名称]:[数据名]:[id]
长度不超过44字节
不包含特殊字符

例如：我们的登录业务，保存用户信息，其key是这样的：

优点：

可读性强
避免key冲突
方便管理
更节省内存：key是string类型，底层编码包含int、embstr和raw三种。embstr在小于44字节使用，采用连续内存空间，内存占用更小

2. 拒绝BigKey

BigKey通常以Key的大小和Key中成员的数量来综合判定，例如：

Key本身的数据量过大：一个String类型的Key，它的值为5MB
Key中的成员数过多：一个ZSET类型的Key，它的成员数量为10,000个
Key中成员的数据量过大：一个Hash类型的Key，它的成员数量虽然只有1,000个，但这些成员的Value（值）总大小为100MB

推荐值：

单个key的value小于10KB
对于集合类型的key，建议元素数量小于1000

BigKey的危害

网络阻塞

对BigKey执行读请求时，少量的QPS就可能导致带宽使用率被占满，导致Redis实例，乃至所在物理机变慢

数据倾斜

BigKey所在的Redis实例内存使用率远超其他实例，无法使数据分片的内存资源达到均衡

Redis阻塞

对元素较多的hash、list、zset等做运算会耗时较久，使主线程被阻塞

CPU压力

对BigKey的数据序列化和反序列化会导致CPU的使用率飙升，影响Redis实例和本机其它应用

如何发现BigKey

（1）redis-cli --bigkeys

利用redis-cli提供的--bigkeys参数，可以遍历分析所有key，并返回Key的整体统计信息与每个数据的Top1的big key

（2）scan扫描

自己编程，利用scan扫描Redis中的所有key，利用strlen、hlen等命令判断key的长度（此处不建议使用MEMORY USAGE）

打开之前的jedis-demo项目，在单元测试类中添加一个方法 textScan() ：

package com.heima;

import com.heima.JedisUtil.JedisConnectionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.ScanResult;

import java.util.List;
import java.util.Map;

public class JedisTest {
    private Jedis jedis;

    final static int STR_MAX_LEN = 10 * 1024;
    final static int HASH_MAX_LEN = 500;

    @Test
    void textScan() {
        int maxLen = 0;
        long len = 0;

        String cursor = "0";
        do {
            // 扫描并获取一部分key
            ScanResult<String> result = jedis.scan(cursor);
            // 记录cursor
            cursor = 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;
                    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"));
    }

    @BeforeEach
    void setUp() {
       jedis = JedisConnectionFactory.getJedis();
       jedis.auth("leadnews");
       jedis.select(0);
    }

    @Test
    void testString() {
        // 1. 存入数据
        String result = jedis.set("name", "虎哥");
        System.out.println("result = " + result);
        // 2. 获取数据
        String name = jedis.get("name");
        System.out.println("name = " + name);
    }

    @Test
    void testHash() {
        // 插入hash数据
        jedis.hset("user:1", "name", "Jack");
        jedis.hset("user:1", "age", "20");

        // 获取
        Map<String, String> map = jedis.hgetAll("user:1");
        System.out.println(map);
    }

    @AfterEach
    void tearDown() {
        // 关闭连接
        if (jedis != null) {
            jedis.close();
        }
    }
}

（3）第三方工具

利用第三方工具，如Redis-Rdb-Tools分析RDB快照文件，全面分析内存使用情况

Github地址：https://github.com/sripathikrishnan/redis-rdb-tools

（4）网络监控

自定义工具，监控进出Redis的网络数据，超出预警值时自动告警

如何删除BigKey

BigKey内存占用较多，即便删除这样的key也需要耗费很长时间，导致Redis主线程阻塞，引发一系列问题。

redis 3.0及以下版本

如果是集合类型，则遍历BigKey的元素，先逐个删除子元素，最后删除BigKey

Redis 4.0以后

Redis在4.0以后提供了异步删除的命令：unlink

3. 恰当的数据类型

例1：比如存储一个User对象，我们有三种存储方式：

方式一：json字符串

优点：实现简单粗暴
缺点：数据耦合，不够灵活

方式二：字段打散

优点：可以灵活访问对象任意字段
缺点：占用空间大、没办法做到统一控制

方式三：hash

优点：底层使用ziplist，空间占用小，可以灵活访问对象的任意字段
缺点：代码相对复杂

例2：假如有hash类型的key，其中有100万对field和value，field是自增id，这个key存在什么问题？如何优化？

存在的问题：

①hash的entry数量超过500时，会使用哈希表而不是ZipList，内存占用较多

②可以通过hash-max-ziplist-entries配置entry上限。但是如果entry过多就会导致BigKey问题

例2：假如有hash类型的key，其中有100万对field和value，field是自增id，这个key存在什么问题？如何优化？

方案二：拆分为string类型：

存在的问题：

①string结构底层没有太多内存优化，内存占用较多

②想要批量获取这些数据比较麻烦

方案三：拆分为小的hash，将 id / 100作为key，将id % 100作为field，这样每100个元素作为一个Hash

单元测试：

package com.heima;

import com.heima.JedisUtil.JedisConnectionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.ScanResult;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class JedisTest {
    private Jedis jedis;

    // ... ...

    @BeforeEach
    void setUp() {
       jedis = JedisConnectionFactory.getJedis();
       jedis.auth("leadnews");
       jedis.select(0);
    }


    @AfterEach
    void tearDown() {
        // 关闭连接
        if (jedis != null) {
            jedis.close();
        }
    }

    @Test
    void testBigHash() {
        HashMap<String, String> map = new HashMap<>();
        for (int i = 1; i <= 100000; i++) {
            map.put("key_" + i, "value_" + i);
        }
        jedis.hmset("test:big:hash", map);
    }

    @Test
    void testBigString() {
        for (int i = 1; i <= 100000; i++) {
            jedis.set("test:str:key_" + i, "value_" + i);
        }
    }

    @Test
    void 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);
            }
        }
    }
}

总结：

Key的最佳实践：

固定格式：[业务名]:[数据名]:[id]
足够简短：不超过44字节
不包含特殊字符

Value的最佳实践：

合理的拆分数据，拒绝BigKey
选择合适的数据结构
Hash结构的entry数量不要超过1000
设置合理的超时时间

二、批处理优化

1. Pipeline

大量数据导入的方式

一堆粮食要搬运到仓库，有几种方法？

单个命令的执行流程

一次命令的响应时间 = 1次往返的网络传输耗时 + 1次Redis执行命令耗时

    @Test
    void testFor() {
        long b = System.currentTimeMillis();
        for (int i = 1; i <= 100000; i++) {
            jedis.set("test:key_" + i, "value_" + i);
        }
        long e = System.currentTimeMillis();
        System.out.println("time:" + (e - b));
    }

N条命令批量执行

N次命令的响应时间 = 1次往返的网络传输耗时 + N次Redis执行命令耗时

MSET

Redis提供了很多Mxxx这样的命令，可以实现批量插入数据，例如：

mset
hmset

利用mset批量插入10万条数据：

    @Test
    void 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:ket_" + i;
            arr[j + 1] = "value_" + i;
            if (j == 0) {
                jedis.mset(arr);
            }
        }
        long e = System.currentTimeMillis();
        System.out.println("time:" + (e - b));
    }

注意：不要在一次批处理中传输太多命令，否则单次命令占用带宽过多，会导致网络阻塞

Pipeline

MSET虽然可以批处理，但是却只能操作部分数据类型，因此如果有对复杂数据类型的批处理需求，建议使用Pipeline功能：

    @Test
    void testPipeline() {
        long b = System.currentTimeMillis();

        // 创建管道
        Pipeline pipeline = jedis.pipelined();

        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));
    }

总结：

批量处理的方案：

原生的M操作
Pipeline批处理

注意事项：

批处理时不建议一次携带太多命令
Pipeline的多个命令之间不具备原子性

2. 集群模式下的批处理

如MSET或Pipeline这样的批处理需要在一次请求中携带多条命令，二此时如果Redis是一个集群，那批处理命令的多个key必须落在一个插槽中，否则就会导致执行失败：

	串行命令	串行slot	并行slot	hash_tag
实现思路	for循环遍历，依次执行每个命令	在客户端计算每个key的slot，将slot一致分为一组，每组都利用Pipeline批处理。串行执行各组命令	在客户端计算每个key的slot，将slot一致分为一组，每组都利用Pipeline批处理。并行执行各组命令	将所有key设置相同的hash_tag，则所有key的slot一定相同
耗时	N次网络耗时 + N次命令耗时	m次网络耗时 + N次命令耗时 m = key的slot个数	1次网络耗时 + N次命令耗时	1次网络耗时 + N次命令耗时
优点	实现简单	耗时较短	耗时非常短	耗时非常短、实现简单
缺点	耗时非常久	实现稍复杂 slot越多，耗时越久	实现复杂	容易出现数据倾斜

（1）Jedis串行执行：jedis-demo项目

    @Test
    void testMSet() {
        // redis.clients.jedis.exceptions.JedisClusterOperationException: No way to dispatch this command to Redis Cluster because keys have different slots
        jedisCluster.mset("name", "Jack", "age", "21", "sex", "male");
    }

    @Test
    void 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())
                ));
        // 串行的去执行mset的逻辑
        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);
        }
    }

ClusterSlotHashUtil：

package com.heima.JedisUtil;

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;
    }
}

ByteUtil：

package com.heima.JedisUtil;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Some handy methods for dealing with {@code byte} arrays.
 *
 * @author Christoph Strobl
 * @author Mark Paluch
 * @since 1.7
 */
public final class ByteUtils {

	private ByteUtils() {}

	/**
	 * Concatenate the given {@code byte} arrays into one, with overlapping array elements included twice.
	 * <p />
	 * The order of elements in the original arrays is preserved.
	 *
	 * @param array1 the first array.
	 * @param array2 the second array.
	 * @return the new array.
	 */
	public static byte[] concat(byte[] array1, byte[] array2) {

		byte[] result = Arrays.copyOf(array1, array1.length + array2.length);
		System.arraycopy(array2, 0, result, array1.length, array2.length);

		return result;
	}

	/**
	 * Concatenate the given {@code byte} arrays into one, with overlapping array elements included twice. Returns a new,
	 * empty array if {@code arrays} was empty and returns the first array if {@code arrays} contains only a single array.
	 * <p />
	 * The order of elements in the original arrays is preserved.
	 *
	 * @param arrays the arrays.
	 * @return the new array.
	 */
	public static byte[] concatAll(byte[]... arrays) {

		if (arrays.length == 0) {
			return new byte[] {};
		}
		if (arrays.length == 1) {
			return arrays[0];
		}

		byte[] cur = concat(arrays[0], arrays[1]);
		for (int i = 2; i < arrays.length; i++) {
			cur = concat(cur, arrays[i]);
		}
		return cur;
	}

	/**
	 * Split {@code source} into partitioned arrays using delimiter {@code c}.
	 *
	 * @param source the source array.
	 * @param c delimiter.
	 * @return the partitioned arrays.
	 */
	public static byte[][] split(byte[] source, int c) {

		if (ObjectUtils.isEmpty(source)) {
			return new byte[][] {};
		}

		List<byte[]> bytes = new ArrayList<>();
		int offset = 0;
		for (int i = 0; i <= source.length; i++) {

			if (i == source.length) {

				bytes.add(Arrays.copyOfRange(source, offset, i));
				break;
			}

			if (source[i] == c) {
				bytes.add(Arrays.copyOfRange(source, offset, i));
				offset = i + 1;
			}
		}
		return bytes.toArray(new byte[bytes.size()][]);
	}

	/**
	 * Merge multiple {@code byte} arrays into one array
	 *
	 * @param firstArray must not be {@literal null}
	 * @param additionalArrays must not be {@literal null}
	 * @return
	 */
	public static byte[][] mergeArrays(byte[] firstArray, byte[]... additionalArrays) {

		Assert.notNull(firstArray, "first array must not be null");
		Assert.notNull(additionalArrays, "additional arrays must not be null");

		byte[][] result = new byte[additionalArrays.length + 1][];
		result[0] = firstArray;
		System.arraycopy(additionalArrays, 0, result, 1, additionalArrays.length);

		return result;
	}

	/**
	 * Extract a byte array from {@link ByteBuffer} without consuming it.
	 *
	 * @param byteBuffer must not be {@literal null}.
	 * @return
	 * @since 2.0
	 */
	public static byte[] getBytes(ByteBuffer byteBuffer) {

		Assert.notNull(byteBuffer, "ByteBuffer must not be null!");

		ByteBuffer duplicate = byteBuffer.duplicate();
		byte[] bytes = new byte[duplicate.remaining()];
		duplicate.get(bytes);
		return bytes;
	}

	/**
	 * Tests if the {@code haystack} starts with the given {@code prefix}.
	 *
	 * @param haystack the source to scan.
	 * @param prefix the prefix to find.
	 * @return {@literal true} if {@code haystack} at position {@code offset} starts with {@code prefix}.
	 * @since 1.8.10
	 * @see #startsWith(byte[], byte[], int)
	 */
	public static boolean startsWith(byte[] haystack, byte[] prefix) {
		return startsWith(haystack, prefix, 0);
	}

	/**
	 * Tests if the {@code haystack} beginning at the specified {@code offset} starts with the given {@code prefix}.
	 *
	 * @param haystack the source to scan.
	 * @param prefix the prefix to find.
	 * @param offset the offset to start at.
	 * @return {@literal true} if {@code haystack} at position {@code offset} starts with {@code prefix}.
	 * @since 1.8.10
	 */
	public static boolean startsWith(byte[] haystack, byte[] prefix, int offset) {

		int to = offset;
		int prefixOffset = 0;
		int prefixLength = prefix.length;

		if ((offset < 0) || (offset > haystack.length - prefixLength)) {
			return false;
		}

		while (--prefixLength >= 0) {
			if (haystack[to++] != prefix[prefixOffset++]) {
				return false;
			}
		}

		return true;
	}

	/**
	 * Searches the specified array of bytes for the specified value. Returns the index of the first matching value in the
	 * {@code haystack}s natural order or {@code -1} of {@code needle} could not be found.
	 *
	 * @param haystack the source to scan.
	 * @param needle the value to scan for.
	 * @return index of first appearance, or -1 if not found.
	 * @since 1.8.10
	 */
	public static int indexOf(byte[] haystack, byte needle) {

		for (int i = 0; i < haystack.length; i++) {
			if (haystack[i] == needle) {
				return i;
			}
		}

		return -1;
	}

	/**
	 * Convert a {@link String} into a {@link ByteBuffer} using {@link StandardCharsets#UTF_8}.
	 *
	 * @param theString must not be {@literal null}.
	 * @return
	 * @since 2.1
	 */
	public static ByteBuffer getByteBuffer(String theString) {
		return getByteBuffer(theString, StandardCharsets.UTF_8);
	}

	/**
	 * Convert a {@link String} into a {@link ByteBuffer} using the given {@link Charset}.
	 *
	 * @param theString must not be {@literal null}.
	 * @param charset must not be {@literal null}.
	 * @return
	 * @since 2.1
	 */
	public static ByteBuffer getByteBuffer(String theString, Charset charset) {

		Assert.notNull(theString, "The String must not be null!");
		Assert.notNull(charset, "The String must not be null!");

		return charset.encode(theString);
	}

	/**
	 * Extract/Transfer bytes from the given {@link ByteBuffer} into an array by duplicating the buffer and fetching its
	 * content.
	 *
	 * @param buffer must not be {@literal null}.
	 * @return the extracted bytes.
	 * @since 2.1
	 */
	public static byte[] extractBytes(ByteBuffer buffer) {

		ByteBuffer duplicate = buffer.duplicate();
		byte[] bytes = new byte[duplicate.remaining()];
		duplicate.get(bytes);

		return bytes;
	}
}

ObjectUtils：

package com.heima.JedisUtil;

import com.sun.istack.internal.Nullable;

import java.lang.reflect.Array;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;

public class ObjectUtils {
    public static boolean isEmpty(@Nullable Object obj) {
        if (obj == null) {
            return true;
        }

        if (obj instanceof Optional) {
            return !((Optional<?>) obj).isPresent();
        }
        if (obj instanceof CharSequence) {
            return ((CharSequence) obj).length() == 0;
        }
        if (obj.getClass().isArray()) {
            return Array.getLength(obj) == 0;
        }
        if (obj instanceof Collection) {
            return ((Collection<?>) obj).isEmpty();
        }
        if (obj instanceof Map) {
            return ((Map<?, ?>) obj).isEmpty();
        }

        // else
        return false;
    }
}

Assert：

package com.heima.JedisUtil;

public class Assert {
    public static void notNull(Object obj, String msg){
        if (obj == null) {
            throw new RuntimeException(msg);
        }
    }
    public static void hasText(String str, String msg){
        if (str == null) {
            throw new RuntimeException(msg);
        }
        if (str.trim().isEmpty()) {
            throw new RuntimeException(msg);
        }
    }
}

（2）并行slot方案：Spring的StringRedisTemplate（redis-demo项目）

    @Test
    void testMSetInCluster() {
        Map<String, String> map = new HashMap<>(3);
        map.put("name", "Rose");
        map.put("age", "21");
        map.put("sex", "Female");
        stringRedisTemplate.opsForValue().multiSet(map);


        List<String> strings = stringRedisTemplate.opsForValue().multiGet(Arrays.asList("name", "age", "sex"));
        strings.forEach(System.out::println);
    }

集群配置：appIication.yml

spring:
  application:
    name: redis-demo
  redis:
#    host: 192.168.200.130
#    port: 6379
#    password: leadnews
    lettuce:
      pool:
        max-active: 8
        max-idle: 8
        min-idle: 0
        max-wait: 1000
    cluster:
      nodes:
        - 192.168.200.130:7001
        - 192.168.200.130:7003
        - 192.168.200.130:7003
        - 192.168.200.130:8001
        - 192.168.200.130:8002
        - 192.168.200.130:8003

三、服务端优化

1. 持久化配置

Redis的持久化虽然可以保证数据安全，但也会带来额外的开销，因此持久化请遵循下列建议：

①用来做缓存的Redis实例尽量不要开启持久化功能

②建议关闭RDB持久化功能，使用AOF持久化

③利用脚本定期在slave节点做RDB，实现数据备份

④设置合理的rewrite阈值，避免频繁的bgrewrite

⑤配置no-appendfsync-on-rewrite=yes，禁止在rewrite期间做aof，避免因AOF引起的阻塞

部署有关的建议：

①Redis实例的物理机要预留足够内存，应对fork和rewrite

②单个Redis实例内存上限不要太大，例如4G或8G。可以加快fork的速度、减少主从同步、数据迁移压力

③不要与CPU密集型应用部署在一起

④不要与高硬盘负载应用一起部署。例如：数据库、消息队列

2. 慢查询

慢查询：在Redis执行时耗时超过某个阈值的命令，称为慢查询

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

slowlog-log-slower-than：慢查询阈值，单位是微秒。默认是10000，建议1000

慢查询会被放入慢查询日志中，日志的长度有上限，可以通过配置指定：

slowlog-max-len：慢查询日志（本质是一个队列）的长度。默认是128，建议1000

修改这两个配置可以使用：config set命令：（临时生效）

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

3. 命令及安全配置

Redis会绑定在0.0.0.0:6379，这样会将Redis服务暴露在公网上，而Redis如果没有做身份认证，会出现严重的安全漏洞。

漏洞重现方式：Redis未授权访问配合SSH key文件利用分析-腾讯云开发者社区-腾讯云

漏洞出现的核心原因有以下几点：

Redis未设置密码
利用了Redis的config set命令动态修改Redis配置
使用了Root账号权限启动Redis

为了避免这样的漏洞，这里给出一些建议：

Redis一定要设置密码
禁止线上使用下面的命令：keys、flushall、flushdb、config set等命令。可以利用rename-command禁用

bind：限制网关，禁止外网网卡访问
开启防火墙
不要使用Root账户启动Redis
尽量不使用默认的端口

4. 内存配置

当Redis内存不足时，可能导致Key频繁被删除、响应时间变长、QPS不稳定等问题。当内存使用率达到90%以上时就需要我们警惕，并快速定位到内存占用的原因。

内存占用	说明
数据内存	是Redis最主要的部分，存储Redis的键值信息。主要问题是BigKey问题、内存碎片问题
进程内存	Redis主进程本身运行肯定需要占用内存，如代码、常量池等等；这部分内存大约几兆，在大多数生产环境中与Redis数据占用的内存相比可以忽略
缓冲区内存	一般包括客户端缓冲区、AOF缓冲区、复制缓冲区等。客户端缓冲区又包括输入缓冲区和输出缓冲区两种。这部分内存占用波动较大，不当使用BigKey，可能导致内存溢出。

数据内存的问题

Redis提供了一些命令，可以查看Redis目前的内存分配状态：

info memory
memory xxx

内存缓冲区配置

内存缓冲区常见的有三种：

复制缓冲区：主从复制的repl_backlog_buf，如果太小可能导致频繁的全量复制，影响性能。通过repl-backlog-size来配置，默认1mb
AOF缓冲区：AOF刷盘之前的缓存区域，AOF执行rewrite的缓冲区。无法设置容量上限
客户端缓冲区：分为输入缓存区和输出缓冲区，输入缓冲区最大1G且不能设置。输出缓冲区可以设置

以上复制缓冲区和AOF缓冲区不会有问题，最关键就是客户端缓冲区的问题

客户端缓冲区：指的就是我们发送命令时，客户端用来缓存命令的一个缓冲区，也就是我们向redis输入数据的输入端缓冲区和redis向客户端返回数据的响应缓存区，输入缓冲区最大1G且不能设置，所以这一块我们根本不用担心，如果超过了这个空间，redis会直接断开，因为本来此时此刻就代表着redis处理不过来了，我们需要担心的就是输出端缓冲区