首页 > 文章列表 > 实现任务调度的Redis分布式锁机制

实现任务调度的Redis分布式锁机制

Redis
396 2024-04-05

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

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

  • 如果处理程序是单进程多线程的,在 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;

}

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