关于redis:Redis在项目中的应用总结

2次阅读

共计 5129 个字符，预计需要花费 13 分钟才能阅读完成。

一个详情页蕴含了比拟多的信息，这些信息大多是动态的，为了进步详情页的加载速度，将这些信息缓存到 Redis 中。

如何解决缓存和数据库的数据不统一问题？

我的项目中应用的是 先更新数据库，后删除缓存，在非并发读写状况下，简直不存在问题，在并发读写状况，当线程 W 更新完数据库，删除缓存后，R 申请此时读取缓存，R 申请读取缓存不存在，此时查询数据库，主从同步还没有实现或者产生了主从同步提早，R 申请从读库上获取了旧的数据，而后写入到缓存中，那么缓存和数据库就会产生不统一。

下面这种状况，能够通过提早再删除的办法解决，W 申请删除缓存后，发送一条提早音讯，音讯的生产端依据理论状况再次删除缓存，生产端比拟缓存中的数据版本和数据库中数据的版本，如果不统一，那么再次删除缓存，如果统一，无需操作。

对于数据库和缓存一致性更加具体的剖析参考这篇文章

所谓的聚合统计，就是指统计多个汇合元素的聚合后果，包含：统计多个汇合的共有元素（交加统计）；把两个汇合相比，统计其中一个汇合独有的元素（差集统计）；统计多个汇合的所有元素（并集统计）。

统计用户 App 每天的新增用户数
用一个汇合记录所有登录过 App 的用户 ID，同时，用另一个汇合记录每一天登录过 App 的用户 ID。而后，再对这两个汇合做差集统计。

咱们能够应用 Set 类型，把 key 设置为 user:id，示意记录的是用户 ID，value 就是一个 Set 汇合，外面是所有登录过 App 的用户 ID，咱们能够把这个 Set 叫作累计用户 Set，如下图所示：

把每一天登录的用户 ID，记录到一个新汇合中，咱们把这个汇合叫作每日用户 Set，它有两个特点：
1、key 是 user:id 以及当天日期，例如 user20200803；
2、value 是 Set 汇合，记录当天登录的用户 ID。

在统计每天的新增用户时，咱们只用计算每日用户 Set 和累计用户 Set 的差集就行，伪代码如下：

Set difference = getRedisTemplate().opsForSet().difference("user:id", "user:id:20200803");

而后，咱们计算累计用户 Set 和 user20200803 Set 的并集后果，后果保留在 user:id 这个累计用户 Set 中，伪代码如下：

Set union = getRedisTemplate().opsForSet().union("user:id","user:id:20200803");

此时，user:id 这个累计用户 Set 中就有了 8 月 3 日的用户 ID。等到 8 月 4 日再统计时，咱们把 8 月 4 日登录的用户 ID 记录到 user20200804 的 Set 中。接下来，计算累计用户 Set 和 user20200804 Set 的差集就是 8 月 4 日的新增用户

计算第二天的留存用户数
当要计算 8 月 4 日的留存用户时，咱们只须要再计算 user20200803 和 user20200804 两个 Set 的交加，就能够失去同时在这两个汇合中的用户 ID 了，这些就是在 8 月 3 日登录，并且在 8 月 4 日留存的用户，伪代码如下：

Set intersect = getRedisTemplate().opsForSet().intersect("user:id:20200803","user:id:20200804");

Set 的差集、并集和交加的计算复杂度较高，在数据量较大的状况下，如果间接执行这些计算，会导致 Redis 实例阻塞。所以，我给你分享一个小倡议：你能够从主从集群中抉择一个从库，让它专门负责聚合计算，或者是把数据读取到客户端，在客户端来实现聚合统计，这样就能够躲避阻塞主库实例和其余从库实例的危险了。

统计管家 App 上的打卡状况
在签到统计时，每个用户一天的签到用 1 个 bit 位就能示意，一个月（假如是 31 天）的签到状况用 31 个 bit 位就能够，而一年的签到也只须要用 365 个 bit 位，基本不必太简单的汇合类型。这个时候，咱们就能够抉择 Bitmap。Bitmap 能够看作是一个 bit 数组。

Bitmap 提供了 GETBIT/SETBIT 操作，应用一个偏移值 offset 对 bit 数组的某一个 bit 位进行读和写。不过，须要留神的是，Bitmap 的偏移量是从 0 开始算的，也就是说 offset 的最小值是 0。

假如统计 ID 3000 的用户在 2020 年 8 月份的签到状况，就能够依照上面的步骤进行操作。
第一步，执行上面的命令，记录该用户 8 月 3 号已签到：

getRedisTemplate().opsForValue().setBit("uid:sign:3000:202008", 2, true);

第二步，查看该用户 8 月 3 日是否签到：

Boolean sign = getRedisTemplate().opsForValue().getBit("uid:sign:3000:202008", 2);

第三步，统计该用户在 8 月份的签到次数：

long signCount = (long)getRedisTemplate().execute((RedisCallback<Long>) con -> con.bitCount("uid:sign:3000:202008".getBytes()));

基于下面签到签到统计的办法，如果记录了 1 亿个用户 10 天的签到状况，怎么统计出这 10 天间断签到的用户总数？

Bitmap 反对用 BITOP 命令对多个 Bitmap 按位做“与”“或”“异或”的操作，操作的后果会保留到一个新的 Bitmap 中。以按位“与”操作为例来具体解释一下。从下图中，能够看到，三个 Bitmap bm1、bm2 和 bm3，对应 bit 位做“与”操作，后果保留到了一个新的 Bitmap 中（示例中，这个后果 Bitmap 的 key 被设为“resmap”）

在统计 1 亿个用户间断 10 天的签到状况时，能够把每天的日期作为 key，每个 key 对应一个 1 亿位的 Bitmap，每一个 bit 对应一个用户当天的签到状况。

接下来，对 10 个 Bitmap 做“与”操作，失去的后果也是一个 Bitmap。在这个 Bitmap 中，只有 10 天都签到的用户对应的 bit 位上的值才会是 1。最初，用 BITCOUNT 统计下 Bitmap 中的 1 的个数，这就是间断签到 10 天的用户总数了

所以，如果只须要统计数据的二值状态，例如商品有没有、用户在不在等，就能够应用 Bitmap，因为它只用一个 bit 位就能示意 0 或 1。在记录海量数据时，Bitmap 可能无效地节俭内存空间。

基数统计就是指统计一个汇合中不反复的元素个数。对应到理论的场景中，就是统计网页的 UV。

网页 UV 的统计有个独特的中央，就是须要去重，一个用户一天内的屡次拜访只能算作一次。在 Redis 的汇合类型中，Set 类型默认反对去重，所以看到有去重需要时，咱们可能第一工夫就会想到用 Set 类型。但对于大数据量，Set 会耗费很大的内存空间。

HyperLogLog 是一种用于统计基数的数据汇合类型，它的最大劣势就在于，当汇合元素数量十分多时，它计算基数所需的空间总是固定的，而且还很小。在 Redis 中，每个 HyperLogLog 只须要破费 12 KB 内存，就能够计算靠近 2^64 个元素的基数。

在统计 UV 时，把拜访页面的每个用户都增加到 HyperLogLog 中。如下伪代码：

getRedisTemplate().opsForHyperLogLog().add("page1:uv", "user1", "user2", "user3", "user3");

接下来，就能够取得 page1 的 UV 值了，HyperLogLog 的统计后果：

long pfCount = (long)getRedisTemplate().execute((RedisCallback<Long>) con -> con.pfCount("page1:uv".getBytes()));

HyperLogLog 的统计规定是基于概率实现的，所以它给出的统计后果是有肯定误差的，规范误算率是 0.81%。这也就意味着，你应用 HyperLogLog 统计的 UV 是 100 万，但理论的 UV 可能是 101 万。尽管误差率不算大，然而，如果你须要准确统计后果的话，最好还是持续用 Set 或 Hash 类型。

加锁代码：

public String lock(final CacheKey key, final Long expire, final TimeUnit unit) {
    try {String result = (String)this.redisCacheTemplate.execute(new RedisCallback<String>() {public String doInRedis(RedisConnection connection) throws DataAccessException {JedisCommands commands = (JedisCommands)connection.getNativeConnection();
                String uuid = UUID.randomUUID().toString();
                String result = commands.set(key.getCacheKey(), uuid, "NX", "PX", unit.toMillis(expire));
                return "OK".equals(result) ? uuid : null;
            }
        });
        return result;
    } catch (Exception var5) {logger.error("set redis occured an exception", var5);
        return null;
    }
}

解锁代码：

public boolean releaseLock(CacheKey key, String token) {if (token == null) {logger.warn("lock token can not be null for key=" + key);
        return false;
    } else {
        try {final List<String> keys = new ArrayList();
            keys.add(key.getCacheKey());
            final List<String> args = new ArrayList();
            args.add(token);
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Long result = (Long)this.redisCacheTemplate.execute(new RedisCallback<Long>() {public Long doInRedis(RedisConnection connection) throws DataAccessException {Object nativeConnection = connection.getNativeConnection();
                    if (nativeConnection instanceof JedisCluster) {return (Long)((JedisCluster)nativeConnection).eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", keys, args);
                    } else {return nativeConnection instanceof Jedis ? (Long)((Jedis)nativeConnection).eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", keys, args) : 0L;
                    }
                }
            });
            return result != null && result == 1L;
        } catch (Exception var7) {logger.error("release lock occured an exception", var7);
            return false;
        }
    }
}

对于分布式锁的高可用性，为了防止 Redis 实例故障而导致的锁无奈工作的问题，Redis 的开发者 Antirez 提出了分布式锁算法 Redlock。

Redlock 算法的基本思路，是让客户端和多个独立的 Redis 实例顺次申请加锁，如果客户端可能和半数以上的实例胜利地实现加锁操作，那么就认为，客户端胜利地取得分布式锁了，否则加锁失败。

正文完