redis数据结构

1 string（字符串）->opsForValue

介绍

一个 string 类型的键最大可以存储 512 MB 的数据

应用场景

1 缓存数据，提高访问速度和降低数据库压力。
2 计数器，利用 incr 和 decr 命令实现原子性的加减操作。
3 分布式锁，利用 setnx 命令实现互斥访问。
4 限流，利用 expire 命令实现时间窗口内的访问控制。
5 可以缓存json对象，这个比较常见和简单
6 session来保存用户信息
7 全局ID
适合频繁读操作

java使用

@Test
void stringTest() {
    //字符串的序列化器
    RedisSerializer redisSerializer=new StringRedisSerializer();
    redisTemplate.setKeySerializer(redisSerializer);
    //首先查一下缓存数据
    String stringTest= (String) redisTemplate.opsForValue().get("stringTest");
    System.out.println(stringTest);
    if (StringUtils.isBlank(stringTest)){
        String newStr="wo shi string type";
        //把查询的值放进缓存中
        redisTemplate.opsForValue().set("stringTest", newStr);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

源码api

    /**
    *设置 key 的值为 value
    *如果key不存在添加key 保存值为value
    *如果key存在则对value进行覆盖
    */
    void set(K key, V value);

    /**
     * 设置 key 的值为 value
     * 其它规则与 set(K key, V value)一样
     * @param key 不能为空
     * @param value 设置的值
     * @param timeout 设置过期的时间
     * @param unit 时间单位。不能为空
     * @see Redis Documentation: SETEX
     */
    void set(K key, V value, long timeout, TimeUnit unit);

    /**
     *如果key不存在，则设置key 的值为 value. 存在则不设置
     *设置成功返回true 失败返回false
     * @param key key不能为空
     * @param value 设置的值
     */
    Boolean setIfAbsent(K key, V value);

    /**
     * 把一个map的键值对添加到redis中，key-value 对应着 key value。如果key已经存在就覆盖，
     * @param map不能为null 为null抛出空指针异常 可以为空集合
     */
    void multiSet(Map<? extends K, ? extends V> map);

    /**
     * 把一个map的键值对添加到redis中，key-value 对应着 key value。 当且仅当map中的所有key都
     * 不存在的时候，添加成功返回 true，否则返回false.
     * @param map map不能为空 可以为empty
     */
    Boolean multiSetIfAbsent(Map<? extends K, ? extends V> map);

    /**
     * 根据 key 获取对应的value 如果key不存在则返回null
     * @param key 不能为null
     */
    V get(Object key);

    /**
     * 设置key的值为value 并返回旧值。 如果key不存在返回为null
     * @param key 不能为null
     */
    V getAndSet(K key, V value);

    /**
     * 根据提供的key集合按顺序获取对应的value值
     * @param 集合不能为null 可以为empty 集合
     */
    List<V> multiGet(Collection<K> keys);

    /**
     * 为key 的值加上 long delta. 原来的值必须是能转换成Integer类型的。否则会抛出异常。
     * @param key 不能为null
     * @param delta 需要增加的值
     */
    Long increment(K key, long delta);

    /**
     * 为key 的值加上 double delta. 原来的值必须是能转换成Integer类型的。否则会抛出异常。
     * 添加double后不能再加整数。已经无法在转换为Integer
     * @param key 不能为null
     * @param 增加的值
     */
    Double increment(K key, double delta);

    /**
     * 为 key的值末尾追加 value 如果key不存在就直接等于 set(K key, V value)
     *
     * @param key 不能为null
     * @param value 追加的值
     * @see Redis Documentation: APPEND
     */
    Integer append(K key, String value);

    /**
     * 获取key 值从 start位置开始到end位置结束。 等于String 的 subString 前后闭区间
     *0 -1 整个key的值
     *-4 -1 从尾部开始往前截长度为4
     * @param key 不能为null
     * @param start 起始位置
     * @param end   结束位置
     * @see Redis Documentation: GETRANGE
     */
    String get(K key, long start, long end);

    /**
     * 将value从指定的位置开始覆盖原有的值。如果指定的开始位置大于字符串长度，先补空格在追加。
     * 如果key不存在，则等于新增。长度大于0则先补空格 set("key10", "abc", 3) 得到结果为：
     * 3空格 +"abc"
     * @param key 不能为null
     * @param value 值
     * @param offset 开始的位置
     */
    void set(K key, V value, long offset);

    /**
     * 获取key的value的长度。key不存在返回0
     * @param key 不能为空
     */
    Long size(K key);

    /**
     * 设置key的值偏移量为offset的bit位上的值为0或者1.true:1 false:0
     *
     * @param key 不能为空
     * @param offset 偏移量
     * @param value true or false
     */
    Boolean setBit(K key, long offset, boolean value);

    /**
     * 获取key的值偏移量offset的bit位的值。 返回true or false
     *
     * @param key 不能为空
     * @param offset 偏移量
     * 可以通过redis的 JedisConverters 对布尔结果进行转换
     */
    Boolean getBit(K key, long offset);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

存储结构截图

2 hash（哈希）->opsForHash

介绍

Hash 是一个键值对（key - value）集合，其中 value 的形式如： value=[{field1，value1}，…{fieldN，valueN}]。Hash 特别适合用于存储对象

应用场景

1 Hash很适合缓存对象，比如将一个购物车信息存储在Hash里面，Redis存储java对象常用String，那为什么还要用hash来存储?
String 存储通常应用在频繁读操作，它存储格式为JSON,即将java对象转换为json，然后存入redis。
Hash存储场景应用在频繁写操作，即当对象的某个属性频繁修改时，不适用string+json数据结构。
适用hash，就可以针对单个属性单独修改。例如商品库存、价格、关注数等经常变动的数据。
2 Hash实现购物车

java使用

/**
     * map测试
     */
    @Test
    void mapTest() {
        //字符串的序列化器
        RedisSerializer redisSerializer=new StringRedisSerializer();
        redisTemplate.setKeySerializer(redisSerializer);
            Map<String ,String> map = new HashMap<>();
            map.put("1","22");
            map.put("11","22");
            map.put("11","22");
            //putAll 添加多个key-value    添加一个使用put(H key, HK hashKey, HV value)
            redisTemplate.opsForHash().put("mapTest","hashMapTest",map);
            Object o = redisTemplate.opsForHash().get("mapTest", "hashMapTest");
            System.out.println(o);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

源码api

    /**
     * 从散列中删除给定的多个元素
     * @param key 不能为null 散列的名称
     * @param hashKeys 需要删除的keys集合
     */
    Long delete(H key, Object... hashKeys);

    /**
     * 判断散列中是否存在某个key
     */
    Boolean hasKey(H key, Object hashKey);

    /**
     * 得到某个三散列中key的hash值
     */
    HV get(H key, Object hashKey);

    /**
     * 得到多个key的值。
     */
    List<HV> multiGet(H key, Collection<HK> hashKeys);

    /**
     *为散了中某个值加上 整型 delta
     */
    Long increment(H key, HK hashKey, long delta);

    /**
     * 为散了中某个值加上 double delta
     */
    Double increment(H key, HK hashKey, double delta);

    /**
     * 获取散列中所有的key集合
     */
    Set<HK> keys(H key);

    /**
     * 获取散列的大小
     */
    Long size(H key);

    /**
     * 为散列添加多个key-value键值对
     *
     * @param key must not be {@literal null}.
     * @param m must not be {@literal null}.
     */
    void putAll(H key, Map<? extends HK, ? extends HV> m);

    /**
     * 为散列添加或者覆盖一个 key-value键值对
     */
    void put(H key, HK hashKey, HV value);

    /**
     * 为散列添加一个key-value键值对。如果存在则不添加不覆盖。返回false
     */
    Boolean putIfAbsent(H key, HK hashKey, HV value);

    /**
     * 获取散列的value集合
     */
    List<HV> values(H key);

    /**
     * 获取散列的key-value键值对集合
     */
    Map<HK, HV> entries(H key);

    /**
     * 获取散列的游标。
     * 可以参考：http://blog.csdn.net/pengdandezhi/article/details/78909041
     */
    Cursor<Map.Entry<HK, HV>> scan(H key, ScanOptions options);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

存储结构截图

3 list（列表）->opsForList

介绍

List 列表是简单的字符串列表，按照插入顺序排序，可以从头部或尾部向 List 列表添加元素。
列表的最大长度为 2^32 - 1，也即每个列表支持超过 40 亿个元素。

应用场景

1 消息队列,消息队列在存取消息时，必须要满足三个需求，分别是消息保序、处理重复的消息和保证消息可靠性。
Redis 的 List 和 Stream 两种数据类型，就可以满足消息队列的这三个需求。我们先来了解下基于 List 的消息队列实现方法，后面在介绍 Stream 数据类型时候，在详细说说 Stream。
1>、如何满足消息保序需求？
List 本身就是按先进先出的顺序对数据进行存取的，所以，如果使用 List 作为消息队列保存消息的话，就已经能满足消息保序的需求了。
List 可以使用 LPUSH + RPOP （或者反过来，RPUSH+LPOP）命令实现消息队列。
2> 如何处理重复的消息？
消费者要实现重复消息的判断，需要 2 个方面的要求：

• 每个消息都有一个全局的 ID。
• 消费者要记录已经处理过的消息的 ID。当收到一条消息后，消费者程序就可以对比收到的消息 ID 和记录的已处理过的消息 ID，来判断当前收到的消息有没有经过处理。如果已经处理过，那么，消费者程序就不再进行处理了。

但是 List 并不会为每个消息生成 ID 号，所以我们需要自行为每个消息生成一个全局唯一ID，生成之后，我们在用 LPUSH 命令把消息插入 List 时，需要在消息中包含这个全局唯一 ID。

3>如何保证消息可靠性？
当消费者程序从 List 中读取一条消息后，List 就不会再留存这条消息了。所以，如果消费者程序在处理消息的过程出现了故障或宕机，就会导致消息没有处理完成，那么，消费者程序再次启动后，就没法再次从 List 中读取消息了。
为了留存消息，List 类型提供了 BRPOPLPUSH 命令，这个命令的作用是让消费者程序从一个 List 中读取消息，同时，Redis 会把这个消息再插入到另一个 List（可以叫作备份 List）留存。
这样一来，如果消费者程序读了消息但没能正常处理，等它重启后，就可以从备份 List 中重新读取消息并进行处理了。

• 消息保序：使用 LPUSH + RPOP；
• 阻塞读取：使用 BRPOP；
• 重复消息处理：生产者自行实现全局唯一 ID；
• 消息的可靠性：使用 BRPOPLPUSH

2 公众账号的关注列表，粉丝列表

3 Redis的分页功能，消息队列，发红包场景
List 不支持多个消费者消费同一条消息，因为一旦消费者拉取一条消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费。要实现一条消息可以被多个消费者消费，那么就要将多个消费者组成一个消费组，使得多个消费者可以消费同一条消息，但是 List 类型并不支持消费组的实现。
这就要说起 Redis 从 5.0 版本开始提供的 Stream 数据类型了，Stream 同样能够满足消息队列的三大需求，而且它还支持「消费组」形式的消息读取。

java使用

    /**
     * list测试
     */
    @Test
    void listTest() {
        //字符串的序列化器
        RedisSerializer redisSerializer=new StringRedisSerializer();
        redisTemplate.setKeySerializer(redisSerializer);
            List<String> list = new ArrayList<>();
            list.add("11");
            list.add("22");
            list.add("666");
            //从list头部插入value
            redisTemplate.opsForList().leftPush("listTest",list);
            //获取指定下标数据   使用range（0  -1）可以获取所有数据
            Object listTest1 = redisTemplate.opsForList().index("listTest", 0);
            System.out.println(listTest1);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

源码api

存储结构截图

4 set（集合）->opsForSet

介绍

Set 类型是一个无序并唯一的键值集合，它的存储顺序不会按照插入的先后顺序进行存储。 Set 类型除了支持集合内的增删改查，同时还支持多个集合取交集、并集、差集。底层用了整数集合或哈希表

应用场景

1 黑名单数据可以放在redis的set集合中。 可以sismember命令来判断在不在黑名单。
2 共同关注，常用在首页展示栏中，好友推荐、文章推荐、商品推荐
3 存储某活动中中奖的用户名 ，Set 类型因为有去重功能，可以保证同一个用户不会中奖两次。
4 点赞
Set 类型和 List 类型的区别如下：

• List 可以存储重复元素，Set 只能存储非重复元素；
• List 是按照元素的先后顺序存储元素的，而 Set 则是无序方式存储元素的。

java使用

 /**
     * set测试
     */
    @Test
    void setTest() {
        //字符串的序列化器
        RedisSerializer redisSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(redisSerializer);
        Set<String> set = new HashSet<>();
        set.add("11");
        set.add("22");
        //从list头部插入value
        redisTemplate.opsForSet().add("setTest", set);
        //查询所有
        Set<Object> setTest1 = redisTemplate.opsForSet().members("setTest");
        System.out.println(setTest1);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

源码api

    /**
     * 给集合key添加多个值，集合不存在创建后再添加
     *
     * @param key 不能为null
     * @param values
     * @return
     */
    Long add(K key, V... values);

    /**
     * 移除集合中多个value值
     * @param key 不能为null
     * @param values
     * @return
     */
    Long remove(K key, Object... values);

    /**
     * 随机删除集合中的一个值，并返回。
     *
     * @param key 不能为null
     * @return
     */
    V pop(K key);

    /**
     * 把源集合中的一个元素移动到目标集合。成功返回true.
     *
     * @param key 不能为null
     * @param value
     * @param destKey must not be {@literal null}.
     * @return
     */
    Boolean move(K key, V value, K destKey);

    /**
     * 返回结合的大小
     *
     * @param key 不能为null
     * @return
     * @see Redis Documentation: SCARD
     */
    Long size(K key);

    /**
     * 检查集合中是否包含某个元素
     *
     * @param key 不能为null
     * @param o
     * @return
     */
    Boolean isMember(K key, Object o);

    /**
     * 求指定集合与另一个集合的交集
     *
     * @param key 不能为null
     * @param otherKey must not be {@literal null}.
     * @return
     */
    Set<V> intersect(K key, K otherKey);

    /**
     *求指定集合与另外多个个集合交集
     *
     * @param key 不能为null
     * @param otherKeys 不能为null
     * @return
     * @see Redis Documentation: SINTER
     */
    Set<V> intersect(K key, Collection<K> otherKeys);

    /**
     * 求指定集合与另一个集合的交集，并且存储到目标集合中
     *
     * @param key 不能为null
     * @param otherKey 不能为null
     * @param destKey 不能为null
     * @return 返回目标集合的长度
     */
    Long intersectAndStore(K key, K otherKey, K destKey);

    /**
     * 求指定集合与另外多个集合中的交集保存到目标集合
     *
     * @param key 不能为null
     * @param otherKeys 不能为null
     * @param destKey 不能为null
     * @return
     */
    Long intersectAndStore(K key, Collection<K> otherKeys, K destKey);

    /**
     * 求指定集合与另一个集合的并集 并返回并集
     *
     * @param key 不能为null
     * @param otherKey 不能为null
     * @return
     * @see Redis Documentation: SUNION
     */
    Set<V> union(K key, K otherKey);

    /**
     * 求指定集合与另外多个集合的并集 并返回并集
     *
     * @param key 不能为null
     * @param otherKeys 不能为null
     * @return
     * @see Redis Documentation: SUNION
     */
    Set<V> union(K key, Collection<K> otherKeys);

    /**
     *求指定集合与另一个集合的并集，并保存到目标集合
     */
    Long unionAndStore(K key, K otherKey, K destKey);

    /**
     *求指定集合与另外多个集合的并集，并保存到目标集合
     */
    Long unionAndStore(K key, Collection<K> otherKeys, K destKey);

    /**
     * 求指定集合与另一个集合的差集
     */
    Set<V> difference(K key, K otherKey);

    /**
     * 求指定集合与另外多个集合的差集
     */
    Set<V> difference(K key, Collection<K> otherKeys);

    /**
     * 求指定集合与另一个集合的差集，并保存到目标集合
     */
    Long differenceAndStore(K key, K otherKey, K destKey);

    /**
     * 求指定集合与另外多个集合的差集，并保存到目标集合
     */
    Long differenceAndStore(K key, Collection<K> otherKeys, K destKey);

    /**
     * 获取集合中的所有元素
     */
    Set<V> members(K key);

    /**
     * 随机获取集合中的一个元素
     */
    V randomMember(K key);

    /**
     *随机返回集合中指定数量的元素。随机的元素不会重复
     */
    Set<V> distinctRandomMembers(K key, long count);

    /**
     * 随机返回集合中指定数量的元素。随机的元素可能重复
     */
    List<V> randomMembers(K key, long count);

    /**
     * 获取集合的游标。通过游标可以遍历整个集合。
     * ScanOptions 这个类中使用了构造者 工厂方法 单例。 通过它可以配置返回的元素
     * 个数 count  与正则匹配元素 match. 不过count设置后不代表一定返回的就是count个。这个只是参考
     * 意义
     *
     * @param key
     * @param options 
     * @return
     * @since 1.4
     */
    Cursor<V> scan(K key, ScanOptions options);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175

存储结构截图

5 zset（有序集合）->opsForZSet

介绍

ZSet有序且不重复，底层使用了跳跃表。相比于 Set 类型多了一个排序属性 score（分值），按照分值排序。跳跃表是一种特殊的有序链表，通过抛硬币方式，向上生成链表指向下一层，减少数据比较，使数据更快速找到定位

应用场景

1 热搜上排行榜如何实现？按照热度统计 每小时、天、周、月的排行情况。假设排行榜热度按照 热度= 转发数+点赞数+评论数，首先以每小时为单位，计算每小时的热度，key可以为当前时间戳 /1000/60/60=小时key,score为热度。那么，按天的热度则为24个小时ZSet的合并。
2 排行榜,有序集合比较典型的使用场景就是排行榜。例如学生成绩的排名榜、游戏积分排行榜、视频播放排名、电商系统中商品的销量排名等。
3 电话、姓名排序

java使用

 /**
     * zset测试
     */
    @Test
    void zsetTest() {
        //字符串的序列化器
        RedisSerializer redisSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(redisSerializer);
        redisTemplate.opsForZSet().add("zsetTest", "111", 1d);
        redisTemplate.opsForZSet().add("zsetTest", "222", 2d);
        redisTemplate.opsForZSet().add("zsetTest", "333", 3d);
        redisTemplate.opsForZSet().add("zsetTest", "444", 4d);
        //获取有序集合中指定分数范围内的成员集合  获取 value 分数
        Set<ZSetOperations.TypedTuple<Object>> zsetTest1 = redisTemplate.opsForZSet().rangeWithScores("zsetTest", 1, 3);
        System.out.println(zsetTest1);
        //从有序集合中获取指定范围内从高到低的成员集合 获取 value
        Set<Object> zsetTest2 = redisTemplate.opsForZSet().reverseRange("zsetTest", 1, 3);
        System.out.println(zsetTest2);
        //指定范围内   （0 -1）返回所有
        Set<Object> zsetTest3 = redisTemplate.opsForZSet().range("zsetTest", 1, 3);
        System.out.println(zsetTest3);
        
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

源码api

存储结构截图

6 stream（流）->opsForStream

介绍

Redis Stream 是 Redis 5.0 版本新增加的数据类型，Redis 专门为消息队列设计的数据类型。
在 Redis 5.0 Stream 没出来之前，消息队列的实现方式都有着各自的缺陷，例如：

• 发布订阅模式，不能持久化也就无法可靠的保存消息，并且对于离线重连的客户端不能读取历史消息的缺陷；
• List 实现消息队列的方式不能重复消费，一个消息消费完就会被删除，而且生产者需要自行实现全局唯一 ID。

基于以上问题，Redis 5.0 便推出了 Stream 类型也是此版本最重要的功能，用于完美地实现消息队列，它支持消息的持久化、支持自动生成全局唯一 ID、支持 ack 确认消息的模式、支持消费组模式等，让消息队列更加的稳定和可靠。

应用场景

1 消息队列

java使用

    /**
     * stream测试
     */
    @Test
    void streamTest() throws InterruptedException {
        RedisStream redisStream = new RedisStream();
        //消费者监听
        redisStream.init(redisTemplate);
        Thread.sleep(1000);
        //生产者发送消息
        redisStream.XADD(redisTemplate,"发送消息");
    }



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

package com.redis.redis01;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class RedisStream {
    public final static ExecutorService executors = Executors.newCachedThreadPool();
    public static final String key = "test:mystream";
    /**
     * add消息
     *
     * @param content
     */
    public void XADD(RedisTemplate redisTemplate,String content) {
        for (int i = 0; i < 10; i++) {
            final int y = i;
            executors.execute(new Runnable() {
                @Override
                public void run() {
                    Map<String, String> hm = new HashMap<String, String>();
                    hm.put("key", content + y);
                    // TODO Auto-generated method stub
                    StringRecord record = StreamRecords.string(hm).withStreamKey(key);
                    //synchronized (RedisStream.class) {
                    redisTemplate.opsForStream().add(record);
                    //	}
                }
            });
        }
    }
    static final Range<String> range = Range.closed("-", "+");//读取区间-代表 0 +最大值
    static final RedisZSetCommands.Limit limit = new RedisZSetCommands.Limit();
    /**
     * 阻塞读取消息
     */
    @SuppressWarnings("unchecked")
    public void XRANGE(RedisTemplate redisTemplate) {
        log.info("启动mq收到消息：:::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::");
        while (true) {
            try {
                limit.count(Integer.MAX_VALUE);//读取数量
                List<MapRecord<String, Object, Object>> list = redisTemplate.opsForStream().range(key, range, limit);
                if (list == null || list.isEmpty()) {
                    /**
                     * 阻塞读取队列
                     * StreamReadOptions.empty().block(Duration.ofDays(1)) 阻塞1day
                     * StreamOffset.latest(key) 需要监听的key
                     * 此方法会监听到队列有变化
                     */
                    redisTemplate.opsForStream().read(StreamReadOptions.empty().block(Duration.ofDays(1)), StreamOffset.latest(key));
                    list = redisTemplate.opsForStream().range(key, range, limit);
                }
                log.info("mq收到消息：" + "消息长度=" + list.size());
                int i = 0;
                int err = 0;
                for (MapRecord<String, Object, Object> mapRecord : list) {
                    RecordId recordId = mapRecord.getId();
                    long del = redisTemplate.opsForStream().delete(key, recordId);
                    if (del != 1) {
                        err++;
                        log.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>Redis Stream mq 消息异常 mapRecord=" +mapRecord);
                    } else {
                        i++;
                    }
                }
                log.info("mq消息：成功del消息=" + i + " 合计=" + (i + err));
                log.info("mq消息：失败del消息=" + err + " 合计=" + (i + err));
            } catch (Exception e) {
                continue;
            }
        }
    }
    public void init(RedisTemplate redisTemplate) {
        executors.execute(new Runnable() {
            @Override
            public void run() {
                XRANGE(redisTemplate);
            }
        });
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

源码api

存储结构截图

7 geospatial（地理）->opsForGeo

介绍

Redis GEO 是 Redis 3.2 版本新增的数据类型，主要用于存储地理位置信息，并对存储的信息进行操作。

应用场景

1 滴滴叫车
2 附近商家

java使用

  /**
     * geospatial测试
     */
    @Test
    void geospatialTest() throws InterruptedException {
        GeoUtil geoUtil = new GeoUtil(redisTemplate);
        geoUtil.geoAdd("北京西站", 116.328103, 39.900835);
        geoUtil.geoAdd("北京南站", 116.385488, 39.87128);
        geoUtil.geoAdd("北京西站-南广场", 116.327766, 39.898944);
        geoUtil.geoAdd("北京西站-南进站口", 116.327765, 39.899347);
        geoUtil.geoAdd("中铁设计大厦", 116.328628, 39.896485);
        geoUtil.geoAdd("瑞海大厦", 116.326661, 39.903778);
        // 计算北京南站与北京西站之间的距离
        double distance = geoUtil.distanceBetween("北京西站", "北京南站");
       // 5898.4001
        System.out.println(distance);
      // 查询距离北京西站5000米范围内的地方
        Map<String, Double> distanceInclude = geoUtil.distanceInclude("北京西站", 5000);
        System.out.println(distanceInclude);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

package com.redis.redis01;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResult;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

public class GeoUtil {

    private RedisTemplate<String, Object> redisTemplate;

    private static final String GEO_KEY = "DISTANCE";

    public GeoUtil(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 将经纬度信息添加到redis中
     *
     * @param certId    标识
     * @param longitude 经度
     * @param latitude  纬度
     */
    public void geoAdd(String certId, double longitude, double latitude) {
        GeoOperations geoOperations = redisTemplate.opsForGeo();
        Point point = new Point(longitude, latitude);
        RedisGeoCommands.GeoLocation geoLocation = new RedisGeoCommands.GeoLocation(certId, point);
        geoOperations.add(GEO_KEY, geoLocation);
    }
    /**
     * 两个人之间的距离
     *
     * @param certId1
     * @param certId2
     * @return
     */
    public double distanceBetween(String certId1, String certId2) {
        GeoOperations geoOperations = redisTemplate.opsForGeo();
        Distance distance = geoOperations.distance(GEO_KEY, certId1, certId2);
        return distance.getValue();
    }

    /**
     * 查询距离某个人指定范围内的人，包括距离多少米
     *
     * @param certId
     * @param distance
     * @return
     */
    public Map<String, Double> distanceInclude(String certId, double distance) {
        Map<String, Double> map = new LinkedHashMap<>();

        GeoOperations geoOperations = redisTemplate.opsForGeo();
        RedisGeoCommands.GeoRadiusCommandArgs geoRadiusCommandArgs = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs();
        GeoResults<RedisGeoCommands.GeoLocation<String>> geoResults = geoOperations.radius(GEO_KEY, certId, new Distance(distance), geoRadiusCommandArgs.includeDistance());
        if (geoResults != null) {
            Iterator<GeoResult<RedisGeoCommands.GeoLocation<String>>> iterator = geoResults.iterator();
            while (iterator.hasNext()) {
                GeoResult<RedisGeoCommands.GeoLocation<String>> geoResult = iterator.next();
                // 与目标点相距的距离信息
                Distance geoResultDistance = geoResult.getDistance();
                // 该点的信息
                RedisGeoCommands.GeoLocation<String> geoResultContent = geoResult.getContent();
                map.put(geoResultContent.getName(), geoResultDistance.getValue());
            }
        }
        return map;
    }


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

源码api

存储结构截图

8 bitmap（位图）->opsForValue().setBit

介绍

Bitmap，即位图，最大支持2^32=42.9亿bit=5亿byte=52万KB=512MB，是一串连续的二进制数组（0和1），可以通过偏移量（offset）定位元素。BitMap通过最小的单位bit来进行0|1的设置，表示某个元素的值或者状态，时间复杂度为O(1)。Bitmap 本身是用 String 类型作为底层数据结构实现的一种统计二值状态的数据类型。

应用场景

1 Bitmap 类型非常适合二值状态统计的场景，这里的二值状态就是指集合元素的取值就只有 0 和 1 两种，在记录海量数据时，Bitmap 能够有效地节省内存空间。
2 签到统计
3 判断用户登陆态
4 连续签到用户总数

1亿/8/1024/1024=10MB，bitmap适合百万一下的数据统计，如果是1w个亿级别就是120G，不适合亿级数据
，但bitmaps方法是精确计算的，比hashmap等节省空间，千亿级别适合使用hyperloglog，只存基数本身，每个hyperloglog存2的64次方

java使用

    /**
     *  bitmap测试
     */
    @Test
    void bitmapTest() {
        //用户签到
        // 1.获取当前用户id
        Long userId = 1L;
        // 2.获取当前时间
        LocalDateTime now = LocalDateTime.now();
        // 2.1 获取年月
        String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
        // 2.2 获取日
        int day = now.getDayOfMonth();
        // 3.存入redis
        // 3.1 构造key
        String key = "sign:" + userId + keySuffix;
        // 3.2 存入
        redisTemplate.opsForValue().setBit(key,day - 1,true);
    }

 /**
     * 测试对BitMaps的操作
     * 记录-查询和统计
     */
    @Test
    public void testBitMap() {
        String bitKey = "test:bit:01";
        // 记录数据状态-默认false
        redisTemplate.opsForValue().setBit(bitKey, 1, true);
        redisTemplate.opsForValue().setBit(bitKey, 4, true);
        redisTemplate.opsForValue().setBit(bitKey, 7, true);

        // 查询
        System.out.println(redisTemplate.opsForValue().getBit(bitKey, 0));
        System.out.println(redisTemplate.opsForValue().getBit(bitKey, 1));
        System.out.println(redisTemplate.opsForValue().getBit(bitKey, 2));

        // 统计
        Object execute = redisTemplate.execute(new RedisCallback() {
            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                return redisConnection.bitCount(bitKey.getBytes());
            }
        });

        System.out.println(execute);
    }

/**
 * OR运算
 * 统计3组数据的布尔值, 并对这3组数据做OR运算.
 */
@Test
public void testBitMapOperation() {
    String bitKey2 = "test:bm:02";
    redisTemplate.opsForValue().setBit(bitKey2, 0, true);
    redisTemplate.opsForValue().setBit(bitKey2, 1, true);
    redisTemplate.opsForValue().setBit(bitKey2, 2, true);

    String bitKey3 = "test:bm:03";
    redisTemplate.opsForValue().setBit(bitKey3, 2, true);
    redisTemplate.opsForValue().setBit(bitKey3, 3, true);
    redisTemplate.opsForValue().setBit(bitKey3, 4, true);

    String bitKey4 = "test:bm:04";
    redisTemplate.opsForValue().setBit(bitKey4, 4, true);
    redisTemplate.opsForValue().setBit(bitKey4, 5, true);
    redisTemplate.opsForValue().setBit(bitKey4, 6, true);

    // 合并处理
    String bitKeyOR = "test:bm:or";
    Object obj = redisTemplate.execute(new RedisCallback() {
        @Override
        public Object doInRedis(RedisConnection connection) throws DataAccessException {
            connection.bitOp(RedisStringCommands.BitOperation.OR,
                    bitKeyOR.getBytes(), bitKey2.getBytes(), bitKey3.getBytes(), bitKey4.getBytes());
            return connection.bitCount(bitKeyOR.getBytes());
        }
    });

    System.out.println(obj);    // 统计的个数

    // 合并后，每位的状态
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 0));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 1));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 2));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 3));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 4));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 5));
    System.out.println(redisTemplate.opsForValue().getBit(bitKeyOR, 6));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

源码api

存储结构截图

9 bitfield（位域）

介绍

应用场景

java使用

源码api

存储结构截图

10 hyperloglog（基数统计）

介绍

Redis HyperLogLog 是 Redis 2.8.9 版本新增的数据类型，是一种用于「统计基数」的数据集合类型，基数统计就是指统计一个集合中不重复的元素个数。但要注意，HyperLogLog 是统计规则是基于概率完成的，不是非常准确，标准误算率是 0.81%,HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的内存空间总是固定的、并且是很小的。因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身;
什么是基数:数据集中不重复的元素的个数

应用场景

1 百万级网页 UV 计数,Redis HyperLogLog 优势在于只需要花费 12 KB 内存，就可以计算接近 2^64 个元素的基数，和元素越多就越耗费内存的 Set 和 Hash 类型相比，HyperLogLog 就非常节省空间。
所以，非常适合统计百万级以计网站的独立访客数场景。
2 注册 IP 数、每日访问 IP 数、页面实时UV）、在线用户数等

java使用

    /**
     *  hyperloglog测试
     */
    @Test
    void hyperloglogTest() {
        // 添加100 000个不重复的数、100 000个重复的数-共20万个数
        String pfKey = "test:hll:01";
        for (int i = 0; i < 100000; i++) {
            redisTemplate.opsForHyperLogLog().add(pfKey, i);
        }
        for (int i = 0; i < 100000; i++) {
            int r = (int)(Math.random() * 100000);
            redisTemplate.opsForHyperLogLog().add(pfKey, r);
        }
        // 统计指定key中所有不重复的基数个数
        long size = redisTemplate.opsForHyperLogLog().size(pfKey);
        System.out.println(size);

    }

 /**
     * 合并数据-并统计合并后的基数
     */
    @Test
    public void testHyperLogLogUnion() {
        String pfKey2 = "test:hll:02";
        String pfKey3 = "test:hll:03";
        String pfKey4 = "test:hll:04";
        for (int i = 0; i < 10000; i++) {
            redisTemplate.opsForHyperLogLog().add(pfKey2, i);
        }
        for (int i = 5000; i < 15000; i++) {
            redisTemplate.opsForHyperLogLog().add(pfKey3, i);
        }
        for (int i = 10000; i < 20000; i++) {
            redisTemplate.opsForHyperLogLog().add(pfKey4, i);
        }
        // 合并三组数
        String unionKey = "test:hll:union";
        redisTemplate.opsForHyperLogLog().union(unionKey, pfKey2, pfKey3, pfKey4);
        // 统计合并后的基数
        long size = redisTemplate.opsForHyperLogLog().size(unionKey);
        System.out.println(size);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

源码api

存储结构截图