redis数据安全与性能保障

一.持久化

1. 快照： 将存在于某一时刻的所有数据都写入硬盘里面

方法：

1. 客户端通过向redis发送bgsave命令（创建子进程）
2. 客户端通过向redis发送save命令，但是会阻塞其他命令，所以只有内存不够，或者不怕阻塞的时候才可以用。但是不要创建子进程，不会导致redis停顿，并且由于没有子进程抢资源所以比bgsave快。
3. 设置了save选项：比如 save 60 10000，表示从最近一次创建快照之后开始算起，当有60s内有10000次写入的时候就会触发bgsave命令，可以有多个save配置，任意一个满足即可。
4. 通过shutdown接收到关闭请求时，或者接收到标准的term信号，执行save命令
5. 当一个redis服务器连接另一个redis服务器，想对方发送sync时，若主服务器没执行bgsave，或者并非刚刚执行完，那么主服务器就会执行bgsave。

缺点：当redis、系统或者硬件中的一个发生崩溃，将丢失最近一次创建快照后的数据。

TIPS: 将开发环境尽可能的模拟生产环境以得到正确的快照生成速率配置。

2. AOF：在执行写命令时，将被执行的写命令复制到硬盘里面

使用appendonlyyes配置选项打开，下图是appendfsync配置选项。

选项目	同步频率
always	每个写操作都要同步写入，严重降低redis速度损耗硬盘寿命
everysec	每秒执行一次，将多个写入同步，墙裂推荐
no	让os决定，不稳定，不知道会丢失多少数据

自动配置aof重写：

• auto-aof-rewrite-percentage 100
• auto-aof-rrewrite-min-size 64
  当启用aof持久化之后，当aof文件体积大于64mb并且体积比上一次大了100%，就会执行bgrewriteaof命令。

缺点：1.aof文件过大，2. 文件过大导致还原事件过长。
但是可以对其进行重写压缩。

二. 复制

就像之前所说当一个从服务器连接一个主服务器的时候，主服务器会创建一个快照文件并将其发送到从服务器。

在配置中包含slaveof host port选项指定主服务器，启动时候会先执行aof或者快照文件。

也可以通过发送flaveof no one命令来终止复制操作，通过slaveof host port命令来开始复制一个主服务器，会直接执行下面的连接操作。

步骤	主服务器操作	从服务器操作
1	（等待命令）	连接主服务器，发送sync命令
2	开始执行bgsave，并使用缓冲区记录bgsave之后执行的所有写命令	根据配置选项决定使用现有数据处理客户端请求还是返回错误
3	Bgsave执行完毕，向从服务器发快照，并在发送期间继续用缓冲区记录写命令	丢弃所有旧数据，载入快照文件
4	快照发送完毕，向从服务器发送缓冲区里的写命令	完成快照解释，开始接受命令
5	缓冲区存储的写命令发送完毕：从现在起每执行一个写命令都发给从服务器	执行主服务器发来的所有存储在缓冲区里的写；并接受执行主服务器发来的写命令

三. 处理故障系统

验证快照和aof文件

• redis-check-aof
• redis-check-dump

检查aof和快照文件的状态，在有需要的情况下对aof文件进行修复。

更换新的故障主服务器

假设A为主服务器，B为从服务器，当机器A发生故障的时候，更换服务器的步骤如下：
首先向机器B发送一个save命令，将这个快照文件发送给机器C，在C上启动Redis，让B成为C的从服务器。

将从服务器升级为主服务器

将从服务器升级为主服务器，为升级后的主服务器创建从服务器。

redis事务

四. 事务

multi: 标记一个事务块的开始。

事务块内的多条命令会按照先后顺序被放进一个队列当中，最后由 EXEC 命令原子性(atomic)地执行。

exec: 执行所有事务块内的命令。

假如某个(或某些) key 正处于 WATCH 命令的监视之下，且事务块中有和这个(或这些) key 相关的命令，那么 EXEC 命令只在这个(或这些) key 没有被其他命令所改动的情况下执行并生效，否则该事务被打断(abort)。

redis的事务包裹在multi命令和exec命令之中，在jedis中通过如下实现

        public class RedisJava extends Thread{
    static Response<String> ret;
    Jedis conn = new Jedis("localhost");
    @Override
    public void run() {
        
        Transaction t = conn.multi();
        t.incr("notrans:");
        Response<String> result1 = t.get("notrans:");
        try {
            Thread.sleep(1L);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        t.incrBy("notrans:", -1);
        t.exec();
        String foolbar = result1.get();       
        System.out.println(foolbar);
    }


    public static void main(String[] args) {
        Jedis conn = new Jedis("localhost");
        Thread t1 = new RedisJava();
        Thread t2 = new RedisJava();
        Thread t3 = new RedisJava();
        t1.start();
        t2.start();
        t3.start();
        

    }
}

• wathc：
  监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。
• unwatch：
  取消 WATCH 命令对所有 key 的监视。
  如果在执行 WATCH 命令之后， EXEC 命令或 DISCARD 命令先被执行了的话，那么就不需要再执行 UNWATCH 了。
  的监视，因此这两个命令执行之后，就没有必要执行 UNWATCH 了。
• discard :取消事务，放弃执行事务块内的所有命令。取消watch，清空任务队列。
  如果正在使用 WATCH 命令监视某个(或某些) key，那么取消所有监视，等同于执行命令 UNWATCH 。

一个简单的商品买卖demo如下：

key	type
inventory：id	set
market	zset
user:id	hash

public boolean listItem(
           Jedis conn, String itemId, String sellerId, double price) {

       String inventory = "inventory:" + sellerId;
       String item = itemId + '.' + sellerId;
       long end = System.currentTimeMillis() + 5000;

       while (System.currentTimeMillis() < end) {
           conn.watch(inventory);
           if (!conn.sismember(inventory, itemId)){
               conn.unwatch();
               return false;
           }

           Transaction trans = conn.multi();
           trans.zadd("market:", price, item);
           trans.srem(inventory, itemId);
           List<Object> results = trans.exec();
           // null response indicates that the transaction was aborted due to
           // the watched key changing.
           if (results == null){
               continue;
           }
           return true;
       }
       return false;
   }
public boolean purchaseItem(
           Jedis conn, String buyerId, String itemId, String sellerId, double lprice) {

       String buyer = "users:" + buyerId;
       String seller = "users:" + sellerId;
       String item = itemId + '.' + sellerId;
       String inventory = "inventory:" + buyerId;
       long end = System.currentTimeMillis() + 10000;

       while (System.currentTimeMillis() < end){
           conn.watch("market:", buyer);

           double price = conn.zscore("market:", item);
           double funds = Double.parseDouble(conn.hget(buyer, "funds"));
           if (price != lprice || price > funds){
               conn.unwatch();
               return false;
           }

           Transaction trans = conn.multi();
           trans.hincrBy(seller, "funds", (int)price);
           trans.hincrBy(buyer, "funds", (int)-price);
           trans.sadd(inventory, itemId);
           trans.zrem("market:", item);
           List<Object> results = trans.exec();
           // null response indicates that the transaction was aborted due to
           // the watched key changing.
           if (results == null){
               continue;
           }
           return true;
       }

总结：相比于一般关系型数据库的悲观锁，redis的事务是典型的乐观锁，没有对事务进行封锁，以避免客户端运行过慢造成长时间的阻塞

非事务型流水线

使用流水线，减少通信次数提高性能，以jedis为例，对比使用和没使用流水线的函数方法调用次数：

public void updateTokenPipeline(Jedis conn, String token, String user, String item) {
    long timestamp = System.currentTimeMillis() / 1000;
    Pipeline pipe = conn.pipelined();
    pipe.multi();
    pipe.hset("login:", token, user);
    pipe.zadd("recent:", timestamp, token);
    if (item != null){
        pipe.zadd("viewed:" + token, timestamp, item);
        pipe.zremrangeByRank("viewed:" + token, 0, -26);
        pipe.zincrby("viewed:", -1, item);
    }
    pipe.exec();
}

//对比没有使用流水线的方法
public void updateToken(Jedis conn, String token, String user, String item) {
    long timestamp = System.currentTimeMillis() / 1000;
    conn.hset("login:", token, user);
    conn.zadd("recent:", timestamp, token);
    if (item != null) {
        conn.zadd("viewed:" + token, timestamp, item);
        conn.zremrangeByRank("viewed:" + token, 0, -26);
        conn.zincrby("viewed:", -1, item);
    }
}
//测试函数如下
 public void benchmarkUpdateToken(Jedis conn, int duration) {
    try{
        @SuppressWarnings("rawtypes")
        Class[] args = new Class[]{
            Jedis.class, String.class, String.class, String.class};
        Method[] methods = new Method[]{
            this.getClass().getDeclaredMethod("updateToken", args),
            this.getClass().getDeclaredMethod("updateTokenPipeline", args),
        };
        for (Method method : methods){
            int count = 0;
            long start = System.currentTimeMillis();
            long end = start + (duration * 1000);
            while (System.currentTimeMillis() < end){
                count++;
                method.invoke(this, conn, "token", "user", "item");
            }
            long delta = System.currentTimeMillis() - start;
            System.out.println(
                    method.getName() + ' ' +
                    count + ' ' +
                    (delta / 1000) + ' ' +
                    (count / (delta / 1000)));
        }
    }catch(Exception e){
        throw new RuntimeException(e);
    }
}

运行结果如图所示，在本地运行性能提升大概17.8倍。

tips：可以使用redis-benchmark工具进行性能测试。

五. References

《Redis实战》