一.统计每个页面的UV

UV（Unique Visitor）独立访客，统计1天内访问某站点的用户数。每个用户每天在同一个页面浏览多次，也只记为一次。技术方案有如下几种：
1.大数据部门使用Spark、Flink等进行处理
2.Set
3.bitmap
4.HyperLogLog 算法

Set

以PageID:UV作为key，用户ID作为V，直接进行存放。使用SADD添加数据，由于本身Set不会重复的特性，重复提交也不会有问题。如果每天计算一次，那么按照日期PageID:日期:UV作为Key即可。

package com.brianxia.redisinaction;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.ArrayList;
import java.util.List;

@SpringBootTest
class UVTests {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static List<Long> userIds = new ArrayList<>();

    @BeforeAll
    static void addUser(){
        //添加用户ID
        for (long i = 0; i < 100000; i++) {
            userIds.add(i);
        }
    }


    @Test
    void set() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";
        BoundSetOperations<String, String> setOperations = stringRedisTemplate.boundSetOps(key);
        userIds.forEach(id -> {
            setOperations.add(String.valueOf(id));
        });
    }

}

添加后的内存情况:

image.png

添加前的内存情况

image.png
总计使用内存：7,652,864字节。

Bitmap

如果userId是整型，而且是从1开始连续自增的，那么使用bitmap也是不错的选择。只需要在bitmap执行的位上设置成1就可以代表用户在当天访问了该页面（比如userId是100，那么就在第100位上将bit设置成1）。

  @Test
    void bitmap() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";
        userIds.forEach(id -> {
            stringRedisTemplate.opsForValue().setBit(key,id,true);
        });
    }

添加后的内存情况:

image.png

添加前的内存情况

image.png

总计使用内存：20,544字节。

HyperLogLog 算法

Redis 在 2.8.9 版本添加了 HyperLogLog 结构, 它的优势就是每个key仅需12kb的内存, 就能存储 2^64 个不同元素的基数, 存储空间小且固定, 缺点就是元数据无法直接提取了（无法判断某个用户是否看过此页面）。HyperLogLog 提供不精确的去重计数方案，标准误差大概在 0.81%，这样的精确度已经可以满足上面的用户访问量的统计需求了。

 @Test
    void hyperloglog() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";
        userIds.forEach(id -> {
            stringRedisTemplate.opsForHyperLogLog().add(key,String.valueOf(id));
        });

        System.out.println(stringRedisTemplate.opsForHyperLogLog().size("1:uv"));
    }

添加后的内存情况:

image.png

添加前的内存情况

image.png

总计使用内存：16,448字节。

统计性能比较

分别循环十万次对每种算法统计一个key对应的count数，代码如下:

package com.brianxia.redisinaction;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.ArrayList;
import java.util.List;

@SpringBootTest
class UVTests {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static List<Long> userIds = new ArrayList<>();

    @BeforeAll
    static void addUser(){
        //添加用户ID
        for (long i = 0; i < 100000; i++) {
            userIds.add(i);
        }

    }


    @Test
    void set() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";

        stringRedisTemplate.delete(key);

        BoundSetOperations<String, String> setOperations = stringRedisTemplate.boundSetOps(key);
        userIds.forEach(id -> {
            setOperations.add(String.valueOf(id));
        });

        //6597
        Long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            Long size = setOperations.size();
        }
        Long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

    @Test
    void bitmap() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";
        stringRedisTemplate.delete(key);
        userIds.forEach(id -> {
            stringRedisTemplate.opsForValue().setBit(key,id,true);
        });

        //6976
        Long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            Long size = stringRedisTemplate.execute((RedisCallback<Long>) con -> con.bitCount(key.getBytes()));
        }
        Long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

    @Test
    void hyperloglog() {
        //1.构造redis key,页面ID暂时固定为1
        String key = "1:uv";
        stringRedisTemplate.delete(key);
        userIds.forEach(id -> {
            stringRedisTemplate.opsForHyperLogLog().add(key,String.valueOf(id));
        });

        //6442
        Long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            Long size = stringRedisTemplate.opsForHyperLogLog().size("1:uv");
        }
        Long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

}

基本上每种数据结构的执行时间都比较接近，具体参考https://redis.io/commands。

数据结构	执行时间（ms/十万次）	时间复杂度
set scard	6597	O(1)
bitmap bitcount	6976	O(n)
hyperloglog pfcount	6442	O(1)

结论

如果是需要精确计算，建议使用bitmap，并且bitmap可以判断某个用户是否浏览过此页面。不需要精确计算的场景下，建议使用hyperloglog。

二.移动端签到

需求1：千万级别用户，需要统计用户的签到情况。

//需求1：千万级别用户，需要统计用户的签到情况。
    @Test
    void action1() {
        //key的设计 signin:用户id:月份
        Long userId = 1000L;
        String key = "signin:" + userId + "202103";

        for (int i = 1; i <= 30; i++) {
            //日期能被2整除就模拟为签到，否则就未签到
            stringRedisTemplate.opsForValue().setBit(key,i,i%2 == 0? true :false );
        }
        //获取当月登录总天数
        Long size = stringRedisTemplate.execute((RedisCallback<Long>) con -> con.bitCount(key.getBytes()));
        System.out.println(size);

        for (int i = 1; i <= 30; i++) {
            //获取每天登录情况
            Boolean bit = stringRedisTemplate.opsForValue().getBit(key, i);
            System.out.println("2021/3/" + i + " " + bit);
        }
    }

使用bitmap上的位记录某一天是否登录，设计key时，使用signin:用户id:月份用来定位某个用户在某一个月份的数据。

需求2：千万级别用户，统计用户连续签到情况。

 //需求2：千万级别用户，统计用户连续签到情况。
    @Test
    void action2() {
        //key的设计 signin:day:日期
        Long userId = 1000L;
        String key = "signin:day:2021:3:";

        for (int i = 1; i <= 31; i++) {
            //日期能被2整除就模拟为签到，否则就未签到
            stringRedisTemplate.opsForValue().setBit(key+i,userId,i==1?false:true );
        }
        //计算7天的累加，放到signin:day:result中
        Long size = stringRedisTemplate.execute((RedisCallback<Long>) con
                -> con.bitOp(RedisStringCommands.BitOperation.AND,"signin:day:result".getBytes(),
                "signin:day:2021:3:1".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:2".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:3".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:4".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:5".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:6".getBytes(StandardCharsets.UTF_8),
                "signin:day:2021:3:7".getBytes(StandardCharsets.UTF_8)));

        System.out.println(stringRedisTemplate.opsForValue().getBit("signin:day:result",userId));
    }

使用1个bitmap存储每个用户的登录情况，每天存储一份，key设计为 signin:day:日期。通过bitop and指令计算交集之后存储到signin:day:result中，最后找到指定位置（案例中就是1000）就可以判断用户是否在这7天连续登录了。

三.统计社交网站的用户好友

需求1：查找A和B的共同好友。

//需求1：查找A和B的共同好友。
    @Test
    void action1() {
        //使用set保存好友数据
        Long userIdA = 1L;
        Long userIdB = 2L;

        BoundSetOperations<String, String> setOperationsA = stringRedisTemplate.boundSetOps("friend:" + userIdA);
        BoundSetOperations<String, String> setOperationsB = stringRedisTemplate.boundSetOps("friend:" + userIdB);

        //模拟数据
        for (int i = 100; i < 200; i++) {
            setOperationsA.add(String.valueOf(i));
            if(i % 2 ==0){
                setOperationsB.add(String.valueOf(i));
            }
        }

        //交集操作
        System.out.println(setOperationsA.intersect("friend:" + userIdB));
    }

将好友数据放入到两个set中，直接使用intersect交集操作即可。

需求2：查找A的潜在好友（BCD有但是A没有的好友）。

//需求2：查找A的潜在好友（BCD有但是A没有的好友）。
    @Test
    void action2() {
        //使用set保存好友数据
        Long userIdA = 1L;
        Long userIdB = 2L;
        Long userIdC = 2L;

        stringRedisTemplate.delete(Arrays.asList("friend:" + userIdA,"friend:" + userIdB,"friend:" + userIdC));
        BoundSetOperations<String, String> setOperationsA = stringRedisTemplate.boundSetOps("friend:" + userIdA);
        BoundSetOperations<String, String> setOperationsB = stringRedisTemplate.boundSetOps("friend:" + userIdB);
        BoundSetOperations<String, String> setOperationsC = stringRedisTemplate.boundSetOps("friend:" + userIdC);

        //模拟数据
        for (int i = 100; i < 200; i++) {
            if(i % 8 ==0){
                setOperationsA.add(String.valueOf(i));
            }
            if(i % 2 ==0){
                setOperationsB.add(String.valueOf(i));
            }
            if(i % 4 ==0){
                setOperationsC.add(String.valueOf(i));
            }
        }

        //将B和C取交集，共同好友
        setOperationsB.intersectAndStore("friend:" + userIdC,"friend:result");
        BoundSetOperations<String, String> setOperationsR = stringRedisTemplate.boundSetOps("friend:result");
        //取差集
        Set<String> diff = setOperationsR.diff("friend:" + userIdA);
        //获取数据
        System.out.println(diff);
    }

先将BC取交集获取共同好友，再将结果存入redis的set中。最后将结果集合和A的集合取差集即可。