Redis 基础知识

Install

$ wget http://download.redis.io/releases/redis-5.0.7.tar.gz
$ tar xzf redis-5.0.7.tar.gz
$ cd redis-5.0.7
$ make
$ src/redis-server

$ src/redis-cli   // Redis using the built-in client
redis-cli -h host -p port -a password   // remote server  --raw 防止乱码
redis> set foo bar
OK
redis> get foo
"bar"
CONFIG GET  # check config

Redis五种数据类型

Redis 命令参考
all commands

del key
dump key
exists key
expire key s
expireat key timestamp
pexpire key ms
pexpireat key milliseconds-timestamp
KEYS pattern  //查找所有符合给定模式( pattern)的 key 
MOVE key db_name
persist key  
PTTL key
TTL key
randomkey
rename key newkey
renameNX key newkey
TYPE key

string（字符串）： k,v
hash（哈希）: key field:value {"id":1,"name":"zs","age":18}
list（列表）: key ["v1","v2","v3"]
set（集合）: key (score1 member1, score2 member2)
zset(sorted set：有序集合) : zrevrange key 0 3

• 字符串

get key
getrange key start end
getset key new-value
getbit key
mget key1 key2
setbit key
setnx key
setex key
setrange key start end
strlen key
mset key value [...]
msetnx ...
psetex key 
incr key // num+1
incrby key  increment // 将 key 所储存的值加上给定的增量值（increment)
incrbyfloat key increment
decr key // num-1
decrby key decrement   // key 所储存的值减去给定的减量值（decrement）
append key value  // +value

• HASH

hset key field value 
hsetnx key field value
hmget key field1 field2
hmset key field1 field2
hdel key field1 
hexists key field
hget key field
hgetall key
hincrby key field increment  // 
hincrbyfloat key field increment
hkeys key
hlen key
hvals  key   // 获取哈希表中所有值
HSCAN key cursor [MATCH pattern] [COUNT count]  //迭代哈希表中的键值对。

• List

lpush key value  // 插入表头
lpushx key value
blpop key timeout
brpop key timeout
brpoplpush source destination timeout
lindex key index 
linsert key before|after pivot value
llen key
lpop key   // 移出并获取列表的第一个元素
lrange key start stop
lrem key count value
lset key index value 
ltrim key start stop
rpop key   // 移出并获取列表的最一个元素
rpoplpush source destination
rpush key value  // 在列表中添加一个或多个值
rpushx key value 

• Set

sadd key member1
scard key   // 获取集合的成员数
sdiff key1   //差集
sdiffstore destination key1 [key2]]
sinter key  //交集
sinterstore  destination key1   // 返回给定所有集合的交集并存储在 destination 中
sismember key member   //判断 member 元素是否是集合 key 的成员
smembers key
smove source destination member   // 将 member 元素从 source 集合移动到 destination 集合
spop key    //移除并返回集合中的一个随机元素 
srandmember key [count]   //返回集合中一个或多个随机数
srem key member1   // 移除集合中一个或多个成员
sunion key1 // 并集
sunionstore destination key1   //所有给定集合的并集存储在 destination 集合中
sscan key cursor [match pattern][Count count]  //迭代集合中的元素

Sorted set
zadd  key1 member1
zcard key
zcount key min max 
zincrby key increment member
zinterstore destination numbers key
zlexcount key min max 
zrange key start stop
zrangebylex key min max
zrangebyscore key min max [WITHSCORES] [LIMIT]

• Redis HyperLogLog 基数统计

pfadd key element
pfcount key
pfmerge key destkey sourcekey

pub/sub 发布订阅

SUBSCRIBE channel [channel ...] // SUBSCRIBE redisChat
PUBLISH channel message // PUBLISH redisChat "Redis is a great caching technique"  
PSUBSCRIBE pattern 
PUBSUB subcommand [argument [argument ...]]  //查看订阅与发布系统状态
PUNSUBSCRIBE [pattern [pattern ...]]  //退订所有给定模式的频道
UNSUBSCRIBE [channel [channel ...]]  //退订给定的频道

事务与Scripting

MULTI   // 开始
EXEC  //退出
DISCARD  //取消事务
UNWATCH  //
WATCH  //监听

EVAL script numkeys key [key ...] arg [arg ...] // EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second
script kill
script flush  //从缓存中移除所有脚本
script load script
script existis script

connect

CONFIG set requirepass "password"
auth "password"
echo message
ping
quit
select index
Jedis jedis = new Jedis("IP", 6379);
jedis.auth("你给redis设置的密码")

服务器

info
bgsave save //数据备份
CONFIG GET dir //如果需要恢复数据，只需将备份文件 (dump.rdb) 移动到 redis/bin并启动服务即可
config get maxclients  // 最大连接数

问题

• 数据为什么会过期？ redis是用来做数据缓存的： 过期时间到了，内存溢出
• redis的过期策略是什么样的？定期删除+惰性删除
• 内存淘汰机制
  ① noeviction： 当内存不足以容纳新写入数据时，新写入操作会报错，无法写入新数据，一般不采用。
  ②allkeys-lru：当内存不足以容纳新写入数据时，在键空间中，移除最近最少使用的key，这个是最常用的。
  ③allkeys-random：当内存不足以容纳新写入的数据时，在键空间中，随机移除key，一般也不使用。
  ④volatile-lru：volatile-lru：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，移除最近最少使用的key（这个一般不太合适） 。
  ⑤volatile-random：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，随机移除某个key 。
  ⑥volatile-ttl：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，有更早过期时间的key优先移除。

maxmemory 100mb  # 设置最大内存
maxmemory-policy volatile-lru   #默认是noeviction，即不进行数据淘汰 config

• 手写一个LRU算法

//基于JavaLinkedHashMap实现
public class LRUCache<K,V> extends LinkedHashMap<K,V>{
private final int CACHE_SIZE;
      
//保存传递进来的最大数据量
public LRUCache(int cacheSize){
//设置hashmap的初始大小，同时最后一个true指的是让linkedhashmap按照访问顺序来进行排序，
//最近访问的放在头，最老访问的放在尾
super((int)Math.ceil(cacheSize/0.75)+1,0.75f,true);
          CACHE_SIZE = CacheSize；
      }

@Override
protected boolean removeEldestEntry(Map.Entry eldest){
//当map中的数据量大于指定的缓存个数的时候，就自动删除最老的数据。
return size() > CACHE_SIZE;
      }
}

• 库存控制（DECR）和自增序列（INCR）

Redis性能调优

• 避免长耗时命令
  SSCAN/HSCAN/ZSCAN等命令，分别用于对Set/Hash/Sorted Set中的元素进行游标式遍历
  slowlog 自动记录耗时较长的命令

slowlog-log-slower-than xxxms  #执行时间慢于xxx毫秒的命令计入Slow Log
slowlog-max-len xxx  #Slow Log的长度，即最大纪录多少条Slow Log
SLOWLOG GET [number]
SLOWLOG RESET

• 网络引发的延迟 : 尽可能使用长连接或连接池, 使用Pipeline
• 数据持久化引发的延迟
• 制定合理的持久化策略
  1. AOF + fsync always的设置虽然能够绝对确保数据安全，但每个操作都会触发一次fsync，会对Redis的性能有比较明显的影响。
  2. AOF + fsync every second是比较好的折中方案，每秒fsync一次
  3. AOF + fsync never会提供AOF持久化方案下的最优性能
  4. 使用RDB持久化通常会提供比使用AOF更高的性能，但需要注意RDB的策略配置
  5. 每一次RDB快照和AOF Rewrite都需要Redis主进程进行fork操作。fork操作本身可能会产生较高的耗时，与CPU和Redis占用的内存大小有关。根据具体的情况合理配置RDB快照和AOF Rewrite时机，避免过于频繁的fork带来的延迟
• Swap引发的延迟
• 数据淘汰引发的延迟
• 引入读写分离机制

主从复制与集群分片

• 主从复制
  Redis支持一主多从的主从复制架构。一个Master实例负责处理所有的写请求，Master将写操作同步至所有Slave。
  1. 主从之间的数据以Redis Protocol进行同步
  2. 借助Redis Sentinel实现高可用
• 集群分片
  1.为何要做集群分片?
  Redis中存储的数据量大，一台主机的物理内存已经无法容纳
  Redis的写请求并发量大，一个Redis实例以无法承载
  2.分片方案：Redis Cluster
  能够自动将数据分散在多个节点上
  当访问的key不在当前分片上时，能够自动将请求转发至正确的分片
  当集群中部分节点失效时仍能提供服务
  3.Redis Cluster分片原理

Redis Cluster中共有16384个hash slot，Redis会计算每个key的CRC16，将结果与16384取模，来决定该key存储在哪一个hash slot中，同时需要指定Redis Cluster中每个数据分片负责的Slot数，Slot的分配在任何时间点都可以进行重新分配。
客户端在对key进行读写操作时，可以连接Cluster中的任意一个分片，如果操作的key不在此分片负责的Slot范围内，Redis Cluster会自动将请求重定向到正确的分片上。
hash tags -- 在基础的分片原则上，Redis还支持hash tags功能，以hash tags要求的格式明明的key，将会确保进入同一个Slot中
使用Redis Cluster时，pipelining、事务和LUA Script功能涉及的key必须在同一个数据分片上，否则将会返回错误。
如要在Redis Cluster中使用上述功能，就必须通过hash tags来确保一个pipeline或一个事务中操作的所有key都位于同一个Slot中。
有一些客户端（如Redisson）实现了集群化的pipelining操作，可以自动将一个pipeline里的命令按key所在的分片进行分组，分别发到不同的分片上执行。
但是Redis不支持跨分片的事务，事务和LUA Script还是必须遵循所有key在一个分片上的规则要求。

• 主从复制 vs 集群分片

Redis Cluster能够解决单节点上数据量过大的问题
Redis Cluster能够解决单节点访问压力过大的问题
Redis Cluster包含了主从复制的能力
在主从复制和集群分片两个方案中做出选择时，应该从应用软件的功能特性、数据和访问量级、未来发展规划等方面综合考虑，只在确实有必要引入数据分片时再使用Redis Cluster。

开发

Redis的Java客户端很多，官方推荐的有三种：Jedis、Redisson和lettuce。
Jedis：轻量，简洁，便于集成和改造
Redisson： 性能高，文档丰富

//todo： 自定义redis source 
/**
 * 
 * hset areas AREA_US US
 * hset areas AREA_CT TW,HK
 * hset areas AREA_AR PK,KW,SA
 * hset areas AREA_IN IN
 *
 * HashMap
 *
 * US，AREA_US
 * TW，AREA_CT
 * HK，AREA_CT
 *
 */
public class KkbRedisSource implements SourceFunction<HashMap<String,String>> {

    private Logger logger = LoggerFactory.getLogger(KkbRedisSource.class);

    private Jedis jedis;
    private boolean isRunning=true;

    @Override
    public void run(SourceContext<HashMap<String, String>> cxt) throws Exception {
        this.jedis = new Jedis("192.168.1.120",6379);

        HashMap<String, String> map = new HashMap<>();
        while(isRunning){
          try{
              map.clear();
              Map<String, String> areas = jedis.hgetAll("areas");
              for(Map.Entry<String,String> entry: areas.entrySet()){
                  String area = entry.getKey();
                  String value = entry.getValue();
                  String[] fields = value.split(",");
                  for(String country:fields){
                      //
                      map.put(country,area);
                  }
              }
              if(map.size() > 0 ){
                  cxt.collect(map);
              }
              Thread.sleep(60000);
          }catch (JedisConnectionException e){
              logger.error("redis连接异常",e.getCause());
              this.jedis = new Jedis("192.168.167.254",6379);
          }catch (Exception e){
              logger.error("数据源异常",e.getCause());
          }

        }

    }

    @Override
    public void cancel() {
        isRunning=false;
        if(jedis != null){
            jedis.close();
        }

    }
}