微信登录

分布式锁

分布式锁

设定work_test方法为分布式锁，定时50秒内不可重复

  1. def t_work(self):
  2.     info = "每月定时任务"
  3.     global evn
  4.     flag = is_run("work"+evn, 50, "work"+evn, info)
  5.     if flag == "run":
  6.         print("work"+"跑了一次")
  7.         work("?")
  8.     if flag == "running":
  9.         pass
  11. def is_run(key, time, data, info):
  12.     s = select_0_M()
  13.     value = s.set_nx_true(key, data, time, info)
  14.     if value == 1 :
  15.         return "run"
  16.     if value == 0:
  17.         return "running"
  1. class select_0_M():
  2.     # 分布式锁
  3.     def set_nx_true(self, key, value, time, info):
  4.         try:
  5.             data = myRedis.set(key, value, ex=time, nx=True,)
  6.             return data
  7.         except Exception as e:
  8.             logger.error(info+'缓存数据设置失败，' + str(e))
分布式锁