Redis与Java - 实践

标签： Java与NoSQL

Transaction

Redis事务(transaction)是一组命令的集合,同命令一样也是Redis的最小执行单位, Redis保证一个事务内的命令执行不被其他命令影响.

`MULTI`    SADD user:1:following 2    SADD user:2:follower 1`EXEC`

vs. RDBMS

事务操作	MySQL	Redis
开启	`start transaction`	`MULTI`
语句	DML	普通命令
取消	`rollback`	`DISCARD`
执行	`commit`	`EXEC`

MySQL的rollback与Redis的DISCARD有一定的区别.
假设现在已经成功执行了事务内的前2条语句, 第3条语句出错:
1. MySQLrollback后,前2条的语句影响消失.
2. Redis可以分为两种情况:
  - 语法错误: 事务中断, 所有语句均得不到执行;
  - 运行错误: (如语法正确,但适用数据类型不对: 像ZADD操作List), EXEC会执行前2条语句, 并跳过第3条语句.
    这样的部分成功会导致数据不一致, 而这一点需要由开发人员负责, 比如提前规划好缓存key的设计.

乐观锁与`WATCH`

悲观锁(Pessimistic Lock): 很悲观,每次读写数据都认为别人会修改,所以每次读数据都会上锁,这样如果别人也想读写这条数据就会阻塞, 直到加锁的人把锁释放. 传统的RDBMS中用到了很多这种锁机制, 如行锁、表锁、读锁、写锁等.
乐观锁(Optimistic Lock): 顾名思义非常乐观, 每次读写数据时候都认为别人不会修改,所以不再上锁,但在更新数据时会判断一下在此期间有没有人更新了这条数据, 这个判断过程可以使用版本号等机制实现, 而Redis默认就对乐观锁提供了支持 –WATCH命令.

WATCH命令可以监控一个/多个key, 一旦其中有一个被修改/删除, 则之后的事务就不会执行,如用WATCH命令来模拟抢票场景:

SET ticket 1        # 现在假设只有一张票了`WATCH` ticket      # 监控票数变化`MULTI`    DECRBY username 400    DECR ticket        [DECR ticket]   # 现在假设有另外一个用户直接把这张票买走了`EXEC`    -> `(nil)`  # 则这条事务执行就不会成功

小结

WATCH命令的作用只是当被监控的key值修改后阻止事务执行,并不能阻止其他Client修改. 所以一旦EXEC执行失败, 可以重新执行整个方法或使用UNWATCH命令取消监控.

乐观锁适用于读多写少情景,即冲突真的很少发生,这样可以省去大量锁的开销. 但如果经常产生冲突,上层应用需要不断的retry,反倒是降低了性能,所以这种情况悲观锁比较适用.

Expire & Cache

Redis可以使用EXPIRE命令设置key的过期时间, 到期后Redis会自动删除它.

命令	作用
`EXPIRE key seconds`	Set a timeout on key.
`TTL key`	Get the time to live for a key
`PERSIST key`	Remove the expiration for a key

除了PERSIST命令之外,SET/GETSET为key赋值的同时也会清除key的过期时间.另外如果WATCH监控了一个拥有过期时间的key,key到期自动删除并不会被WATCH认为该key被修改.

缓存DB数据
为了提高网站负载能力,常需要将一些访问频率较高但对CPU/IO消耗较大的操作结果缓存起来,并希望让这些缓存过期自动删除, 下面我们就使用Redis缓存DB数据, 场景介绍可以参考:.

/** * @author jifang. * @since 2016/6/13 20:08. */public class RedisDAO {
         private static final int _1M = 60 * 1000;    private static final DataSource dataSource;    private static final Jedis redis;    static {        Properties properties = new Properties();        try {            properties.load(ClassLoader.getSystemResourceAsStream("db.properties"));        } catch (IOException ignored) {        }        /** 初始化连接池 **/        HikariConfig config = new HikariConfig();        config.setDriverClassName(properties.getProperty("mysql.driver.class"));        config.setJdbcUrl(properties.getProperty("mysql.url"));        config.setUsername(properties.getProperty("mysql.user"));        config.setPassword(properties.getProperty("mysql.password"));        dataSource = new HikariDataSource(config);        /** 初始化Redis **/        redis = new Jedis(properties.getProperty("redis.host"), Integer.valueOf(properties.getProperty("redis.port")));    }    public List
      
       > executeQuery(String sql) {        List
       
        
         > result;        try {            /** 首先请求Redis **/            String key = sql.replace(' ', '-');            String string = redis.get(key);            // 如果key未命中, 再请求DB            if (string == null || string.trim().isEmpty()) {                ResultSet resultSet = dataSource.getConnection().createStatement().executeQuery(sql);                /** 获得列数/列名 **/                ResultSetMetaData meta = resultSet.getMetaData();                int columnCount = meta.getColumnCount();                List
         
           columnName = new ArrayList<>();                for (int i = 1; i <= columnCount; ++i) {                    columnName.add(meta.getColumnName(i));                }                /** 填充实体 **/                result = new ArrayList<>();                while (resultSet.next()) {                    Map
          
            entity = new HashMap<>(columnCount); for (String name : columnName) { entity.put(name, resultSet.getObject(name)); } result.add(entity); } /**写入Redis**/ String value = JSON.toJSONString(result); redis.set(key, value, "NX", "PX", _1M); } else { result = JSON.parseObject(string, List.class); } } catch (SQLException e) { throw new RuntimeException(e); } return result; } public static void main(String[] args) { RedisDAO dao = new RedisDAO(); List
           
            
             > execute = dao.executeQuery("select * from user"); System.out.println(execute); }}

当服务器内存有限时,如果大量使用缓存而且过期时间较长会导致Redis占满内存; 另一方面为了防止占用内存过大而设置过期时间过短, 则有可能导致缓存命中率过低而使系统整体性能下降.因此为缓存设计一个合理的过期时间是很纠结的, 在Redis中可以限制能够使用的最大内存,并让Redis按照一定规则的淘汰不再需要的key: 修改maxmemory参数,当超过限制会依据maxmemory-policy参数指定的策略来删除不需要的key:

`maxmemory-policy`	规则说明
`volatile-lru`	只对设置了过期时间的key使用LRU算法删除
`allkey-lru`	使用LRU删除一个key
`volatile-random`	只对设置了过期时间的key随机删除一个key
`allkey-random`	随机删除一个key
`volatile-ttl`	删除过期时间最近的一个key
`noevication`	不删除key, 只返回错误(默认)

Sort

Redis的SORT命令可以对List、Set、Sorted-Set类型排序, 并且可以完成与RDBMS 连接查询 类似的任务:

SORT key    [BY pattern]             [LIMIT offset count]             [GET pattern [GET pattern ...]]             [ASC|DESC]             [ALPHA]             [STORE destination]

参数	描述
`ALPHA`	`SORT`默认会将所有元素转换成双精度浮点数比较,无法转换则会提示错误,而使用`ALPHA`参数可实现按字典序比较.
`DESC`	降序排序(`SORT`默认升序排序).
`LIMIT`	指定返回结果范围.
`STORE`	`SORT`默认直接返回排序结果, `STORE`可将排序后结果保存为`List`.

注: SORT在对Sorted-Set排序时会忽略元素分数,只针对元素自身值排序.

BY

很多情况下key实际存储的是对象ID, 有时单纯对ID自身排序意义不大,这就用到了BY参数, 对ID关联的对象的某个属性进行排序:

[BY pattern]

pattern可以是字符串类型key或Hash类型key的某个字段(表示为键名 -> 字段名).如果提供了BY参数, SORT将使用ID值替换参考key中的第一个*并获取其值,然后根据该值对元素排序.

SORT mi.blog:1:my BY mi.blog:*:data->time DESC

注意:
- 当pattern不包含*时, SORT将不会执行排序操作;
- 当ID元素的参考key不存在时,默认设置为0;
- 如果几个ID元素的pattern值相同,则会再比较元素本身值排序.

GET

GET参数不影响排序过程,它的作用是使SORT返回结果不再是元素自身的值,而是GET参数指定的键值:

[GET pattern [GET pattern ...]]

同BY一样, GET参数也支持String类型和Hash类型, 并使用*作为占位符.

SORT mi.blog:1:my BY mi.blog:*:data->time GET mi.blog:*:data->content GET mi.blog:*:data->time

注: GET参数获取自身值需要使用#: GET #

性能

SORT的时间复杂度为O(N+M*log(M)):

where N is the number of elements in the list or set to sort, and M the number of returned elements. When the elements are not sorted, complexity is currently O(N) as there is a copy step that will be avoided in next releases.

所以开发过程中使用SORT需要注意:
1. 尽可能减小待排序key中元素数量(减小N);
2. 使用LIMIT参数限制结果集大小(减小M);
3. 如果待排序数据量较大,尽可能使用STORE将结果缓存.

Message

1. 消息队列

消息队列就是”传递消息的队列”,与消息队列进行交互的实体有两类, 一是生产者: 将需要处理的消息放入队列; 一是消费者: 不断从消息队列中读出消息并处理.

使用消息队列有如下好处:

松耦合: 生产者和消费者无需知道彼此的实现细节, 只需按照协商好的

消息格式读/写, 即可实现不同进程间通信,这就使得生产者和消费者可以由

不同的团队使用

不同的开发语言编写.

易扩展: 消费者可以有多个,且可以分布在不同的Server中, 降低单台Server负载, 横向扩展业务.

Redis提供了BRPOP/BLPOP命令来实现消息队列:

命令	描述
`BRPOP key [key ...] timeout`	Remove and get the last element in a list, or block until one is available
`BLPOP key [key ...] timeout`	Remove and get the first element in a list, or block until one is available
`BRPOPLPUSH source destination timeout`	Pop a value from a list, push it to another list and return it; or block until one is available

注: 若Redis同时监听多个key, 且每个key均有元素可取,则Redis按照从左到右的顺序去挨个读取key的第一个元素.

2. 消息订阅

前面的BRPOP/BLPOP实现的消息队列有一个限制: 如果一个队列被多个消费者监听, 生产者发布一条消息只会被其中一个消费者获取. 因此Redis还提供了一组命令实现“发布/订阅”模式, 同样可用于进程间通信:

“发布/订阅”模式也包含两种角色: 发布者与订阅者. 订阅者可以订阅一个/多个频道, 而发布者可向指定频道发送消息, 所有订阅此频道的订阅者都会收到此消息.

命令	描述
`PUBLISH channel message`	Post a message to a channel
`SUBSCRIBE channel [channel ...]`	Listen for messages published to the given channels
`UNSUBSCRIBE [channel [channel ...]]`	Stop listening for messages posted to the given channels
`PSUBSCRIBE pattern [pattern ...]`	Listen for messages published to channels matching the given patterns
`PUNSUBSCRIBE [pattern [pattern ...]]`	Stop listening for messages posted to channels matching the given patterns

MessagesQueue

/** * @author jifang * @since 16/7/11 下午2:36. */public class MessageQueue
     
       {    private Jedis redis;    private String chanel;    public MessageQueue(Jedis redis, String chanel) {        this.redis = redis;        this.chanel = chanel;    }    public Long publish(T message) {        String json = JSON.toJSONString(message);        return redis.publish(chanel, json);    }    public void subscribe(final MessageHandler
      
        handler) {        redis.subscribe(new JedisPubSub() {            @Override            public void onMessage(String channel, String message) {                for (Type type : handler.getClass().getGenericInterfaces()) {                    if (type instanceof ParameterizedType) {                        ParameterizedType pType = (ParameterizedType) type;                        Type handlerClass = pType.getActualTypeArguments()[0];                        T result = JSONObject.parseObject(message, handlerClass);                        handler.handle(result);                    }                }            }        }, chanel);    }}

public interface MessageHandler
     
       {    void handle(T object);}

注: 发送的消息不会持久化,一个订阅者只能接收到后续发布的消息,之前发送的消息就接收不到了.

持久化

Redis支持两种持久化方式: RDB与AOF. RDB: Redis根据指定的规则“定时”将内存数据快照到硬盘; AOF:Redis在每次执行命令后将命令本身记录下来存放到硬盘.两种持久化方式可结合使用.

RDB

快照执行过程:

Redis使用fork()函数复制一份当前进程副本;

父进程继续接收并处理客户端请求, 而子进程将所有内存数据写入磁盘临时文件;

当子进程将所有数据写完会用该临时文件替换旧的RDB文件, 至此一次快照完成(可以看到自始至终RDB文件都是完整的).

Redis会在以下几种情况下对数据进行快照:

根据配置规则
配置由两个参数构成: 时间窗口M和改动key个数N; 当时间M内被改动的key的个数大于N时, 即符合自动快照条件:

save 900 1save 300 10save 60 10000

用户执行SAVE/BGSAVE/FLUSHALL命令:
除了让Redis自动快照, 当进行服务重启/手动迁移以及备份时也需要我们手动执行快照.

命令	描述
`SAVE`	`SAVE`命令会使Redis同步地执行快照操作(过程中会阻塞所有来自客户端的请求, 因此尽量避免线上使用)
`BGSAVE`	在后台异步执行快照操作,Redis还可继续响应请求
`FLUSHALL`	`FLUSHALL`会清空所有数据,无论是否触发了自动快照条件(只要有配置了),Redis都会执行一次快照
`LASTSAVE`	获取最近一次成功执行快照时间

执行复制
当设置了主从模式, Redis会在复制初始化时执行快照,即使没有配置自动快照条件.

通过RDB方式实现持久化, Redis在启动后会读取RDB快照文件, 将数据从硬盘导入内存, 但如果在持久化过程中Redis异常退出, 就会丢失最后一次快照以后更改的所有数据.

RDB其他配置参数

dir ./                  # 设置工作目录,RDB文件(以及后面的AOF文件)会写入该目录dbfilename dump.rdb     # 设置RDB文件名rdbcompression yes      # 导出RDB是否压缩rdbchecksum yes         # 存储和加载RDB校验完整性stop-writes-on-bgsave-error yes     # 后台备份进程出错时,主进程停止写入.

AOF

AOF将Redis执行的每一条命令追加到硬盘文件中.然后在启动Redis时逐条执行AOF文件中的命令将数据载入内存.

Redis默认没有开启AOF, 需要以如下参数启用:

appendonly yesno-appendfsync-on-rewrite  yes: # 正在导出RDB快照的过程中,停止同步AOF.

AOF重写

开启AOF后, Redis会将每一条有可能更改数据的命令写入AOF文件,这样就导致AOF文件越来越大,即使有可能内存中实际存储的数据并没多少. 因此Redis每当达到一定条件就自动重写AOF文件,这个条件可以在配置文件中设置:

auto-aof-rewrite-percentage 100 # 比起上次重写时的大小,AOF增长率100%时重写auto-aof-rewrite-min-size 64mb  # AOF大小超过64M时重写

此外, 我们还可以使用BGREWRITEAOF命令手动执行AOF重写.

硬盘数据同步

执行AOF持久化时, 由于操作系统缓存机制, 数据并没有真正写入磁盘,而是进入了磁盘缓存, 默认情况下系统每30S执行一次同步操作, 将缓存内容真正写入磁盘, 如果在这30S的系统异常退出则会导致磁盘缓存数据丢失, 如果应用无法忍受这样的损失, 可通过appendfsync参数设置同步机制:

# appendfsync always    # 每次执行写入都执行同步appendfsync everyse     # 每秒执行一次同步操作# appendfsync no        # 不主动进行同步, 而是完全由操作系统执行.

集群

1. Replication

复制(replication)中,Redis的角色可以分为两类, Master:可以执行读/写操作,当写操作导致数据修改时会自动将数据同步给Slave; Slave:一般是只读的,并接受Master同步过来的数据(Slave自身也可以作为Master存在, 如图):

replication复制时序
- Slave启动后向Master发送SYNC命令;Master收到后在后台保存RDB快照, 并将快照期间接收到的所有命令缓存.
- 快照执行完, Master将快照文件与所有缓存的命令发送给Slave;
- Slave接收并载入快照, 然后执行所有收到的缓存命令,这一过程称为复制初始化.
- 复制初始化完成后,Master每接收到写命令就同步给Slave,从而保证主从数据一致.

通过Redis的复制功能可以实现以下应用:
- 读写分离:
  通过复制可实现读写分离, 以提高服务器的负载能力, 可以通过复制建立多个Slave节点, Master只进行写操作, 而由Slave负责读操作, 这种一主多从的结构很适合读多写少的场景.
- Slave持久化
  持久化是一个相对耗时的操作, 为了提高性能, 可以通过复制功能建立一个/多个Slave, 并在Salve中启用持久化, Master禁用持久化. 当Master崩溃后:
  1. 在Slave使用SLAVEOF NO ONE命令将Slave提升成Master继续服务;
  2. 启用之前崩溃的Master, 然后使用SLAVEOF将其设置为新Master的Slave, 即可将数据同步回来.

注意: 当开启复制且Master关闭持久化时, Master崩溃后一定不能直接重启Master, 这是因为当Master重启后, 因为没有开启持久化, 所以Redis内的所有数据都会被清空, 这时Salve从Master接受数据, 所有的Slave也会被清空, 导致Slave持久化失去意义.

关于Redis复制的详细介绍以及配置方式可参考博客:.

2. Sentinel

当Master遭遇异常中断服务后, 需要手动选择一个Slave升级为Master, 以使系统能够继续提供服务. 然而整个过程相对麻烦且需要人工介入, 难以实现自动化. 为此Redis提供了哨兵Sentinel.

Sentinel哨兵是Redis高可用性解决方案之一: 由一个/多个Sentinel实例组成的Sentinel系统可以监视任意多个Master以及下属Slave, 并在监控到Master进入下线状态时, 自动将其某个Slave提升为新的Master, 然后由新的Master代替已下线的Master继续处理命令请求.

如图: 若此时Master:server1进入下线状态, 那么Slave: server2,server3,server4对Master的复制将被迫中止,并且Sentinel系统也会察觉到server1已下线, 当下线时长超过用户设定的下线时长时, Sentinel系统就会对server1执行故障转移操作:

Sentinel会挑选server1下属的其中一台Slave, 将其提升为新Master;

然后Sentinel向server1下属的所有Slave发送新的复制指令,让他们成为新Master的Salve, 当所有Salve都开始复制新Master时, 故障转移操作完成.

另外, Sentinel还会继续监视已下线的server1, 并在他重新上线时, 将其设置为新Master的Slave.

关于Redis哨兵的详细介绍以及配置方式可参考博客:.

3. Cluster

Cluster是Redis提供的另一高可用性解决方案:Redis集群通过分片(sharding)来进行数据共享, 并提供复制与故障转移功能.

一个 Redis 集群通常由多个节点组成, 最初每个节点都是相互独立的,要组建一个真正可工作的集群, 必须将各个独立的节点连接起来.连接各个节点的工作可以使用CLUSTER MEET命令完成:

CLUSTER MEET

向一个节点发送CLUSTER MEET命令,可以使其与ip+port所指定的节点进行握手,当握手成功时, 就会将目标节点添加到当前节点所在的集群中.

案例
假设现在有三个独立的节点 127.0.0.1:7000 、 127.0.0.1:7001 、 127.0.0.1:7002:
- 通过向节点 7000 发送CLUSTER MEET 127.0.0.1 7001命令,可将节点7001添加到节点7000所在的集群中:
- 继续向节点7000发送CLUSTER MEET 127.0.0.1 7002命令,同样也可将节点7002也拉进来:
- 至此, 握手成功的三个节点处于同一个集群:

关于Redis-Cluster的详细介绍以及更多配置方式可参考博客:.

管理

1. 数据库密码

通过在配置文件中使用requirepass参数可为Redis设置密码:

requirepass œ∑´®†¥¨ˆøπ

这样客户端每次连接都需要发送密码,否则Redis拒绝执行客户端命令:

AUTH œ∑´®†¥¨ˆøπ

2. 重命名

Redis支持在配置文件中将命令重命名, 以保证只有自己的应用可以使用该命令:

rename-command FLUSHALL qwertyuiop

如果希望禁用某个命令,可将命令重命名为空字符串.

3. 工具

SLOWLOG
当一条命令执行超过时间限制时,Redis会将其执行时间等信息加入耗时统计日志, 超时时间等可通过以下配置实现:

slowlog-log-slower-than 10000   # 超时限制(单位微秒)slowlog-max-len 128             # 记录条数限制

MONITOR : 监控Redis执行的所有命令

注意: MONITOR命令非常影响Redis性能, 一个客户端使用MONITOR会降低Redis将近一半的负载能力. Instagram团队开发了一个基于MONITOR命令的Redis查询分析工具redis-faina, 可根据MONITOR的监控结果分析出最常用的命令/访问最频繁的key等信息, 详细可参考博客:.

其他常用管理工具

TIME        # 系统时间戳与微秒数DBSIZE      # 当前数据库的key数量INFO        # Redis服务器信息CONFIG GET  # 获取配置信息CONFIG SET  # 设置配置信息CONFIG REWRITE  # 把值写到配置文件CONFIG RESTART  # 更新INFO命令信息CLIENT LIST # 客户端列表CLIENT KILL # 关闭某个客户端CLIENT SETNAME  # 为客户端设置名字CLIENT GETNAME  # 获取客户端名字DEBUG OBJECT key    # 调试选项,查看一个key的信息DEBUG SEGFAULT      # 模拟段错误,使服务器崩溃OBJECT (refcount|encoding|idletime) key

参考&拓展