①String#

SET

1
SET key value [NX|XX]  [GET]  [EX seconds | PX milliseconds | EXAT Unix-time-seconds |
2
PXAT Unix-time-milliseconds | KEEPTTL]

MGET/MSET/MSETNX

1
MGET/MSET/MSETNX KEY VALUE KEY VALUE ....
2
MGET/MSET/MSETNX KEY VALUE KEY VALUE ....

GETRANGE/SETRANGE

1
GETRANGE KEY start end
2
SETRANGE KEY offset value

INCR/DECR INCRBY/DECRBY

1
INCR/DECR KEY
2
INCRBY/DECRBY KEY STEP

STRLEN、APPEND

1
STRLEN KEY
2
APPEND KEY VALUE

SETNX/SETEX(分布式锁)

1
SETEX KEY TTL VALUE
2
SETNX KEY VALUE

GETSET

1
GETSET KEY

②List#

LPUSH/RPUSH/LRANGE

1
LPUSH/RPUSH KEY VALUE
2
LRANGE KEY start end

​

LPOP/RPOP

1
LPOP/RPOP KEY

​

LINDEX

1
LINDEX KEY INDEX

​

LLEN

1
LLEN KEY

​

LREM

1
LREM KEY N VALUE

​

LTRIM

1
ITRIM KEY start end

​

RPOPLPUSH

1
RPOPLPUSH list list

​

LSET

1
LSET KEY INDEX VALUE

​

LINSERT

1
LINSERT KEY BEFORE / AFTER VALUE-1 VALUE-2

​

③Hash#

HSET/HGET/HMSET/HMGET/HGETALL/HDEL

1
HSET KEY KEY VALUE [KEY VALUE]
2
HGET KEY KEY
3
HMSET KEY KEY VALUE [KEY VALUE]
4
HMGET KEY KEY [KEY]
5
HGETALL KEY
6
HDEL KEY KEY

​ ​ ​

HLEN

1
HLEN KEY

​

HEXISTS

1
HEXISTS KEY KEY

​

HKEYS/HVALS

1
HKEYS KEY
2
HVALS KEY

​

HINCRBY/HINCRBYFLOAT

1
HINCRBY KEY KEY STEP
2
HINCRBYFLOAT KEY KEY STEP

​

HSETNX

1
HSETNX KEY KEY VALUE [KEY VALUE]

​

④Set#

SADD

1
SADD KEY VALUE [VALUE]

​

SMEMBERS

1
SMEMBERS KEY

​

SISMEMBER

1
SISMEMBER KEY VALUE

​

​

SREM

1
SREM KEY VALUE [VALUE]

​

SCARD

1
SCARD KEY

​

SRANDMEMBER

1
SRANDMEMBER KEY N

​

SPOP

1
SPOP KEY

​

SMOVE

1
SMOVE KEY-1 KEY-2 VALUE

​

集合运算

差集 SDIFF

1
SDIFF KEY-1 KEY-2

​

并集 SUNION

1
SUNION KEY-1 KEY-2

​

交集 SINTER/SINTERCARD

1
SINTER KEY [KEY]
2
SINTERCARD N KEY [KEY] [LIMIT M]

​ ​

⑤ZSet#

ZADD

1
ZADD KEY SCORE VALUE [SCORE VALUE]

​

ZRANGE

1
ZRANGE KEY start end [WITHSCORES]

​ ​

ZREVRANGE

1
ZREVRANGE KEY start end [WITHSCORES]

​

ZRANGEBYSCORE

1
ZRANGEBYSCORE KEY [(] MIN MAX [WITHSCORE] [LIMIT offset count]

​

ZSCORE

1
ZSCORE KEY VALUE

​

ZCARD

1
ZCARD KEY

​

ZREM

1
ZREM KEY VALUE

​

ZINCRBY

1
ZINCRBY KEY N VALUE

​

ZCOUNT

1
ZCOUNT KEY MIN MAX

​

ZMPOP

1
ZMPOP N KEY [KEY] [MAX|MIN] count M

​

ZRANK

1
ZRANK KEY VALUE

​

ZREVRANK

1
ZREVRANK KEY VALUE

​

⑥BitMap#

​

SETBIT

1
SETBIT KEY offset value

​

GETBIT

1
GETBIT KEY offset

​

STRLEN

1
STRLEN KEY

​

BITCOUNT

1
BITCOUNT KEY [start end]

​

BITOP

1
BITOP OPERATION DESTKEY KEY [KEY]

​

⑦HyperLogLog#

PFADD

1
PFADD KEY ELEMENT [ELEMENT]

​

PFCOUNT

1
PFCOUNT KEY [KEY]

​

PFMERGE

1
PFMERGE DESTKEY SOURCEKEY [KEY]

​

​

⑧Geo#

​ PS：GEO类型是ZSET的子类，可以用ZSET的API

GEOADD

1
GEOADD KEY LONGITUDE LATITUDE VALUE [LONGITUDE LATITUDE VALUE]

​

GEPPOS

1
GEOPOS KEY VALUE

​

GEODIST

1
GEODIST KEY value value [m|km|ft|mi]

​

GEOHASH

1
GEOHASH KEY VALUE [VALUE]

​

GEORADIUS

1
GEORADIUS KEY LONGTITUDE LATITUDE RADIUS [m/km/ft/mi] [WITHDIST] [WITHCOORD] [WITHHASH] [COUNT N] [DESC]

​

GEORADIUSBYMEMBER

1
GEORADIUS KEY VALUE RADIUS [m/km/ft/mi] [WITHDIST] [WITHCOORD] [WITHHASH] [COUNT N] [DESC]

​

⑨Stream#

说白了就是Redis自己的消息中间件，很可惜在正式学习Redis-Stream之前我还没学过消息队列。。。 这代表我注定不能从这次学习中获得什么触类旁通的经历，唉，烦死了。有种立刻投身去学习Kafka的冲动。这次就当是给学MQ打基础了

​

​ Touch => ​ 在Stream出现前，Redis实现MQ的两种方案=>

​ ①:使用List实现MQ，是很典型的异步队列。将需要延后处理的任务对象插入Redis队列中，随后另一个线程实现轮 询处理，不过这也就注定了这种MQ实现更适合一对一的形式，一对多的业务就显得力不从心了。 ​

​ ②使用Redis提供的[Pub/Sub](#五.Redis Pub/Sub)，也就是发布订阅。不过Pub/Sub也有一些缺点，比如无法实现持久化，如果有网络波动、Redis服务器寄了这些情况，尚在MQ中的消息甚至会被遗弃掉。也没有Ack机制来保证数据的可靠性，如果光有生产者，没有消费者，还会出现所有消息都被舍弃这种逆天的情况。

综上，在Redis5.0之前，简单的MQ业务用List实现，稍微复杂一些的用Pub/Sub实现，但一直槽点满满。终于在Redis5.0，官方发布了新的数据结构——Stream。它诞生的目的就是为了实现Redis本体的MQ功能，而不是再去依靠例如Kafka、Rabbit、Rocket这些中间件。

所以，Steam几乎能干所有其他MQ中间件的事，例如实现MQ、消息持久化、自动生成全局唯一id、支持ack确认模式、支持消费组模式等等。

Structure Of Stream (官方)=>

MessageContent ：消息内容 ConsumerGroup：消费组，联想到集群订阅 LastDeliveredId：游标，每个消费组都有，该消费组的消费组只要读取了消息，就会使游标移动，以辨识哪条消息是最新的消息。 Consumer：消费者 PendingIds：实现了Ack的玩意，会记录已读未确认的消息。官网管它叫PEL（Pending Entries List），作用主要就是确保客户端至少消费了消息一次，而不是放任消息在网络运输中被丢失。

生产方

XADD

1
XADD KEY [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|id field value [field value]

​

XRANGE

1
XRANGE KEY start end [COUNT count]

​

XREVRANGE

1
XREVRANGE KEY start end [COUNT count]

​

XDEL

1
XDEL KEY ID

​

XLEN

1
XLEN KEY

​

XTRIM

1
XTRIM KEY MAXLEN|MINID N [LIMIT COUNT]

​

XREAD

1
XREAD [COUNT count] [BLOCK milliseconds] STREAMS KEY [KEY...] ID [ID...]

​

TIPS：学到这真心觉得像JAVA里的阻塞队列，这就是替身使者之间的吸引力吗

消费方

XGROUP CREATE

1
XGROUP CREATE KEY GROUPNAME id

​

XREADGROUP GROUP

1
XREADGROUP GROUP GROUPNAME CONSUMER [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS KEY [KEY...] > ID [ID...]

​

XPENDING

1
XPENDING KEY GROUPNAME

​

XACK

1
XACK KEY GROUPNAME ID

​

XINFO

1
XINFO STREAM|CONSUMER|GROUP VALUE

​

⑩BitField#

BITFIELD

1
BITFIELD KEY [] [GET TYPE OFFSET]
2
BITFIELD KEY [] [SET TYPE OFFSET VALUE]
3
BITFIELD KEY [] [INCRBY TYPE OFFSET INCREMENT]
4
//[] = [BITFIELD KEY OVERFLOW [WRAP|SAT|FAIL]]

​

RDB#

干的事儿一句话说明白：在指定的时间间隔后，将内存中的数据集快照写进硬盘。也就是上面那张图里的snapshot。恢复的时候再从硬盘写进内存，也就是上面的loads persisted data。值得注意的是，RDB执行的是全量快照，也就说每次写入都会把所有的数据都写入硬盘的一个dump.rdb文件。

在Redis7.0之后，指出手动触发和自动触发两种形式

AOF#

上面提到了RDB实现数据持久化的形式是生成RDB文件，AOF其实也是，生成了一个叫appendonly.aof的文件，但AOF生成的并不是二进制的数据本身，它生成的是一个日志，这个日志记录了Redis执行过的每一条写指令，并且日志只允许追加不允许修改。Redis重启时，就读取这个日志文件来进行数据库的重建。

Tips：AOF默认是不开启的，要去Config文件中修改appendonly yes AOF保存流程如下↓

RDB&AOF#

混合使用两种方式，在恢复时，AOF优先级大于RDB。RDB能够在指定的时间轮转片内对数据进行快照存储，AOF可以记录每次对服务器写得操作，服务器重启时会重新执行这些操作来恢复原始的数据，AOF命令以Redis协议来追加保存每次写的操作。先使用RDB进行快照存储，然后使用AOF持久化记录所有的写操作，当重写策略满足或手动触发重写的时候，将最新的数据存储为新的RDB记录。这样的话，重启服务的时候会从RDB和AOF两部分恢复数据，既保证了数据完整性，又提高了恢复数据的性能。

图示：

纯缓存模式#

同时关闭RDB和AOF，让Redis仅做缓存用

1
save ""
2
appendonly no

同数据库事务间的异同#

<1> 首先，Redis的事务没有单独的隔离操作，仅仅能保证事务内的操作不会连续不断的，因为Redis的指令操作时单线程架构的，所以在本次事务完成前不会执行其他客户端请求的指令。 <2> Redis的事务是没有隔离级别的，因为事务在被执行前任何指令都不会被实际执行，也就没有所谓的”事务内的更新对事务内的查询透明，对事务外的查询不透明”这码事。 <3> 值得注意的是，Redis的事务不具有原子性，Mysql的原子性可以保证事务内的sql要么全成功，要么全失败，保证数据的安全性。但是Redis有好几种可能，可以让事务内成功的成功，失败的失败，或者全成功全失败…好多呢，按下不表，稍后再议。 <4> 最后，Redis的事务具有排他性，一个事务结束前，不允许任何事务外的指令执行，相当于这个事务就是个队列，里面塞满了命令，这个队列一旦开始执行，就令线程阻塞了，直到这个队列执行完成为止。

实操#

1
MULTI => 开启事务
2
EXEC => 执行事务
3
DISCARD => 抛弃事务
4
WATCH => 监测事务
5
UNWATCH => 放弃监测

​ 总的来说，Redis事务的执行情况一般有五种 ​ ①全成功 => 没有任何错误异常，正常执行下来 ​ ②手动抛弃 => 事务块最后写入DISCARD表示事务被抛弃 ​ ③全失败 => 事务中存在语法错误或检测出异常，Redis不予执行 ​ ④有成功有失败 => Redis没有检测出错误，但确实有错误，前面正确的部分予以执行，出错处以后被抛弃 ​ ⑤被监控 => ​ 就是加锁，Redis采用的是乐观锁，假设每次取数据都不会有修改操作，所以不上锁，但是在更新完成时会检测一下改没改(Tips：提交版本必须高于记录当前版本才能执行更新)

CAS(Check-And-Set)#

就是Redis对乐观锁的一种实现方式。

1
WATCH KEY
2
UNWATCH KEY

在监测期间，如果这个key在外面被修改了，那事务直接执行失败。

与事务间的异同#

事务具有原子性，管道没有 管道是一次性将多条指令发送到服务器，事务是一条一条发，事务在接收到EXEC后才会执行，管道不是。 执行事务时会阻塞事务外的指令，管道不会，欢迎加塞

与原生批处理操作间的异同#

原生批处理操作是原子性的，pipeline不是 原生批处理操作一次只能执行一种命令，pipeline可以执行多种 原生批处理操作主要靠服务端实现，pipeline要服务端和客户端一起实现

Tips：pipeline缓冲的指令只是会一次执行，不保证原子性，如果执行中指令发生异常，将会继续执行后续的指令。 使用pipeline组装的指令个数不能太多，不然数据量过大客户端阻塞的时间可能过久，同时服务端此时也被迫回复组装的队列，占用的内存会过高。

实操#

1
SUBSCRIBE CHANNEL [CHANNEL...]
2
PUBLISH CHANNEL MESSAGE
3
PSUBSCRIBE PATTERN [PATTERN]
4
PUBSUB SUBCOMMAND [ARGS]
5
UNSUBSCRIBE CHANNEL [CHANNEL...]
6
PUNSUBSCRIBE PATTERN [PATTERN...]

​ 常用指令就这些，从上往下分别是： ​ 订阅某个频道，可一次订阅多个 ​ 向某个频道发布内容 ​ 按照某个模式订阅频道，包括 * ？ 等通配符 ​ 查看订阅和发布系统的状态，subcommand包括 channels（返回活跃频道列表）、numsub （某个频道有多少订阅者）、numpat（只统计拥有模式订阅的返回到服务器的订阅唯一模式数量）等。 ​ 取消订阅的两种形式，按照指定频道名取消订阅，按照指定模式取消订阅 ​

总结#

​ pubsub可以实现消息中间件mq的功能，通过发布订阅来实现消息的引导和分流，不过专业的事情最好交给专业的人来干，还是让kafka、rmq去干比较好。 pubsub也有一些缺点，比如发布的消息在redis中不能实现持久化，需要先订阅再发布才能收到，否则发布的消息将会被直接遗弃。 而且消息是即发即失的，pubsub没有ack机制，无法保证消息的消费是否成功。 这就让pubsub变得食之无味，弃之可惜。在redis5.0之后出现了Stream之后，就解决了这些情况。但还是不推荐使用Stream，还是用专业的MQ中间件比较好。

操作一览#

1
INFO REPLICATION
2
REPLICAOF IP PORT 写入配置文件
3
SLAVEOF IP PORT
4
SLAVEOF NO ONE

​ slaveof ip port这种操作，如果当前从机已有主机，会转而连接给定的新主机，成为那个主机的从机。 ​ salveof no one会取消当前Redis从机的身份，变为主机。

Touches（常见Q&A）#

Touch=>配置和手动指向两者间有何区别？ 配置文件的形式持久稳定，命令当次有效。当使用命令去指定master之后，shutdown再重启之后还是会参考配置文件来指定当前Redis的master，并不会自动修改配置文件。

Touch=>主从之间的权限分配 只有主机才能写，从机执行写指令会报错。

Touch=>从机转向新主机后数据如何变化？ 会清除掉以前的旧数据，转而录入新主机的数据

主从复制原理和工作流程一览#

①Slave启动，进行同步

Slave启动成功连接到Master之后，会发送一个SYNC命令。旋即进行数据同步 Slave首次连接到一个全新的未连接过的Master之后，会将自身全部的数据清除，转而存储Master的全量快照。

②首次连接，全量复制

Master节点收到SYNC命令之后会自动在后台保存快照（RDB持久化），同时收集所有的修改命令，在RDB执行完RDB持久化后，会将RDB快照文件和修改命令一起发送给Slave，以完成一次全量同步。 而Slave接受到数据之后，会将数据存盘并加载到内存中，这样一次主从复制就完成了。

③心跳持续，保持通信

Master会按照配置文件中的指定信息来以固定时间轮转片为间隔来向Slave进行同步。

1
repl-ping-replica-period 10

④进入平稳，增量复制

Master会将从客户端接收到的所有修改操作依次自动地传给Slave们，以保证信息一致性

⑤从机下线，重连续传

Slave在连接上Master之后，会参考两者backlog中的offset来进行Slave的数据增加，类似于断点续传。所以Slave宕机与否都不影响主从间的数据一致性，Slave会自己同步的。

缺点#

由于所有的写操作都是先在Master上操作，然后同步更新到Slave上，所以从Master同步到Slave机器有一定的延迟，当系统很繁忙的时候，延迟问题会更加严重，Slave机器数量的增加也会使这个问题更加严重。 Master挂了会怎么样？Slave可没有什么再找个主人的能力，所以需要有监控，这也就引出了下一章哨兵。

​

​

配置文件具体参数设置#

1
bind //服务器监听地址，用于客户端连接，默认本机地址
2
daemonize //是否以后台daemon的方式运行
3
protected-mode //是否开启安全保护模式
4
port //端口
5
logfile //日志文件路径
6
pidfile //pid文件路径
7
dir //工作目录
8
*sentinel monitor <master-name> <ip> <redis-port> <quorum>
9
*sentinel auth-pass <master-name> <password> //设置连接master的密码

​

​ 其他设置（不太重要） ​

1
sentinel down-after-milliseconds <master-name> <milliseconds>：
2
//指定多少毫秒之后，主节点没有应答哨兵，此时哨兵主观上认为主节点下线
3

4
sentinel parallel-syncs <master-name> <nums>：
5
//表示允许并行同步的slave个数，当Master挂了后，哨兵会选出新的Master，此时，剩余的slave会向新的master发起同步数据
6

7
sentinel failover-timeout <master-name> <milliseconds>：
8
//故障转移的超时时间，进行故障转移时，如果超过设置的毫秒，表示故障转移失败
9

10
sentinel notification-script <master-name> <script-path>
11
//配置当某一事件发生时所需要执行的脚本
12

13
sentinel client-reconfig-script <master-name> <script-path>：
14
//客户端重新配置主节点参数脚本

​ 加入哨兵机制之后，如果被监视的主机down掉了，那么接下来会在剩下的从机中选举出一个新主机，让他上位，此时从机数据完整一致。 这种情况下，原来的主机重新连接进来，那么它将会作为现在的主机的从机，进入机组。

​

Tips：值得注意的是，原有的config文件，在哨兵运行的时候，会被动态修改。Master-Slave切换后，master_redis.conf、slave_redis.conf和sentinel.conf的内容都会发生改变，即master_redis.conf中会多出一行slaveof的配置，sentinel.conf的监控对象也会随之改变。

小注：一个哨兵可以同时监视多个master，一行一个续写就行。

运行流程及选举原理#

当一个主从配置中的master down掉之后，sentinel就可以选举出一个新的master来接替原来的master的工作，主从配置中的其他rredis服务器也可以自动指向新立得master来同步数据。一般sentinel采用奇数台，防止某一台sentinel无法连接到master导致误切换。

Sentinel正常运行中…

SDown主观下线：就是单个Sentinel在同主机的心跳检测中，主观地认为主机挂掉了，从Sentinel的角度看，如果向主机发送了Ping心跳，在一定时间内如果没有收到合法回复，那就达到了SDwon的条件 Tips：在Sentinel的配置文件中可以通过配置down-after-milliseconds来设置主观下线的时间长度。

ODown客观下线：就是上面提到过的，不多赘述了。由Sentinel们投票来决定主机是否下线。

选举环节（Raft算法）：当主机被客观投票下线后，各个Sentinel就会开始进行协商，使用Raft算法从哨兵中选出一个leader，然后由这个leader来做failover

​

​ 替换流程

master替换：由上一个环节选出的leader来做新master的上位。这主要有三个环节

①选举新master：在所有的slave中，优先挑选conf文件中slave-priority或replica-priority最高的节点。（数字越小优先级越高），复制replication offset最大的从节点，按照Run id最小的从节点（ASCII码排序，字典顺序） Tips：有点像DFS，但也就有点像。

②Slave转拜：执行slaveof no one 来让选出来的从节点成为新的主节点，再通过slaveof命令来让其他节点成为它的从节点。Sentinel leader会对选举出的master执行slaveof no one来成为master节点。最后，Sentinel leader会让其他slave转拜这个节点为master

③旧master认主：原先的master重连时，会让它也转拜选出的新master为主机。Sentinel leader会让原来的master降级为slave进行正常工作。

Tips#

​ ①哨兵节点的数量应为多个，哨兵本身应该集群，保证High Available ​ ②哨兵节点的个数应为奇数个 ​ ③各个哨兵节点的配置应一致 ​ ④如果哨兵节点部署在Docker中，应注意端口的正确映射 ​ ⑤哨兵集群+主从复制，并不能保证数据完全无丢失，所以才会出现集群机制

集群算法-分片-插槽Slot#

Redis集群没有使用一致性Hash，而是引入了哈希槽的概念 Redis集群拥有16384个哈希槽，每个Key都通过CRC16校验后对16384取模来决定放在哪个槽位中。

分片

图示： ![slot ./slot.png)

优势

Slot槽位映射#

业内一般有三种处理方式：Hash取余分区、一致性哈希算法分区、哈希槽分区 各有利弊，但综合下来使用哈希槽最好，下面细说

集群环境#

1
redis-cli -a passpord --cluster create --cluster-replicas N ip:port ip:port*N [ip:port ip:port] (主从)

1
info replication
2
cluster info
3
cluster nodes

1
redis-cli -a password -c

1
CLUSTER FAILOVER //在主机中

1
redis-cli -a password --cluster add-node ip:port ip:port

1
redis-cli -a password --cluster del-node ip:port id

1
redis-cli -a password --cluster reshard ip:port

总结#

①集群下不在同一个slot中的kv不能使用批处理操作获取，需要使用通识占位符。

1
mset k1{z} z1 k2{z} z2 k3{z} z3
2
mget k1{z} k2{z} k3{z}

第一个命令表示将不同slot中的k1 k2 k3打包进z这个组，再从z中拿出。

②哈希槽分配中的CRC16在哪儿执行？

在Cluster.c这个原文件中，有一个叫做 KeyHashSlot的方法，由他执行调用。

③常用命令

1
cluster-require-full-coverage

​

1
cluster countkeysinslot num

​ 查看指定编号的槽位上是否有数据，1被占用，0没被占用

1
cluster keyslot key

​ 上面介绍过了，查看某个键应该在哪个槽位上

Jedis#

①引入依赖

1
        <dependency>
2
            <groupId>redis.clients</groupId>
3
            <artifactId>jedis</artifactId>
4
            <version>4.3.1</version>
5
        </dependency>

②创建Jedis实例进行操作

1
Jedis jedis = new Jedis("host",port);
2
jedis.auth("password")
3
jedis.操作

具体操作详见 此处

Lettuce#

①引入依赖

1
        <dependency>
2
            <groupId>io.lettuce</groupId>
3
            <artifactId>lettuce-core</artifactId>
4
            <version>6.2.1.RELEASE</version>
5
        </dependency>

②使用流式编程创建Lettuce实例进行操作

1
        RedisURI uri = RedisURI.builder()
2
            .withHost("IP")
3
            .withPort(port)
4
                 .withAuthentication("default", "password")
5
                 .build();
6
        RedisClient redisClient = RedisClient.create(uri);
7
        StatefulRedisConnection con = redisClient.connect();
8
        RedisCommands redisCommand = con.sync();
9
    redisCommand.操作
10
        con.close();

具体操作详见 此处

二者区别

RedisTemplate#

①引入依赖

1
     <!--SpringBoot与Redis整合依赖-->
2
    <dependency>
3
            <groupId>org.springframework.boot</groupId>
4
            <artifactId>spring-boot-starter-data-redis</artifactId>
5
         </dependency>
6
         <dependency>
7
            <groupId>org.apache.commons</groupId>
8
            <artifactId>commons-pool2</artifactId>
9
         </dependency>
10
         <!--swagger2-->
11
         <dependency>
12
             <groupId>io.springfox</groupId>
13
             <artifactId>springfox-swagger2</artifactId>
14
             <version>2.9.2</version>
15
         </dependency>
16
         <dependency>
17
             <groupId>io.springfox</groupId>
18
             <artifactId>springfox-swagger-ui</artifactId>
19
             <version>2.9.2</version>
20
         </dependency>

②编辑配置文件

1
# ========================logging=====================
2
logging.level.root=info
3
logging.level.com.proj.redis=info
4
logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger- %msg%n
5

6
logging.file.name=D:/mylogs2023/redis7_study.log
7
logging.pattern.file=%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger- %msg%n
8

9
# ========================swagger=====================
10
spring.swagger2.enabled=true
11
#在springboot2.6.X结合swagger2.9.X会提示documentationPluginsBootstrapper空指针异常，
12
#原因是在springboot2.6.X中将SpringMVC默认路径匹配策略从AntPathMatcher更改为PathPatternParser，
13
# 导致出错，解决办法是matching-strategy切换回之前ant_path_matcher
14
spring.mvc.pathmatch.matching-strategy=ant_path_matcher
15

16
# ========================redis单机=====================
17
spring.redis.database=0
18
spring.redis.host= #ip
19
spring.redis.port= #port
20
spring.redis.password= #password
21
spring.redis.lettuce.pool.max-active=8
22
spring.redis.lettuce.pool.max-wait=-1ms
23
spring.redis.lettuce.pool.max-idle=8
24
spring.redis.lettuce.pool.min-idle=0

③编写业务

1
 @Resource
2
    private RedisTemplate redisTemplate;
3

4
 public void biz(){
5
     redisTemplate.操作
6
 }

序列化问题#

Q：为什么会出现序列化问题？

键(key)和值(value)都是通过Spring提供的Serializer序列化到数据库的。 RedisTemplate默认使用的是JdkSerializationRedisSerializerstringRedisTemplate默认使用的是stringRedisSerializer.KEY 被序列化成这样，线上通过 KEY 去查询对应的 VALUE非常不方便

解决方案一

使用StringRedisTemplate来替代RedisTemplate，这个类内部使用了这样的构造器↓

1
    public StringRedisTemplate() {
2
        this.setKeySerializer(RedisSerializer.string());
3
        this.setValueSerializer(RedisSerializer.string());
4
        this.setHashKeySerializer(RedisSerializer.string());
5
        this.setHashValueSerializer(RedisSerializer.string());
6
    }

​ 改变了原先的编码格式，这样在与redis的交互中，起码前端页面中获取的数据不会是乱码，但还是有不足，在Redis内部存储的数据，尤其是value的值，还会是出现乱码，在传到客户端，使用中文编码才会得到解决，下面介绍更好地处理方式。

解决方案二

先前分析过出现序列化问题的罪魁祸首，就是RedisTemplate内部设置的serializer不支持中文编码，但是进入RedisTemplate的源码查看相关部分，发现是这样的

那么解决方案就应然而出了，只要修改成咱们需要的Serializer就行了

1
@Configuration
2
public class RedisConfig {
3
    @Bean
4
    public RedisTemplate<String,Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory){
5
        RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();
6
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
7
        //设置key序列化方式为String
8
        redisTemplate.setKeySerializer(new StringRedisSerializer());
9
        //设置value的序列化方式为Json，使用GenericJackson2JsonRedisSerializer来替换原有序列化器
10
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
11

12
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
13
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
14

15
        redisTemplate.afterPropertiesSet();
16

17
        return redisTemplate;
18
    }
19
}

​ 完成以上配置后，在业务中注入的就是配置类中引入的bean，这样就能有效解决序列化问题了。

Tips：在Redis本机上查看交互的数据时，登录要键入—raw参数，以查看中文数据。

集群连接#

①改写yaml

1
spring.redis.cluster.nodes= ......

加入这行配置，表示集群中的节点ip

②通过微服务访问集群

直接使用swagger跟平常一样该怎么测怎么测就行。

Touch：如果在服务端访问redis服务器时，Master宕了，按照正常流程，应该遵从哨兵机制那套，自动实行选举，从机上位。但是微服务可没那么只能，它会跟个傻子似的一直向那台寄了的主机发请求，这严重影响了我们服务的高可用，该怎么解决呢？

解决方案一 关闭Lettuce使用Jedis，这样就能解决问题了，但是不推荐。Jedis本身就是个线程不安全的技术，这样做属实是丢了西瓜捡了芝麻了。

解决方案二 重写Lettuce的LettuceConnectionFactory源码，这更不推荐，需要技术力极高不说，改错了还背大锅。而且版本更新还要做修改，不是一劳永逸的活。

解决方案三 刷新节点集群拓扑动态感应，这是最推荐的方式，官网也推荐这么干 需要修改改写下yaml文件↓

1
#支持集群拓扑动态感应刷新,自适应拓扑刷新是否使用所有可用的更新，默认false关闭
2
spring.redis.lettuce.cluster.refresh.adaptive=true
3
#定时刷新
4
spring.redis.lettuce.cluster.refresh.period=2000
5
spring.redis.cluster.nodes=......

Redis支持多线程吗？#

一图了然

准确来说，是从Redis6.0开始，才石锤支持了多线程。

在Redis正常工作流程中，一般的命令工作线程是单线程的，但是在一些功能，例如RDB、AOF、异步删除，这些是多线程的，所以这样说，Redis命令工作线程是单线程的，整个Redis是多线程的。

Redis4.0前为什么一直使用单线程？#

①使用单线程模型让Redis的开发和维护更简单，因为单线程模型方便研发和调试

②即使使用单线程也能并发处理多客户端的请求，主要使用的是IO多路复用和非阻塞式IO

③对于Redis来说，CPU并不是影响它性能的瓶颈，内存和网络才是

Redis为什么要引入多线程？#

首先从硬件方面说，摩尔定律大家都知道，硬件的发展速度也快的一批，现在多核的CPU早就普及了，即使CPU对Redis性能的限制不大，但为了适应新的环境，还是对Redis进行的修改，只为了让它更快。

其次，单线程有很多缺陷，打个比方

这就是redis3.x单线程时代最经典的故障，大key删除的头疼问题。

所以在后续版本中，Redis慢慢引入了一些异步操作，例如 unlink key 、flushdb async、 flushall asnyc… 它们就是把具体操作流程交给了开辟了子进程(bio进程)，做了异步操作。

lazy free的本质就是把某些cost(主要时间复制度，占用主线程cpu时间片)较高删除操作。 从redis主线程剥离让bio子线程来处理，极大地减少主线阻塞时间。从而减少删除导致性能和稳定性问题。

Redis为什么那么快？#

①基于内存操作

Redis的所有数据都存于内存中，所有的运算都是内存级别的，所以从硬件角度它就该比其他数据库快

②数据结构简单

Redis的数据结构都是专门设计过的，而这些简单的数据结构的查找和操作的时间复杂度大部分都是O(1)，所以性能很高

③多路复用和非阻塞式IO

Redis使用的是I/O多路复用来监听多个socket连接客户端， 这样就可以使用一个线程来处理多个请求，减少了线程切换带来的性能占用，也避免了IO阻塞

④避免上下文切换

因为是单线程模型，所以能避免不必要的上下文切换和多线程竞争，省去了多线程切换带来的性能消耗、时间损失，而且单线程不会有死锁产生

简述及多线程协作流程#

​ Before=>

首先，IO的读写本身是堵塞的，Redis接收到一个Socket，当这个Socket中有数据时，Redis会通过调用将这些数据拷贝到目态空间，再交给Redis处理，所以，在这个过程中，Socket中的数据越大，阻塞的时间就越长，这是基于单线程模式实现的，有很大弊端。

​

After=>

从Redis6开始，就新增了多线程的功能来提高 I/O 的读写性能，他的主要实现思路是将主线程的 IO 读写任务拆分给一组独立的线程去执行，这样就可以使多个 socket 的读写可以并行化了，采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求（尽量减少网络IO的时间消耗），将最耗时的Socket的读取、请求解析、写入单独外包出去，剩下的命令执行仍然由主线程串行执行并和内存的数据交互。

具体阶段如下↓

阶段一：服务端和客户端建立Socket连接，并分配处理线程 首先，主线程负责接收建立连接请求，当有客户请求和实例建立Socket连接时，主线程就会创建和客户端的连接，并把Socket放入全局等待队列中，紧接着，主线程通过轮询方式把Socket连接分配给IO线程。

阶段二：IO线程读取并解析请求 主线程一旦将Socket分配给IO线程，就会进入阻塞状态，等待IO线程完成客户端请求的读取和解析。因为有多个IO线程在并行处理，所以这个过程很快就能完成

阶段三：主线程执行请求操作 等到IO线程解析完请求，主线程还是会以单线程的方式执行这些命令请求。

阶段四：IO线程回写Socket和主线程清空全局队列 当主线程执行完操作后，会把需要的结果写入缓冲区，然后主线程会阻塞等待IO线程，把这些结果回写到Socket中，并返回给客户端。和IO线程的读取和解析请求一样，IO线程回写到Socket时，也是多个线程在并发执行，所以回写Socket的速度很快。等到IO线程回写完毕，主线程就会清空全局等待队列，等待客户端的后续请求。

Unix网络编程中的五种IO模型#

①Blocking IO - 阻塞式IO

②NoneBlocking IO - 非阻塞式IO

③IO Multiplexing - IO多路复用

先介绍一个概念=>FileDescriptor，文件描述符，简称FD或者句柄，在Linux、Unix架构下的操作系统，所有文件都拥有此属性，它表达了文件的全局唯一性。当主客户端发生交互时(文件)，一般都是通过句柄发送和接收验证来此文件的状态。

那么IO多路复用是什么呢？

• IO多路复用是一种同步的IO模型，它实现一个线程监视多个文件句柄，一旦某个文件句柄就绪，就通知对应应用程序进行响应的读写操作，没有文件句柄就绪时，就阻塞应用程序，然后释放CPU资源。

• I/O ：网络IO，尤其在操作系统层面指的是目态管态间的读写操作

• 多路：多个客户端连接(连接就是套接字描述符，即Socket或Channel)

• 复用：复用一个或几个线程

• IO多路复用：一个或一组线程处理多个TCP连接，使用单进程就能实现同时处理多个客户端的连接，无需创建或维护过多的进程/线程

总结：干的就是一件事，使一个服务端进程能同时处理多个套接字描述符

实现方式

Select => Poll => Epoll

将用户socket对应的文件描述符(FileDescriptor)注册进epoll，然后epoll帮你监听哪些socket上有消息到达，这样就避免了大量的无用操作。此时的socket应该采用非阻塞模式。这样，整个过程只在调用select、poll、epoll这些调用的时候才会阻塞，收发客户消息是不会阻塞的，整个进程或者线程就被充分利用起来，这就是事件驱动，所谓的reactor反应模式。

在单个线程通过记录跟踪每一个Sockek(I/O流)的状态来同时管理多个I/O流. 一个服务端进程可以同时处理多个套接字描述符。

目的是尽量多的提高服务器的吞吐能力。

大家都用过nginx，nginx使用epoll接收请求，ngnix会有很多链接进来， epoll会把他们都监视起来，然后像拨开关一样，谁有数据就拨向谁，然后调用相应的代码处理。redis类似同理，这就是IO多路复用原理，有请求就响应，没请求不打扰。

​

④Signal Driven IO - 信号驱动IO

⑤Asynchronous IO - 异步IO

Tips：在Redis6.0/7.0后，多线程机制默认是关闭的，如若想开启，则需要修改配置文件

1
io-thread-do-reads yes
2
io-threads N

1.设置io-thread-do-reads配置项为yes，表示启动多线程。

2。设置线程个数。关于线程数的设置，官方的建议是如果为 4 核的 CPU，建议线程数设置为 2 或 3，如果为 8 核 CPU 建议线程数设置为 6，线程数一定要小于机器核数，线程数并不是越大越好。

如何避免#

简单来说禁用掉这几个危险命令就行了 可以去配置文件中找对应设置 /rename-command

1
rename-command commandName ""

commandName为你想禁用的命令名

不用Keys * 用什么？#

SCAN

和mySql的Limit很像，但不一样

1
SCAN cursor [MATCH pattern] [COUNT count] [Type type]

• cursor 游标
• pattern 匹配模式
• count 返回条数，默认10

Tips：Scan的遍历顺序十分特别，它不是从第一维数组的第零位一直遍历到末尾，而是采用了高位加法来遍历，这是为了字典的扩容缩容时避免槽位的遍历重复和遗漏

多大算Big Key？#

​ 业内的说法如下

String类型在10KB以内，Hash、List、Set、ZSet元素个数不超过5000个。

​ Tips：非字符串的Big Key不要用Del删除，要用HScan、SScan、ZScan渐进式删除。同时还要注意防止Big Key过期时间自动删除问题(例如一个200万的ZSet设置了1h过期，会触发Del操作，造成阻塞，而且该操作不会出现在慢查询中(LateNCY可查))

Big Key 的危害#

• 内存不均，集群迁移困难
• 超时删除，大Key删除困难
• 网络流量阻塞

Big Key 的高频出现场景

社交类：微博粉丝交互

汇总统计：大公司的报表，经过逐年累月的积累

Big Key 排查#

①redis-cli --bigkeys

给出每种数据结构Top 1 bigkey，同时给出每种数据类型的键值个数+平均大小

②MEMORY USAGE KEY

给出指定KEY的键值的字节数，对于嵌套数据类型，可以选用参数 SAMPLES，其中count表示抽样的元素个数，默认是5，当需要抽样所有元素时，使用Sample 0

Big Key 删除#

• String：Del，特别大的用unlink

• Hash：使用HScan每次获取少量Field-Value，再用HDel删除每个Field，最后Del这个Key

• List：使用Ltrim去渐进性删除，直到全部删除完成，最终Del这个Key

• Set：使用SScan每次获取部分元素，最后使用SRem去删除这个Key

• ZSet：使用ZScan每次获取部分元素，再使用ZREMRANGEBYRANK删除每个元素，最后再Del这个Key

Big Key 生产调优#

​ 先说两句，Redis天生拥有两种删除，阻塞和非阻塞式的，例如Del和Unlink，前者是对象的阻塞删除，意味着服务器将停止处理新的命令，以便以同步方式回收与对象关联的所有内存，如果删除的键与一个小对象相关联，那么执行DEL没命令所需要的时间会非常短，时间复杂度近乎O(1)/O(logn)，但是如果键包含数百万个元素的聚合值关联，则服务器会阻塞很久。 ​ 所以，Redis才提供了后者这种非阻塞式原语，以便在后台回收内存。这些命令在恒定时间内执行，另一个线程来以尽量短的时间来释放后台中的对象。

​ Lazy Freeing

​ 基于上述内容，我们不难发现在生产环境中，这种非阻塞性的删除才更迎合我们的需求，在Redis中也有这种策略，被称为惰性释放(Lazy Freeing)

• 配置RedisConf文件

1
lazyfree-lazy-server yes
2
replica-lazy-flush yes
3
lazy-free-user-del yes

从上往下分别是，服务器del操作、flush操作、客户端del操作，全部改为yes，这样就允许了这些操作的惰性释放权限

缓存双写#

什么是缓存双写？一图了然

• ①Redis里有，直接返回Redis中的数据
• ②redis无，mysql有，从mysql中查到数据返回
• ③完成第二步后，将mysql中的数据写回redis，完成数据一致性

总结一下，简单来说就是如果Redis中有数据，就需要和数据库中的数据值相同，如果Redis中无数据，数据库中的值要是最新值，就把它写回Redis。

缓存分类#

①只读缓存 没什么好说的，一般业务中用的很少，得是多特殊的业务，才只用读不用写？还有这好事？

②读写缓存 一般都是这种缓存，一般处理这种缓存有两种策略

(1)同步直写策略 写数据库后也同步写Redis，缓存和数据库中的数据一致 对于读写缓存，要想保证缓存与数据库数据一致性，就用这种策略

(2)异步缓写策略 正常业务中，mysql的数据变动了，但是可以在业务上允许一段时间后再作用在Redis，异常情况出现时，不得不将失败的动作重新修补，可能需要借用kafka或者rabbitMQ这些中间件的帮助来实现重试重写

双检加锁策略#

按照一般的业务逻辑，在查询Redis无果后，都会直接去Mysql中去查询，这很正常，但如果是在高并发场景下，有可能会出现一瞬间大量的线程压在Mysql上，Mysql可没有Redis这么强的性能，可能会出现缓存穿透的情况，我们要避免这种可能

可以使用双检加锁策略↓

第一次从Redis查询赋值后，给当前线程加锁，再从Redis查一次，若无果，再从Redis中查询，此时其他的线程走到这一步拿不到锁就等着，等第一个线程查询到了数据，然后做缓存，后面的线程进来发现已经有缓存了，就直接走缓存。

1
  public User findUserById(Integer id)
2
    {
3
        User user = null;
4
        String key = CACHE_KEY_USER+id;
5

6
        //1 先从redis里面查询，如果有直接返回结果，如果没有再去查询mysql，
7
        // 第1次查询redis，加锁前
8
        user = (User) redisTemplate.opsForValue().get(key);
9
        if(user == null) {
10
            //2 大厂用，对于高QPS的优化，进来就先加锁，保证一个请求操作，让外面的redis等待一下，避免击穿mysql
11
            synchronized (UserService.class){
12
                //第2次查询redis，加锁后
13
                user = (User) redisTemplate.opsForValue().get(key);
14
                //3 二次查redis还是null，可以去查mysql了(mysql默认有数据)
15
                if (user == null) {
16
                    //4 查询mysql拿数据(mysql默认有数据)
17
                    user = userMapper.selectByPrimaryKey(id);
18
                    if (user == null) {
19
                        return null;
20
                    }else
21
                        //5 mysql里面有数据的，需要回写redis，完成数据一致性的同步工作
22
         redisTemplate.opsForValue().setIfAbsent(key,user,7L,TimeUnit.DAYS);
23
                }
24
            }
25
        }
26
        return user;
27
    }
28
}

缓存与数据库一致性更新策略#

一切为了最终一致性

​ 要不直接停机，然后做数据一致性？

想法很好，下次别想了，很多公司业务根本没有时间让你停机，难道为了做数据一致性就让客户干等着？必须线上做，而且要快

①先更新数据库，再更新缓存(不推荐)

会出现一些异常情况，例如更新数据库成功后，按理说要更新缓存，这时失败了，导致缓存和数据库数据不一致，这样就可能会读到缓存的脏数据。又或者两个线程一起执行操作 ，按照Socket的解析顺序应该是 A A B B，但真到了执行的时候，就有可能会变成 A B A B，这样也会导致数据库和缓存中的数据不一致。

②先更新缓存，再更新数据库(极不推荐)

在业务上，一般是拿mysql当底单数据库的，它应该具有数据的权威性 而且它也会出现①中的第二种情况

③先删除缓存，再更新数据库(不是很推荐)

乍一看挺好的，删除缓存再更新数据库，完成后再查询的时候，肯定会放行到Redis，然后正常回写呗，数据肯定一致啊。不过有两个很坑的地方，听我说哈。

①假设有这么一种情况，一个线程负责删除缓存，并更新数据库，Redis删完了，正更新Mysql呢，突然，又来个线程过来找Redis要数据，这时候Redis是空的啊，没命中肯定要来Mysql查了，这时候Mysql没更新完，给这个线程查到的如果是以前的数据，会发生什么？它会被回写到Redis里！太逆天了，合着第一个线程的活儿全白干了，写回缓存的数据是脏数据。

②就算你这个流程全走完了，Mysql更新完了，Redis删光了，但是我们要缓存的意义是什么？是为了替Mysql挡访问啊，这样的更新方式有个无法避免的弊端，那就是删除后的第一批查询必然是缓存击穿的，如果这时候遇上了高并发的场景，无数线程过来查数据，那么Sorry，全都去Mysql吧，我Redis里什么都没有。不过这不算是什么缺点，只是第一次会出现罢了。

解决方案

延时双删

1
public void deleteOrderData(Order order){
2
  try(Jedis j = RedisUtil.getJedis()){
3
    jedis.del(order.getId()+"");
4
    orderMapper.update(order);
5
    try{
6
      TimeUnit.SECONDS.sleep(2);
7
    }catch(InterruptedException e){
8
      e.printStackTrace();
9
    }
10
    jedis.del(order.getId()+"");
11
  }catch(Exception e){
12
    e.printStackTrace();
13
  }
14
}

Q&A

Q：如何确定阻塞时间？ A：在业务程序运行的时候，统计下线程读数据和写缓存的操作时间，自行评估自己的项目的读数据业务逻辑的耗时，以此为基础来进行估算。然后写数据的休眠时间则在读数据业务逻辑的耗时基础上加百毫秒即可。 或者开启一个后台监控程序watchDog，后面会讲

Q：这种同步淘汰策略，吞吐量降低怎么办？ A：简单，让第二次删除变成异步的，这样写请求就不会延后了，这样就可以增加吞吐量了

1
//针对第二次删除的修改
2
CompletableFuture.supplyAsync(()->{
3
  return jedis.del(order.getId()+"")
4
}).whenComplete((t,u)->{
5
    System.out.println("---t"+t)
6
    System.out.println("---t"+u)
7
}).exceptionally(e->{
8
    System.out.println("---e"+e.getMessage());
9
    return 44L;
10
}).get();

④先更新数据库，再删除缓存(先凑活用这个)

现在业内用的比较多，但还是有小问题，如果缓存删除失败或者来不及，那么这时候来的查询读到的就还是老数据。不过比起上面的策略来说，这种已经很优秀了，大厂基本用的都是这种方案。那么我们也提到过了，在项目运行中，双写是一定会出现纰漏的，我们只要保证最终一致性就行了，那如何做到呢？

微软云在官网提供了一个指导思想，基本和我们的方法一样。查看 阿里巴巴的canal也是类似的，这个中间件在binlog程序中起到了类似作用，通过订阅binlog日志来干这件事

解决方案

所以，像这种类分布式事务问题，解决方案只有一个，就是实现最终一致性

总结#

实战工作应用#

我们在做Redis和Mysql的双写一致性的时候，往往需要考虑性能，说人话，我们的需求是在底单数据库Mysql中的数据发生变化时，尽快地同步到缓存Redis中，一般的思路，无非就是用模拟心跳检测的方式去查看Mysql中的数据变动，然后动态修改Redis数据，但这太繁琐了，每次都要将库全查，性能消耗太大，所以下面我们引入Canal，作为Redis、Mysql双写一致性的最终排版解决方案

Canal

一个数据库中间件，干的事就是基于Mysql的数据库增量日志解析来做增量数据的订阅和消费，通过这件事，能实现的功能就很多了，包括数据库镜像、数据库实时备份、索引构建和实施维护(拆分异构索引、倒排索引)、业务cache刷新、带业务逻辑的增量数据处理

​ 官网地址

Q：Canal是怎么实现这些功能的呢？

和传统的模拟心跳检测不同，Canal是模拟了一台Mysql主机，以主从复制的方式作为从机和要检测的主机建立的联系，Mysql的主从复制流程大家都了解，就是从机通过读取主机中的binary log文件中的偏移量来判断数据是否发生变化，而做出修改。Canal干的也是这件事，就是通过向主机发送dump协议来获取binary log文件来解析数据变化，再通知后续的进程或者中间件来做出对应变化

Tips：

​ 如果要启用Canal，必须对Mysql主机做如下设置 ​ ①log-bin=mysql-bin #开启 binlog ​ ②binlog-format=ROW #选择 ROW 模式 ​ ③server_id=1 #配置MySQL replaction需要定义，不要和canal的 slaveId重复

​ 还需要在主机中加入canal的账号，并赋予对应的权限

​ 还有一点，需要对canal本身进行配置，在canal本地的instance.properties中做如下配置

​

1
canal.instance.master.address=YourIP
2
canal.instance.dbUsername=YourAccount
3
canal.instance.dbPassword=YourPassword

官网示例

​

1
public static void main(String args[]) {
2
    // 创建链接
3
    CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111),"example", "", "");
4
    int batchSize = 1000;
5
    int emptyCount = 0;
6
    try {
7
        connector.connect();
8
        connector.subscribe(".*\\..*");
9
        connector.rollback();
10
        int totalEmptyCount = 120;
11
        while (emptyCount < totalEmptyCount) {
12
            Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
13
            long batchId = message.getId();
14
            int size = message.getEntries().size();
15
            if (batchId == -1 || size == 0) {
16
                emptyCount++;
17
                System.out.println("empty count : " + emptyCount);
18
                try {
19
                    Thread.sleep(1000);
20
                } catch (InterruptedException e) {
21
                }
22
            } else {
23
                emptyCount = 0;
24
                // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
25
                printEntry(message.getEntries());
26
            }
27

28
            connector.ack(batchId); // 提交确认
29
            // connector.rollback(batchId); // 处理失败, 回滚数据
30
        }
31

32
        System.out.println("empty too many times, exit");
33
    } finally {
34
        connector.disconnect();
35
    }
36
}
37

38
private static void printEntry(List<Entry> entrys) {
39
    for (Entry entry : entrys) {
40
        if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
41
            continue;
42
        }
43

44
        RowChange rowChage = null;
45
        try {
46
            rowChage = RowChange.parseFrom(entry.getStoreValue());
47
        } catch (Exception e) {
48
            throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
49
                                       e);
50
        }
51

52
        EventType eventType = rowChage.getEventType();
53
        System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
54
                                         entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
55
                                         entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
56
                                         eventType));
57

58
        for (RowData rowData : rowChage.getRowDatasList()) {
59
            if (eventType == EventType.DELETE) {
60
                printColumn(rowData.getBeforeColumnsList());
61
            } else if (eventType == EventType.INSERT) {
62
                printColumn(rowData.getAfterColumnsList());
63
            } else {
64
                System.out.println("-------&gt; before");
65
                printColumn(rowData.getBeforeColumnsList());
66
                System.out.println("-------&gt; after");
67
                printColumn(rowData.getAfterColumnsList());
68
            }
69
        }
70
    }
71
}
72

73
private static void printColumn(List<Column> columns) {
74
    for (Column column : columns) {
75
        System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
76
    }
77
}

值得注意的是，官网对Canal的配置是默认进行全库全表检测的，但在业务中往往不需要这么庞大的一致性，只需要针对业务相关的库表进行配置就行，使用的是正则匹配。

常见统计类型#

①聚合统计 就是之前说过的交差并等集合运算，用于统计多个集合的聚合结果

②排序统计 在面对需要展示的最新列表、排行榜等场景，数据更新频繁或者需要分页时使用Zset 主要应用于那种评论区或者排行榜。

③二值统计 集合元素只存在0或1两种情况，适用于打卡、签到这种场景，用BitMap最好

④基数统计 统计一个集合中的不重复元素的个数，用HYPERLOGLOG最好

HyperLogLog#

​ 前置跳转

前面都说过了，hyperloglog干的就是基数统计这个活儿，基数统计主要就是去重，说到去重，咱们一般会想到以下几个数据结构，咱们慢慢分析

①HashSet： 学java的肯定第一个想到的就是它，天生如此，不需要任何其他外力介入就能实现去重。但这里我们注重的是持久化，而且如果你读过HashSet的源码的话，你会发现在数据变多的时候，HashSet占用的内存会特别多，所以这里我们不多讲，直接说下一个

②BitMap 用bitmap有个很好的优点，就是数据准确性得到了保证，用bitmap存储就如同是创建了一个特别大的数组，将数据对应的位置置为1，不过缺点也很明显，如果数据量过大，达到亿级访问量的话，一个Bitmap就需要将近12MB，对内存非常不友好，所以我们仍然不用他

③HyperLogLog 用它，它解决了上述两种数据结构几乎所有的痛点，唯一美中不足的地方在于它的数据并不是完全准确的，牺牲了约0.81%的准确性，来实现了概率算法，来使内存占用大幅降低。

代码实操-Service

代码实操-Controller

GEO#

前置跳转

需求解析

主要应用在打车、附近推荐、摇一摇之类的功能，要了解为什么使用这种数据结构，为什么不用MySql，以及数据精准度等问题。

代码实操-Service

代码实操-Controller

BitMap#

前置跳转

主要运用在日活统计、签到统计、年登录天数这种业务的实现，这个之前讲过了，说白了就是一个bit数组，里面存储着二进制状态，很像那个AtomicBoolean

但是根据公司体量的大小， 使用的解决方案还是不同的

小厂：传统Mysql建表=>

1
CREATE TABLE user_sign(
2
  keyid BIGINT NOT NULL PRIMARY KEY AUTO_INCREMENT,
3
  user_key VARCHAR(200), #用户id
4
  sign_date DATETIME, #签到日期
5
  sign_count INT #连续签到天数
6
)

大厂(传统)：

大厂(升级)：

采用BitMap+布隆过滤器，下章细讲