[大数据]【Java从0到架构师】Redis - 进阶（pipline、发布订阅、Bitmap、HyperLogLog、GEO）

Java 从 0 到架构师目录：【Java从0到架构师】学习记录

一些概念：PV（Page View）访问量，UV（Unique Visitor）独立访客

Jedis 的基本使用

参考文章：Redis 笔记之 Java 操作 Redis（Jedis）

<dependencies>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.1.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.13</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

Redis 数据淘汰策略

• lft：根据 key 的使用次数多少进行淘汰
• lru：根据 key 的最近使用时间进行淘汰

在 Redis 中，允许用户设置最大的内存大小（redis.conf 中配置）

maxmemory 1G
maxmemory-policy noeviction

maxmemory-policy：

• volatile-lru 设定超时时间的数据中，删除最不常用的数据
• allkeys-lru 查询所有的 key 中最不常使用的数据进行删除
• volatile-random 在已经设定了超时的数据中随机删除
• allkeys-random 查询所有的 key 之后随机删除
• volatile-ttl 查询全部设定超时时间的数据，进行排序，删除即将过期的数据
• noeviction 默认策略，设置该属性，则不会进行删除操作，如果内存溢出则报错返回
• volatile-lfu 从所有配置了过期的时间的键中删除使用频率最少的键
• allkeys-lfu 从所有键中删除使用频率最少的键

Redis 进阶拓展

命令的执行流程：

pipline - 命令批处理，减少大量命令的网络开销，提高操作性能

pipline 应用：

1. 多个命令之间可以并行执行，没有前后的关联关系

   pipline 不保证原子性，只是将一堆命令发送过去执行；mset 这种命令是保证原子性的

2. 需要批处理执行的操作，需要同时不同多个数据放到内存中

3. 时间的消耗更多的是在网络层面， 在执行命令的时候需要的时间是 us（微秒）级别

代码实现：

public class JedisDemo {

    private Jedis jedis = null;

    @Before
    public void init() throws Exception {
        jedis = new Jedis("192.168.48.101", 6379);
    }

    @Test
    public void testJedis() throws Exception {
        jedis.flushAll();
        Long start = System.currentTimeMillis(); // 获取开始时间
        for (int i = 0; i < 100000; i++) {
            jedis.set("name" + i, i + "");
        }
        Long end = System.currentTimeMillis(); // 获取结束时间
        System.out.println(end - start);
        jedis.close();
    }
    
    @Test
    public void testPipeline() throws Exception {
        jedis.flushAll();
        Long start = System.currentTimeMillis(); // 获取开始时间
        Pipeline pipelined = jedis.pipelined();
        for (int i = 0; i < 100000; i++) {
            pipelined.set("name" + i, i + "");
        }
        pipelined.sync();
        Long end = System.currentTimeMillis(); // 获取结束时间
        System.out.println(end - start);
        jedis.close();
    }
}

发布订阅 - subscribe

了解即可，实现发布订阅有专门的消息中间件

发布订阅中的角色：

• 发布者 publisher
• 订阅者 subscriber
• 频道 channel

# 订阅一个或多个频道 subscribe <channel> 
subscribe sohu:tv

# 发布消息 publish <channel> <message>
publish sohu:tv "hello world"

# 取消订阅 unsubscirbe

测试案例：

• 生产者：

public class PublisherDemo {

    private Jedis jedis = null;

    @Before
    public void init() throws Exception {
        jedis = new Jedis("192.168.48.101", 6379);
    }

    @After
    public void destroy() throws Exception {
        jedis.close();
    }

    @Test
    public void publish() throws Exception {
        for (int i = 0; i < 100; i++) {
            jedis.publish("channel02", "value" + i);
            TimeUnit.MICROSECONDS.sleep(10);
        }
    }
}

• 消费者：

public class Subscriber {

    private Jedis jedis = null;
    
    @Before
    public void init() throws Exception {
        jedis=new Jedis("192.168.48.101", 6379);
    }
    
    @After
    public void destroy() throws Exception {
        jedis.close();
    }
    
    @Test
    public void subscribe() throws Exception {
        JedisPubSub jedisPubSub = new JedisPubSub(){
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("channel = " + channel);
                System.out.println("message = " + message);
            }
        };
        jedis.subscribe(jedisPubSub, "channel01", "channel02");
    }
}

Bitmap - 一串连续的二进制数字（字符串），每一位所在的位置为偏移

Bitmap 是一串连续的 2 进制数字（0 或 1），每一位所在的位置为偏移 (offset)

setbit、getbit、bitcount：

# 设置指定的bit
127.0.0.1:6379> setbit qq:uv 1002 1
127.0.0.1:6379> setbit qq:uv 1003 1

# 获取指定位置的值, 返回1或者0 
127.0.0.1:6379> getbit qq:uv 1003
(integer) 1

# bitcount: 统计值为1的个数
127.0.0.1:6379> BITCOUNT qq:uv 
(integer) 1

bitop：

# 对一个或多个 key 求逻辑与(默认和1)，并将结果保存到 destkey
BITOP AND destkey key [key ...]

1# 对一个或多个 key 求逻辑或(默认和0)，并将结果保存到 destkey
BITOP OR destkey key [key ...]

# 对一个或多个 key 求逻辑异或，并将结果保存到 destkey
BITOP XOR destkey key [key ...]

# 对给定 key 求逻辑非，并将结果保存到 destkey
BITOP NOT destkey key

bitpos：

# bitpos: 用来返回操作的索引位置
# 5:代表从第5个字节开始查找, 8:代表从第8个字节开始查找
127.0.0.1:6379> BITPOS qq:uv 1 5 8

应用：登陆用户数量统计，实现记录用户哪天进行了登录，每天只记录是否登录过，重复登录状态算已登录。不需要记录用户的操作行为，不需要记录用户上次的登录时间和 ip 地址.

比如现在有如下用户:
id	  name
1 	  张三
3	  李四
8	  王五

# bitset key 偏移量 [0,1]
# 李四登录了, 就记录李四今天登录了
bitset login_20191206 3 1
# 此时 bitmap 中存储如下: 00000000 00001000

# 王五登录了, 记录王五今天登陆了
bitset login_20191206 8 1
# 此时bitmap中存储如下: 00000001 00001000

代码实现：

• 初始化数据：

@Test
public void initData() throws Exception {
    // 添加登录过的用户
    jedis.setbit("user:login:20190919", 1, "1"); // 设置id为1的用户登录过
    jedis.setbit("user:login:20190919", 8, "1"); // 设置id为8的用户登录过
    jedis.setbit("user:login:20190919", 12, "1"); // 设置id为12的用户登录过
    jedis.setbit("user:login:20190920", 8, "1"); // 设置id为8的用户登录过
    jedis.setbit("user:login:20190920", 22, "1"); // 设置id为22的用户登录过
    jedis.setbit("user:login:20190921", 8, "1"); // 设置id为8的用户登录过
    jedis.setbit("user:login:20190921", 24, "1"); // 设置id为22的用户登录过
}

• 统计登陆人数：

@Test
public void useData() throws Exception {
    // 1 统计 20190919的登录用户数
    System.out.println(jedis.bitcount("user:login:20190919"));
    // 2 统计 最近三天的登录用户总数
	jedis.bitop(BitOP.OR, "user:login:last3_1", 
						  "user:login:20190919",
						  "user:login:20190920",
						  "user:login:20190921");
    System.out.println(jedis.bitcount("user:login:last3_1"));
    // 3 统计 连续三天都登录过的用户 用户数
    jedis.bitop(BitOP.AND, "user:login:last3_2", 
    					   "user:login:20190919",
    					   "user:login:20190920",
    					   "user:login:20190921");
    System.out.println(jedis.bitcount("user:login:last3_2"));
}

HyperLogLog - 一种基数统计算法

HyperLogLog 是用来做基数统计的算法（不准确，模糊值）

• 在输入元素的数量或者体积非常大时，计算基数所需的空间总是固定的、并且是很小的

在 Redis 中，每个 HyperLogLog 键只需要花费 12KB 内存，就可以计算接近 2^64（约42亿）个不同元素的基数

这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比

使用 HyperLogLog 应当考虑的因素：

• 是否能容忍错误（0.81%）
• 是否需要单条数据
• 是否需要使用很小的内存

# pfadd: 将任意数量的元素添加到指定的 HyperLogLog 里面
pfadd qq:tv:uv 003 004 005

# pfcount: 用来统计元素的个数
127.0.0.1:6379> pfcount qq:tv:uv

# pfmerge: 把多个key合并为一个key, 自动过滤重复元素
pfmerge tv:uv qq:tv:uv aiqiyi:tv:uv

代码实现：

// 添加登录过的用户 添加一万个用户
Pipeline pipelined = jedis.pipelined();
for (int i = 0; i < 5000000; i++) {
    pipelined.pfadd("pf:user:login:20190919","user"+i);
}
pipelined.sync();
System.out.println(jedis.pfcount("pf:user:login:20190919"));

GEO - 地理信息定位，存储经纬度，计算两地距离，计算范围

geoadd 添加一个地理位置信息
geopos 获取地理位置信息
geodist 获取两个地理位置
georadius 获取指定范围内的数据

经纬度查询：https://lbs.amap.com/tools/picker

代码实现：

• 封装经纬度信息：

@Data
@NoArgsConstructor
@AllArgsConstructor
public class CityPosition {
    // 经度
    private double lng;    
    // 纬度
    private double lat;    
    // 商家id
    private String key;    
}

• 初始化对应的城市数据：

@Test
public void initData() throws Exception {
    CityPosition s1 = new CityPosition (113.264385, 23.129112, "广州");
    CityPosition s2 = new CityPosition (121.473658, 31.230378, "上海");
    CityPosition s3 = new CityPosition (114.085947, 22.547, "深圳");
    CityPosition s4 = new CityPosition (113.746262, 23.046237, "东莞");
    jedis.geoadd("city", s1.getLng(), s1.getLat(), s1.getKey());
    jedis.geoadd("city", s2.getLng(), s2.getLat(), s2.getKey());
    jedis.geoadd("city", s3.getLng(), s3.getLat(), s3.getKey());
    jedis.geoadd("city", s4.getLng(), s4.getLat(), s4.getKey());
}

• 获取对应的城市信息：

@Test
public void getInfo() throws Exception {
    System.out.println(jedis.geodist("city", "广州", "东莞", GeoUnit.KM));
    List<GeoRadiusResponse> datas = jedis.georadius("city", 113.264385,23.129112, 100, GeoUnit.KM);
    for (GeoRadiusResponse data : datas) {
        System.out.println(data);
        System.out.println("城市: = " + new String(data.getMember()));
    }
}