Redis的高级特性与应用场景(二)

憧憬。 redis, 服务端技术栈

Redis的高级特性与应用场景(二)

利用Pipeline管道处理多个命令

Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。
这意味着通常情况下一个请求会遵循以下步骤：

  • 客户端向服务端发送一个查询请求，并监听Socket返回，通常是以阻塞模式，等待服务端响应。
  • 服务端处理命令，并将结果返回给客户端。

一个命令的发送到处理,是需要往返时间的,如果是本地回环网络的话还会比较快,可如果是外网的话经常层层网络代理就不一定能接受了。管道的话可以一次将多个命令发送到服务器，而不用等待答复，最后在一个步骤中读取该答复。

例子

我们可以在批量插入和批量获取的时候使用,例如下面: 我批量获取了直播相关的数据

Redis::connection('cache')->pipeline(function ($pipe) use ($fn) {
    foreach ($data() as $v) {
        $likeKeyArr[] = sprintf(LiveRooms::REDIS_LIVE_LIKE_PREFIX, $v['id'], $v['start_live_time']);
        $commentKeyArr[] = sprintf(LiveRooms::REDIS_LIVE_COMMENT_PREFIX, $v['id'], $v['start_live_time']);
        $shareKeyArr[] = sprintf(LiveRooms::REDIS_LIVE_SHARE_PREFIX, $v['id'], $v['start_live_time']);
    }

    $pipe->mget($likeKeyArr);
    $pipe->mget($commentKeyArr);
    $pipe->mget($shareKeyArr);
})

Redis键空间通知

这个特性可以让我们订阅redis的操作,例如在redis中设置好key的生存时间后,希望key过期被删除后能给发一个通知

del key

例如上面删除了一个键, redis 会发送两种不同类型的数据,特定的事件会往特定的频道发送通知,我们只要订阅这个特定的频道等待通知即可.

PUBLISH __keyspace@0__:key del # 键空间通知
PUBLISH __keyevent@0__:del key # 键事件通知

我们可以只启用其中一种通知，以便只传递我们感兴趣的事件子集。

注意: 事件使用Redis的普通发布/订阅层传递,由于Redis的发布/订阅是fire and forget，因此如果你的应用要求可靠的事件通知，目前还不能使用这个功能，也就是说，如果你的发布/订阅客户端断开连接，并在稍后重连，那么所有在客户端断开期间发送的事件将会丢失。

例子

我们可以监听0库里面键过期的事件

<?php

class RedisInstance
{
    private $redis;

    public function __construct($host = '127.0.0.1', $port = 6379)
    {
        $this->redis = new Redis();
        $this->redis->connect($host, $port);
    }

    public function expire($key = null, $time = 0)
    {
        return $this->redis->expire($key, $time);
    }

    public function psubscribe($patterns = array(), $callback)
    {
        $this->redis->psubscribe($patterns, $callback);
    }

    public function setOption()
    {
        $this->redis->setOption(\Redis::OPT_READ_TIMEOUT,-1);
    }

}

echo "程序开始执行..\n";
$redis = new RedisInstance();
$redis->setOption();
$redis->psubscribe(array('__keyevent@0__:expired'), 'callback');
//回调
function callback($redis, $pattern, $chan, $msg)
{
    echo "$pattern\n";
    echo "$chan\n";
    echo "$msg\n";
      /*业务逻辑*/
}

不支持rollback的事务

redis事务与关系型数据库事务不太一样,它的事务不支持回滚,这也使得redis的事务处理效率特别高,但是这个不支持rollback是不是会造成我们的数据混乱呢,这样的事务是不是没有意义呢?

redis的事务是不保证原子性的: redis事务只保证在命令格式只有在都正确的情况下才会都执行，要不就都不执行命令。但是事务的整体是不保证原子性的，且没有回滚，当事务中任意一个命令执行失败，其余的命令依然会执行。

redis事务是将所有的命令发送到队列里面,最终exec才进行执行,redis的命令只会因为语法而失败,或是命令用在了错误类型键上面. 这也就是说,失败命令是由编程造成的,而这些错误应该在开发过程中被发现,而不应该出现在生产环境中. 鉴于更多的问题都是程序员自身的问题,redis直接采用无回滚方式来处理事务

乐观锁例子

我们津津乐道的转账问题,就可以利用事务来处理.

场景:

A余额100元
B余额100元
C余额100元

A准备给B转账50,再同时C要还50元给A,那么A这个余额怎么确定在转账完之后操作还是在转账前操作呢?这属于资源竞争,常见方式就是加锁了,排它锁的话比较消耗资源,我们可以利用watch来实现乐观锁.

当watch的key value发生改变的时候,exec事务会取消, 当 exec 被调用后, 所有的keys都将UNWATCH，不管这个事务会不会终止。

set A 100
set B 100
set C 100


watch A
multi
decrby A 50
incrby B 50

# 在exec之前可以启用第二个客户端,对A账号减少50元,查看watch乐观锁机制是否生效

exec   # 这里就会返回 <nil> 因为事务没有执行
get A  # 第二个客户端转账50元 所以最终为 150

分布式锁

锁的机制也是一个热门话题,不同的进程必须以独占资源的方式实现资源共享

首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

  • 互斥性。在任意时刻，只有一个客户端能持有锁。
  • 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
  • 具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。
  • 解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了

下面找到一个例如大家可以看看: https://github.com/ronnylt/redlock-php

<?php

class RedLock
{
    private $retryDelay;
    private $retryCount;
    private $clockDriftFactor = 0.01;

    private $quorum;

    private $servers = array();
    private $instances = array();

    function __construct(array $servers, $retryDelay = 200, $retryCount = 3)
    {
        $this->servers = $servers;

        $this->retryDelay = $retryDelay;
        $this->retryCount = $retryCount;

        $this->quorum  = min(count($servers), (count($servers) / 2 + 1));
    }

    public function lock($resource, $ttl)
    {
        $this->initInstances();

        $token = uniqid();
        $retry = $this->retryCount;

        do {
            $n = 0;

            $startTime = microtime(true) * 1000;

            foreach ($this->instances as $instance) {
                if ($this->lockInstance($instance, $resource, $token, $ttl)) {
                    $n++;
                }
            }

            # Add 2 milliseconds to the drift to account for Redis expires
            # precision, which is 1 millisecond, plus 1 millisecond min drift
            # for small TTLs.
            $drift = ($ttl * $this->clockDriftFactor) + 2;

            $validityTime = $ttl - (microtime(true) * 1000 - $startTime) - $drift;

            if ($n >= $this->quorum && $validityTime > 0) {
                return [
                    'validity' => $validityTime,
                    'resource' => $resource,
                    'token'    => $token,
                ];

            } else {
                foreach ($this->instances as $instance) {
                    $this->unlockInstance($instance, $resource, $token);
                }
            }

            // Wait a random delay before to retry
            $delay = mt_rand(floor($this->retryDelay / 2), $this->retryDelay);
            usleep($delay * 1000);

            $retry--;

        } while ($retry > 0);

        return false;
    }

    public function unlock(array $lock)
    {
        $this->initInstances();
        $resource = $lock['resource'];
        $token    = $lock['token'];

        foreach ($this->instances as $instance) {
            $this->unlockInstance($instance, $resource, $token);
        }
    }

    private function initInstances()
    {
        if (empty($this->instances)) {
            foreach ($this->servers as $server) {
                list($host, $port, $timeout) = $server;
                $redis = new \Redis();
                $redis->connect($host, $port, $timeout);

                $this->instances[] = $redis;
            }
        }
    }

    private function lockInstance($instance, $resource, $token, $ttl)
    {
        return $instance->set($resource, $token, ['NX', 'PX' => $ttl]);
    }

    private function unlockInstance($instance, $resource, $token)
    {
        $script = '
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
        ';
        return $instance->eval($script, [$resource, $token], 1);
    }
}

$servers = [
    ['127.0.0.1', 6379, 0.01],
    ['127.0.0.1', 6389, 0.01],
    ['127.0.0.1', 6399, 0.01],
];
$redLock = new RedLock($servers);
while (true) {
    $lock = $redLock->lock('test', 10000);
    if ($lock) {
        print_r($lock);
    } else {
        print "Lock not acquired\n";
    }
}

这个是redlock,但是这个也不能保证线程安全,进程由于各种原因pause,类似于上文说的多线程间的时间片切换,比如由于GC停顿等导致锁过期,但是进程并未感知到,同时另一个进程已经获取了该分布式锁,就会导致奇怪的结果发生.

这里有说明为什么redlock不安全原因: https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html 有兴趣的可以阅读一下

但是这种开源可以让我们更好的学习,对于数据要求强一致性的使用 redlock 还是需要慎重, 不推荐使用

您的支持是对我最大的鼓励！

发表于： 作者：憧憬。
关注互联网以及分享全栈工作经验的原创个人博客和技术博客，热爱编程，极客精神
Github 新浪微博 SegmentFault 掘金专栏