Redis

Posted 2020-12-01 浮世间

Redis

Redis消息队列

LIST：基于列表的方式，所有的消费者数据加起来是列表中的所有数据.

发布/订阅：每个消费者订阅独立的channel,每个数据都是独立的。

---

Redis持久化

Redis提供了两种持久化方式：1 RDB快照方式 2 AOF方式

RDB方式：

满足一定条件时，会创建一个子进程，复制当前的数据，把数据写入到硬盘中某个文件，写入完成后替换原来的存储文件。数据一般存储在dump.rdb中。UNIX系统中支持写时复制，即刚开始会执行持久化写入磁盘的操作，如果此时有其他的数据发生改变，就复制一份数据执行。

除了这种自动的快照方式，还支持命令方式持久化：

• SAVE：通过阻塞的方式，用父进程来持久化，此时无法执行其他的请求。
• BGSAVE：通过fork子进程的方式，持久化。

AOF方式：

每次操作都会记录命令，这样会造成某些命令的冗余，比如添加了一个属性，再删除，那么这两个操作都是冗余的。redis提供了一些优化，所以可以避免这些冗余信息。命令记录在appendonly.aof中

---

常用命令

连接操作相关的命令

• quit：关闭连接（connection）
• auth：简单密码认证

持久化

• save：将数据同步保存到磁盘
• bgsave：将数据异步保存到磁盘
• lastsave：返回上次成功将数据保存到磁盘的Unix时戳
• shundown：将数据同步保存到磁盘，然后关闭服务

远程服务控制

• info：提供服务器的信息和统计
• monitor：实时转储收到的请求
• slaveof：改变复制策略设置
• config：在运行时配置Redis服务器

对value操作的命令

• exists(key)：确认一个key是否存在
• del(key)：删除一个key
• type(key)：返回值的类型
• keys(pattern)：返回满足给定pattern的所有key
• randomkey：随机返回key空间的一个
• keyrename(oldname, newname)：重命名key
• dbsize：返回当前数据库中key的数目
• expire：设定一个key的活动时间（s）
• ttl：获得一个key的活动时间
• select(index)：按索引查询
• move(key, dbindex)：移动当前数据库中的key到dbindex数据库
• flushdb：删除当前选择数据库中的所有key
• flushall：删除所有数据库中的所有key

对String操作的命令

• set(key, value)：给数据库中名称为key的string赋予值value
• get(key)：返回数据库中名称为key的string的value
• getset(key, value)：给名称为key的string赋予上一次的value
• mget(key1, key2,…, key N)：返回库中多个string的value
• setnx(key, value)：添加string，名称为key，值为value,相当于if not exist
• setex(key, time, value)：向库中添加string，设定过期时间time
• mset(key N, value N)：批量设置多个string的值
• msetnx(key N, value N)：如果所有名称为key i的string都不存在
• incr(key)：名称为key的string增1操作
• incrby(key, integer)：名称为key的string增加integer
• decr(key)：名称为key的string减1操作
• decrby(key, integer)：名称为key的string减少integer
• append(key, value)：名称为key的string的值附加value
• substr(key, start, end)：返回名称为key的string的value的子串

对List操作的命令

• rpush(key, value)：在名称为key的list尾添加一个值为value的元素
• lpush(key, value)：在名称为key的list头添加一个值为value的 元素
• llen(key)：返回名称为key的list的长度
• lrange(key, start, end)：返回名称为key的list中start至end之间的元素
• ltrim(key, start, end)：截取名称为key的list
• lindex(key, index)：返回名称为key的list中index位置的元素
• lset(key, index, value)：给名称为key的list中index位置的元素赋值
• lrem(key, count, value)：删除count个key的list中值为value的元素
• lpop(key)：返回并删除名称为key的list中的首元素
• rpop(key)：返回并删除名称为key的list中的尾元素
• blpop(key1, key2,… key N, timeout)：lpop命令的block版本。
• brpop(key1, key2,… key N, timeout)：rpop的block版本。
• rpoplpush(srckey, dstkey)：返回并删除名称为srckey的list的尾元素，并将该元素添加到名称为dstkey的list的头部

对Set操作的命令

• sadd(key, member)：向名称为key的set中添加元素member
• srem(key, member) ：删除名称为key的set中的元素member
• spop(key) ：随机返回并删除名称为key的set中一个元素
• smove(srckey, dstkey, member) ：移到集合元素
• scard(key) ：返回名称为key的set的基数
• sismember(key, member) ：member是否是名称为key的set的元素
• sinter(key1, key2,…key N) ：求交集
• sinterstore(dstkey, (keys)) ：求交集并将交集保存到dstkey的集合
• sunion(key1, (keys)) ：求并集
• sunionstore(dstkey, (keys)) ：求并集并将并集保存到dstkey的集合
• sdiff(key1, (keys)) ：求差集
• sdiffstore(dstkey, (keys)) ：求差集并将差集保存到dstkey的集合
• smembers(key) ：返回名称为key的set的所有元素
• srandmember(key) ：随机返回名称为key的set的一个元素

对Hash操作的命令

• hset(key, field, value)：向名称为key的hash中添加元素field
• hget(key, field)：返回名称为key的hash中field对应的value
• hmget(key, (fields))：返回名称为key的hash中field i对应的value
• hmset(key, (fields))：向名称为key的hash中添加元素field
• hincrby(key, field, integer)：将名称为key的hash中field的value增加integer
• hexists(key, field)：名称为key的hash中是否存在键为field的域
• hdel(key, field)：删除名称为key的hash中键为field的域
• hlen(key)：返回名称为key的hash中元素个数
• hkeys(key)：返回名称为key的hash中所有键
• hvals(key)：返回名称为key的hash中所有键对应的value
• hgetall(key)：返回名称为key的hash中所有的键（field）及其对应的value

对sortedset的操作命令

1	[ZADD key score1 member1 score2 member2] 向有序集合添加一个或多个成员，或者更新已存在成员的分数
2	ZCARD key 获取有序集合的成员数
3	ZCOUNT key min max 计算在有序集合中指定区间分数的成员数
4	ZINCRBY key increment member 有序集合中对指定成员的分数加上增量 increment
5	[ZINTERSTORE destination numkeys key key ...] 计算给定的一个或多个有序集的交集并将结果集存储在新的有序集合 key 中
6	ZLEXCOUNT key min max 在有序集合中计算指定字典区间内成员数量
7	[ZRANGE key start stop WITHSCORES] 通过索引区间返回有序集合成指定区间内的成员
8	[ZRANGEBYLEX key min max LIMIT offset count] 通过字典区间返回有序集合的成员
9	[ZRANGEBYSCORE key min max WITHSCORES] [LIMIT] 通过分数返回有序集合指定区间内的成员
10	ZRANK key member 返回有序集合中指定成员的索引
11	[ZREM key member member ...] 移除有序集合中的一个或多个成员
12	ZREMRANGEBYLEX key min max 移除有序集合中给定的字典区间的所有成员
13	ZREMRANGEBYRANK key start stop 移除有序集合中给定的排名区间的所有成员
14	ZREMRANGEBYSCORE key min max 移除有序集合中给定的分数区间的所有成员
15	[ZREVRANGE key start stop WITHSCORES] 返回有序集中指定区间内的成员，通过索引，分数从高到底
16	[ZREVRANGEBYSCORE key max min WITHSCORES] 返回有序集中指定分数区间内的成员，分数从高到低排序
17	ZREVRANK key member 返回有序集合中指定成员的排名，有序集成员按分数值递减(从大到小)排序
18	ZSCORE key member 返回有序集中，成员的分数值
19	[ZUNIONSTORE destination numkeys key key ...] 计算给定的一个或多个有序集的并集，并存储在新的 key 中
20	[ZSCAN key cursor MATCH pattern] [COUNT count] 迭代有序集合中的元素（包括元素成员和元素分值）

---

Redis源码

源码地址

协议

和Redis Server通信的协议规则都在redis.clients.jedis.Protocol这个类中，主要是通过对RedisInputStream和RedisOutputStream对读写操作来完成。

命令的发送都是通过redis.clients.jedis.Protocol的sendCommand来完成的，就是对RedisOutputStream写入字节流

Java代码

<span style="font-size: small;"><span style="font-size: small;">    private void sendCommand(final RedisOutputStream os, final byte[] command,  
            final byte[]... args) {  
        try {  
            os.write(ASTERISK_BYTE);  
            os.writeIntCrLf(args.length + 1);  
            os.write(DOLLAR_BYTE);  
            os.writeIntCrLf(command.length);  
            os.write(command);  
            os.writeCrLf();  
  
            for (final byte[] arg : args) {  
                os.write(DOLLAR_BYTE);  
                os.writeIntCrLf(arg.length);  
                os.write(arg);  
                os.writeCrLf();  
            }  
        } catch (IOException e) {  
            throw new JedisConnectionException(e);  
        }  
    }</span></span> 

从这里可以看出redis的命令格式

[*号][消息元素个数] ( 消息元素个数 = 参数个数 + 1个命令)

[$号][命令字节个数]

[命令内容]

[$号][参数字节个数]

[参数内容]

[$号][参数字节个数]

[参数内容]

返回的数据是通过读取RedisInputStream 进行解析处理后得到的

Java代码

private Object process(final RedisInputStream is) {  
    try {  
        byte b = is.readByte();  
        if (b == MINUS_BYTE) {  
            processError(is);  
        } else if (b == ASTERISK_BYTE) {  
            return processMultiBulkReply(is);  
        } else if (b == COLON_BYTE) {  
            return processInteger(is);  
        } else if (b == DOLLAR_BYTE) {  
            return processBulkReply(is);  
        } else if (b == PLUS_BYTE) {  
            return processStatusCodeReply(is);  
        } else {  
            throw new JedisConnectionException("Unknown reply: " + (char) b);  
        }  
    } catch (IOException e) {  
        throw new JedisConnectionException(e);  
    }  
    return null;  
}  

通过返回数据的第一个字节来判断返回的数据类型，调用不同的处理函数

[-号] 错误信息

[*号] 多个数据 结构和发送命令的结构一样

[:号] 一个整数

[$号] 一个数据 结构和发送命令的结构一样

[+号] 一个状态码

连接

和Redis Sever的Socket通信是由 redis.clients.jedis.Connection 实现的

Connection 中维护了一个底层Socket连接和自己的I/O Stream 还有Protocol

I/O Stream是在Connection中Socket建立连接后获取并在使用时传给Protocol的

Connection还实现了各种返回消息由byte转为String的操作

Java代码

private String host;  
private int port = Protocol.DEFAULT_PORT;  
private Socket socket;  
private Protocol protocol = new Protocol();  
private RedisOutputStream outputStream;  
private RedisInputStream inputStream;  
private int pipelinedCommands = 0;  
private int timeout = Protocol.DEFAULT_TIMEOUT;  

Java代码

public void connect() {  
    if (!isConnected()) {  
        try {  
            socket = new Socket();  
            socket.connect(new InetSocketAddress(host, port), timeout);  
            socket.setSoTimeout(timeout);  
            outputStream = new RedisOutputStream(socket.getOutputStream());  
            inputStream = new RedisInputStream(socket.getInputStream());  
        } catch (IOException ex) {  
            throw new JedisConnectionException(ex);  
        }  
    }  
}  

可以看到，就是一个基本的Socket

这里分享个经验，timeout这个参数默认是2000，我做的项目中有部分是离线运算的，如果读取比较大的数据(大Set 大List之类的)有可能会超过这个时间，可以在JedisPool的构造参数中增大这个值。在线服务一般不要修改。

原生客户端

redis.clients.jedis.BinaryClient 继承 Connection, 封装了Redis的所有命令(http://redis.io/commands)

从名子可以看出 BinaryClient 是Redis客户端的二进制版本，参数都是byte[]的

BinaryClient 是通过Connection的sendCommand 调用Protocol的sendCommand 向Redis Server发送命令

Java代码

public void get(final byte[] key) {  
    sendCommand(Command.GET, key);  
}  

redis.clients.jedis.Client可以看成是BinaryClient 的高级版本，函数的参数都是String int long 这类的，并由redis.clients.util.SafeEncoder 转成byte后 再调用BinaryClient 对应的函数

Java代码

public void get(final String key) {  
    get(SafeEncoder.encode(key));  
}  

这二个client只完成了发送命令的封装，并没有处理返回数据

Jedis客户端

我们平时用的基本都是由redis.clients.jedis.Jedis类封装的客户端

Jedis是通过对Client的调用， 完成命令发送和返回数据 这个完整过程的

以GET命令为例，其它命令类似

Jedis中的get函数如下

Java代码

public String get(final String key) {  
    checkIsInMulti();  
    client.sendCommand(Protocol.Command.GET, key);  
    return client.getBulkReply();  
}  

checkIsInMulti();

是进行无事务检查 Jedis不能进行有事务的操作带事务的连接要用redis.clients.jedis.Transaction类

client.sendCommand(Protocol.Command.GET, key);

调用Client发送命令

return client.getBulkReply();

处理返回值

分析到这里 一个Jedis客户端的基本实现原理应该很清楚了

连接池

在实现项目中，要使用连接池来管理Jedis的生命周期，满足多线程的需求，并对资源合理使用。

jedis有两个连接池类型， 一个是管理 Jedis， 一个是管理ShardedJedis（jedis通过java实现的多Redis实例的自动分片功能，后面会分析）

他们都是Pool的不同实现

Java代码

public abstract class Pool<T> {  
    private final GenericObjectPool internalPool;  
  
    public Pool(final GenericObjectPool.Config poolConfig,  
            PoolableObjectFactory factory) {  
        this.internalPool = new GenericObjectPool(factory, poolConfig);  
    }  
  
    @SuppressWarnings("unchecked")  
    public T getResource() {  
        try {  
            return (T) internalPool.borrowObject();  
        } catch (Exception e) {  
            throw new JedisConnectionException(  
                    "Could not get a resource from the pool", e);  
        }  
    }  
......  
......  

从代码中可以看出，Pool是通过 Apache Commons Pool 中的GenericObjectPool这个对象池来实现的

（Apache Commons Pool内容可参考http://phil-xzh.iteye.com/blog/320983 ）

在JedisPool中，实现了一个符合 Apache Commons Pool 相应接口的JedisFactory，GenericObjectPool就是通过这个JedisFactory来产生Jedis对你的

其实JedisPoolConfig也是对Apache Commons Pool 中的Config进行的一个封装

当你在调用 getResource 获取Jedis时， 实际上是Pool内部的internalPool调用borrowObject()借给你了一个实例

而internalPool 这个 GenericObjectPool 又调用了 JedisFactory 的 makeObject() 来完成实例的生成 (在Pool中资源不够的时候)

Java代码

public Object makeObject() throws Exception {  
    final Jedis jedis;  
    if (timeout > 0) {  
        jedis = new Jedis(this.host, this.port, this.timeout);  
    } else {  
        jedis = new Jedis(this.host, this.port);  
    }  
  
    jedis.connect();  
    if (null != this.password) {  
        jedis.auth(this.password);  
    }  
    return jedis;  
}   

客户端的自动分片

从这个结构图上可以看出 ShardedJedis 和 BinaryShardedJedis 正好是 Jedis 和 BinaryJedis 的分片版本

其实它们都是 先获取hash(key)后对应的 Jedis 再有这个Jedis进行操作

Java代码

public String get(String key) {  
    Jedis j = getShard(key);  
    return j.get(key);  
} 

分片逻辑都是在 Sharded<R, S extends ShardInfo> 中实现的

它的构造函数如下

Java代码

public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {  
    this.algo = algo;  
    this.tagPattern = tagPattern;  
    initialize(shards);  
}  

shards是一组ShardInfo, 具体实现是JedisShardInfo, 每个里面记录分片信息和权重，并负责完成分片对应Jedis实例创建

Sharded的初始化和一致性哈希（Consistent Hashing）的思想是一样的，但这个并不能实现节点的动态变更，只能体现出节点的 权重分配

Java代码

nodes = **new** TreeMap<Long, S>() 

这个nodes就是一个虚拟的结点分布环，由TreeMap实现，保证按Key有序，Value就是对应的ShardInfo

Java代码

160 * shardInfo.getWeight()  

根据每个shard的weight值，默认是1，生成160倍的虚拟节点，hash后放到nodes中，也就是分布到环上

Java代码

resources.put(shardInfo, shardInfo.createResource()); 

每个shardInfo对应的jedis，也就是真正的操作节点，放到resources中

Java代码

private void initialize(List<S> shards) {  
    nodes = new TreeMap<Long, S>();  
  
    for (int i = 0; i != shards.size(); ++i) {  
        final S shardInfo = shards.get(i);  
        if (shardInfo.getName() == null)  
            for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {  
                nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), shardInfo);  
            }  
        else  
            for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {  
                nodes.put(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n), shardInfo);  
            }  
        resources.put(shardInfo, shardInfo.createResource());  
    }  
}  

通过Key获取对应的jedis时，先对key进行hash，和前面初始化节点环时，使用相同的算法

再从nodes这个虚拟的环中取出 大于等于 这个hash值的第一个节点（shardinfo），没有就取nodes中第一个节点（所谓的环 其实是逻辑上实现的）

最后从resources中取出jedis来

Java代码

public S getShardInfo(byte[] key) {  
    SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));  
    if (tail.size() == 0) {  
        return nodes.get(nodes.firstKey());  
    }  
    return tail.get(tail.firstKey());  
}  

Java代码

public R getShard(String key) {  
    return resources.get(getShardInfo(key));  
}