缓存中间件-Redis(二)
阅读原文时间:2022年04月15日阅读:1

在上一篇中我们简单总结和介绍了Redis的几个方面

1.使用Redis背景

2.Redis通信多路复用的基本原理

3.Redis基本数据结构

4.Redis持久化方式

这一篇我们使用简单的业务场景来介绍Redis的分布式锁集群

1.分布式锁

首先我们应该知道什么是分布式锁,用来做什么的,解决了什么问题,我们应该怎么做?

简单来说,分布式锁就是锁住进程的锁,通常我们分布式锁的使用场景在多实例集群系统中,为了防止多个进程对资源的同时操作,而引发业务逻辑错误而诞生的,和我们的单个实例中的lock作用一样,一个是针对多实例单进程,一个是针对单实例多线程,具体在web中的表现在下面用实例的方式说明。

2.单进程单实例多线程

1.例如现在我们部署了一个服务,他的功能是提供用户来抢红包的接口,我们在发红包时,总共设置了10个红包,我们将红包剩余数存入数据库,所以用一张表来存储红包的个数信息,每当有一个人来抢,我们就需要将红包的个数减掉1,主要的核心业务逻辑如下:

  • 1.用户发起抢红包请求,服务器首先查询,红包剩余个数,如果为0,则直接返回,告诉用户红包抢完了

  • 2.如果红包个数不为0,则提示用户抢到了红包,然后扣减红包总数。

伪代码如下

 //查询剩余红包个数
 int RedEnvelopeCount = getRedEnvelopes();
 if (RedEnvelopeCount.Count == 0)
 {
     //返回用户红包抢完了 do something
     return;
 }
 //通知用户抢到了 do something
 //扣减红包个数
 subtracRedEnvelopeCount();

2.此时如果同一时间多个并发请求进来,我们就会发生多个线程同时执行这段核心逻辑,导致业务错误

  • 1.发生超抢,抢到的人大于实际个数,谁抢到的不算?

  • 2.发生少抢,几个人同时抢到同一个,算谁的?

我相信这2种情况,要是在真实的抢红包中,发生到谁头上,都不爽吧?按东哥的话说我都气的想打人

所以我们要着手解决,解决的思路就是,核心业务的代码同一时刻只能被一个线程执行,所以为了保证原子性,我们就需要使用到来锁住核心业务代码,在这里我们使用C#的Lock来实现

//静态全局唯一
private readonly static object _lock = new object();
using (_lock)
{
    int RedEnvelopeCount = getRedEnvelopes();
    if (RedEnvelopeCount.Count == 0)
    {
        //返回用户红包抢完了
        return;
    }
    //通知用户抢到了
    //扣减红包个数
    subtracRedEnvelopeCount();
}
3.单进程多实例多线程

上面我们列举并介绍了在单个实例中遇到并发产生的业务问题,我们通过锁的方式来暂时解决了问题,那么接下来我们继续思考,如果红包再多一点,抢的并发量起来了,单台服务器不能承载这么多并发,为了保证高可用,我们需要将红包服务,进行集群部署,利用nginx来负载均衡一下。

1.在单个实例中我们为了保证业务执行的原子性,锁住了核心业务代码,保证同一时刻只有一个线程执行,但是现在多个实例来处理请求,假设现在2个用户提交抢红包的请求,在同一时刻被服务器负载均衡到2台不同的实例,那么此时就会遇到2个进程同时操作核心业务逻辑,超抢少抢的事情又会发生,瞬间头又大了,那么针对这种业务情况,我们应该使用分布式锁来解决不同进程之间争夺资源的情况。

2.接下来就是我们最后一个问题,如何实现分布式锁,其实对于集群系统实现分布式锁的方式有很多种,基于Zookeeper,Redis,包括数据库,等都可以实现,相较于Zookeeper和数据库 ,Redis单线程对于实现分布式锁拥有天生绝对的优势,而且它操作内存的方式就性能而言是非常优秀的,鉴于咱们这篇博客主要是跟Redis相关的,所以主要介绍使用Redis的方式,当然基于Redis中也存在不同的做法。

3.实现锁的核心业务就是加锁``释放锁这2个功能,我们根据Redis内部单线程执行的原理自己来初步实现一个

 //设置key,加锁 防止死锁超时,设置60毫秒自动过期
 bool result = database.StringSet("distributed-lock", "lock", expiry: TimeSpan.FromSeconds(60));
 //失败说明锁已经被获取,直接返回
 if (!result)
 {
     // "人有点多,再点点...";
     return;
 }
 //获取剩余个数
 int RedEnvelopeCount = getRedEnvelopes();
 if (RedEnvelopeCount.Count == 0)
 {
     //返回用户红包抢完了
     return;
 }
 //通知用户抢到了
 //扣减红包个数
 subtracRedEnvelopeCount();
 //释放锁
database.KeyDelete("distributed-lock");

上面实现了最简单和最基础的分布式锁,但是还存在缺陷,仔细思考不难发现

  • 1.执行业务的时间大于持有锁的最大时间,导致锁释放,业务还没执行完,乱了那也玩不了

  • 2.获取锁后执行到业务代码时抛异常了,锁没及时释放,而是等待自动释放,所以需要加异常处理

我们根据这2个缺陷来进一步完善,解决上面的问题,首先需要先保证执行业务的情况下,谁加的锁由谁来释放,然后添加异常处理代码用于处理业务异常能及时释放锁

//获取Guid作为当前请求唯一身份信息
string value = Guid.NewGuid().ToString("N");

//设置key,加锁,防止死锁超时,设置60毫秒自动过期
bool result = database.StringSet("distributed-lock", value, expiry: TimeSpan.FromSeconds(60));

//失败说明锁已经被获取,直接返回
if (!result) return;  // "人有点多,再点点...";

try
{
    int RedEnvelopeCount = getRedEnvelopes(); //获取剩余个数
    if (RedEnvelopeCount.Count == 0)return; //返回用户红包抢完了

    //通知用户抢到了
    //扣减红包个数
    subtracRedEnvelopeCount();
}
finally
{
    //保证由当前请求释放,不会别的进来释放导致乱套
    if (value.Equals(database.StringGet("distributed-lock")))
    {
        //释放锁
        bool isDelete = database.KeyDelete("distributed-lock");
    }
}

4.最后我们使用StackExchange.Redis提供的方法来实现封装一个分布式锁,首先扩展IServiceCollection 将StackExchange.Redis 操作实例注入

//扩展连接客户端
public static class RedisServiceCollectionExtensions
{
    public static IServiceCollection AddRedisCache(this IServiceCollection services, string connectionString)
    {
        ConfigurationOptions configuration = ConfigurationOptions.Parse(connectionString);
        ConnectionMultiplexer connectionMultiplexer = ConnectionMultiplexer.Connect(connectionString);
        services.AddSingleton(connectionMultiplexer);
        return services;
    }
}
//注入容器
services.AddRedisCache("127.0.0.1:6379,password=xxx,connectTimeout=2000");

5.具体利用LockTake通过轮训询问的方式获取锁,通过LockRelease释放锁,在实现的源码中,加锁的原理就是设置一个string类型的key value以及过期时间

public class DistributedLock
{
    private readonly ConnectionMultiplexer _connectionMultiplexer;
    private IDatabase db = null;
    public DistributedLock(ConnectionMultiplexer connectionMultiplexer)
    {
       _connectionMultiplexer = connectionMultiplexer;
       db = connectionMultiplexer.GetDatabase(0);
    }

    private void FetchLock()
    {
      while (true)
       {
          //当前锁的持有时间超过60毫秒就会被释放,这个时间应该设置比被锁住业务执行时间长
          bool flag = db.LockTake("distributed-lock", Thread.CurrentThread.ManagedThreadId, TimeSpan.FromSeconds(60));
          if (flag)  // 1、如果加锁成功直接退出,否则继续获取锁
          {
             break;
          }
           Thread.Sleep(10); // 防止死循环。通过等待10毫秒时间,释放资源
       }
   }

  public void UnLock()
  {
     bool flag = db.LockRelease("distributed-lock", Thread.CurrentThread.ManagedThreadId);
     _connectionMultiplexer.Close();
  }

}

6.基于Redis实现的分布式锁,在Redis主从或者哨兵乃至集群中,会存在锁丢失的情况,具体就是当请求给Master节点写入锁后,Master节点宕机,此时主从还没来得及同步,但此时已经选择了一个Slaver为主节点,而最开始的设置的锁不见了,针对这种情况比较极端,可以利用Zookeeper来实现分布式锁作为解决方案,因为在数据一致性上Zookeeper是优于Redis的,当然Redis的性能也是较Zookeeper更为突出一些。

4.Redis容灾高可用

就服务实例而言,单个服务实例保证容灾高可用基本很难,因为一个实例随时都有宕机的风险,所以引入了集群的概念,通过多个服务实例,来共同保证系统提供稳定的服务,其实在Redis中也一样,也有保证容灾和高可用的方案,在一定程度上,实现的方案在大的策略层面思路都差不多,只是具体实现的方式不同罢了,至于最终选择保证数据信息层面的高可用,还是性能资源层面的高可用,需要用户去根据自己业务来分析,选择使用哪一种方案。

在Redis中,有3种方式来实现高可用的方案,分别是

  • 1.主从模式 (Master Slaver)

  • 2.哨兵模式 (Sentinel)

  • 3.数据分片 (Cluster)

1.主从模式和哨兵

主从模式是Redis提供高可用的第一种方案,理解起来也比较简单,一个主节点多个从节点,主节点负责写入,然后会把数据同步到从节点,从节点负责读取,但是会存在不可避免的缺陷

  • 1.如果宕机,它不具备自动容错和恢复功能,需要手动切换从节点为主节点。

  • 2.宕机导致数据未及时同步,降低可用性

  • 3.所有节点数据一样,内存浪费,并且在线扩容复杂,伸缩性差,海量数据存储无法解决。

  • 4.高并发主节点写入数据压力极大。

哨兵模式是Redis提供的第二种方案,可以说是第一种主从方案的完善版本,因为它在主从的基础上,增加了监控功能,主节点宕机后提供哨兵自动选举master并将其他的slave指向新的master,哨兵是一个独立运行的进程,它的实现原理是哨兵进程向所有的redis机器发送心跳消息,从而监控运行状况,同样由于是基于主从而诞生,那么也存在主从中的一些缺陷。

2.集群分片模式

1.集群和分片基本概念

前面2种方案归根结底,就是使用主从复制和读写分离的方式保证高可用,但是内存占用,主节点压力过大,相较于Redis提供的第三种集群和数据分片的方案,存在的缺陷也非常明显,下面我们介绍集群和数据分片

1.集群和数据分片简单理解就是将主从复制的核心策略,做了集群来横向扩展,包含多个主从节点,采用去中心化思想,数据按照 slot 存储分布在多个节点,并且节点间数据共享,可动态调整数据分布,

2.在集群中每一个节点都是互相通信采用 Gossip 病毒协议,在一个节点写入的数据,其他任何节点都会查询到。

2.集群内部数据存取原理

1.Slot 槽,存储数据的容器,总共16384个Slot ,采用平均分配,并且只有主节点才能被分配

2.Hash算法用于对数据进行计算,得到一个Hash值,然后对16384取模选择存到哪个区间的Slot

3.Windows下集群搭建

1.我们部署集群使用redis-trib.rb,首先下载环境 提取码1234

2.安装需要用到Ruby语言运行环境,然后进入安装目录Ruby22-x64中,通过gem命令安装文件

gem install –-local youFloder\redis-3.2.2.gem

3.准备Redis服务实例,此处准备6个实例,并配置Redis配置文件,然后启动每一个服务实例

redis-server redis-6380.conf
redis-server redis-6381.conf
....
....


port 6380 #节点端口
bind 127.0.0.1   #节点主机ip
appendonly yes 开启aof
appendfilename "appendonly.6380.aof"
cluster-enabled yes
cluster-config-file nodes.6380.conf
cluster-node-timeout 15000
cluster-slave-validity-factor 10
cluster-migration-barrier 1
cluster-require-full-coverage yes

3.进入到redis-trib.rb目录打开命令窗口,进行搭建集群

ruby redis-trib.rb create -–replicas 1 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384 127.0.0.1:6385

4.检查集群状态,出现图片中的提示代表搭建成功

ruby redis-trib.rb check 127.0.0.1:6380