基于Redis分布式锁的任务调度怎么实现

这篇“基于Redis分布式锁的任务调度怎么实现”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“基于Redis分布式锁的任务调度怎么实现”文章吧。

在分布式大批量数据采集过程中，信源的管理尤为重要。为保证同一任务在同一时间，只能被一个采集器处理，必须保证任务调度的唯一性。通常我们在进行分布式数据采集时，一般情况下都会有一个调度模块，其主要的职责就是负责采集任务的分发，同时保证任务的唯一性。

由于是分布式，涉及到多台服务器(多机)，每台服务器又涉及到多个采集器(多进程)，每个采集器又有可能涉及到多线程，所以，任务调度模块中的锁机制显得尤为重要。一般情况下，锁的实现方式，按照应用的实现架构，可能会有以下几种类型：

• 如果处理程序是单进程多线程的，在 python下，就可以使用 threading 模块的 Lock 对象来限制对共享变量的同步访问，实现线程安全。

• 单机多进程的情况，在 python 下，可以使用 multiprocessing 的 Lock 对象来处理。

• 多机多进程部署的情况，就得依赖一个第三方组件（存储锁对象）来实现一个分布式的同步锁了。

由于调度模块是多机多进程多线程的处理机制，所以符合第三种方式。

分布式锁实现方式

目前主流的分布式锁实现方式有以下几种：

• 基于数据库来实现，如 mysql

• 基于缓存来实现，如 redis

• 基于 zookeeper 来实现

每种实现方式各有千秋，综合考量，Redis是最为合适的选择。主要原因是：

• redis 是基于内存来操作，存取速度比数据库快，在高并发下，加锁之后的性能不会下降太多

• redis 可以设置键值的生存时间（TTL）

• redis 的使用方式简单，总体实现开销小

但是，使用 redis 实现的分布锁还需要具备以下几个条件：

1. 同一个时刻只能有一个线程占有锁，其他线程必须等待直到锁被释放

2. 锁的操作必须满足原子性

3. 不会发生死锁，例如已获得锁的线程在释放锁之前突然异常退出，导致其他线程会一直在循环等待锁被释放

4. 锁的添加和释放必须由同一个线程来设置

我们使用 redis 来实现一个分布式同步锁，来保证数据的一致性，需满足一下特点：

• 满足互斥性，同一个时刻只能有一个线程可以获取锁

• 利用 redis 的 ttl 来确保不会出现死锁，但同时也会带来由于锁过期引发的多线程同时占有锁的问题，需要我们合理设置锁的过期时间来避免

• 利用锁的唯一性来确保不会出现误删锁的情况

我在实际操作过程中，把调度模块从整个采集系统中拆离了出来，基于Java客户端Jredis（JRedis是一个高性能的Java客户端,用来连接到Redis分布式哈希键-值数据库。提供同步和异步）+SpringBoot，实现了一个独立的服务。以便其他各个采集器，通过HTTP方式请求所要处理的采集任务。其处理过程大致如下：

• 采集器通过HTTP方式，向调度中心发送任务请求；

• 调度中心判断锁是否存在，如果存在则直接返回空集合；

• 如果不存在锁，则对请求加锁，然后根据信源规则获取相应的采集任务；

• 返回获取到的任务(如果没有待处理任务，则返回空)，然后删除锁。

调度模块的代码实现，大致如下所示：

public static List<Object> fetchTask(String lockKeyValue, RedisHashUtils redisHashUtils, HttpServletRequest request,

HashServiceInterface hif, ZSetServiceInterface zScoreSet, String dicName) {

List<Object> result = new ArrayList<Object>();

     try {

         String dicNameLock = "Dispatcher_Task_Lock";// 任务调度锁；

         if (!redisHashUtils.keyIsExit(dicNameLock, lockKeyValue)) {// 判断锁是否存在

         // 添加锁(把任务唯一性标识写入记录)；

          redisHashUtils.addOneData(dicNameLock, lockKeyValue,

            DateUtil.getYMDHMS());

         // 处理任务逻辑

        ..............................................

         // 删除锁（任务唯一性标识）；

         hsdi.remove(redisHashUtils, dicNameLock, lockKeyValue);

         } else {

            //锁已存在

         System.out.println("正在处理任务，暂时返回空集合....");

         }

     } catch (

     Exception e) {e.printStackTrace();

     }

return result;

}

在实际的操作过程中，在进行锁添加时，必须要给锁加上过期时间，否则出现某些不可知的异常时，可能会导致锁无法释放，采集器一直无法获取到采集任务的情况。

以上就是关于“基于Redis分布式锁的任务调度怎么实现”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注亿速云行业资讯频道。