执行lock操作，允许传递watch变量控制是否需要阻塞lock操作
978043
private Boolean acquireLock(final BooleanMutex mutex) {
        try {
            do {
                if (id == null) {// 构建当前lock的唯一标识
                    long sessionId = getSessionId();
                    String prefix = "x-" + sessionId + "-";
                    // 如果第一次，则创建一个节点
                    String path = zookeeper.create(root + "/" + prefix, data, CreateMode.EPHEMERAL_SEQUENTIAL);
                    int index = path.lastIndexOf("/");
                    id = StringUtils.substring(path, index + 1);
                    idName = new LockNode(id);
                }

                if (id != null) {
                    List<String> names = zookeeper.getChildren(root);
                    if (names.isEmpty()) {
                        logger.warn("lock lost with scene:empty list, id[] and node[]", id, idName);
                        unlock();// 异常情况，退出后重新创建一个
                    } else {
                        // 对节点进行排序
                        SortedSet<LockNode> sortedNames = new TreeSet<LockNode>();
                        for (String name : names) {
                            sortedNames.add(new LockNode(name));
                        }

                        if (sortedNames.contains(idName) == false) {
                            logger.warn("lock lost with scene:not contains ,id[] and node[]", id, idName);
                            unlock();// 异常情况，退出后重新创建一个
                            continue;
                        }

                        // 将第一个节点做为ownerId
                        ownerId = sortedNames.first().getName();
                        if (mutex != null && isOwner()) {
                            mutex.set(true);// 直接更新状态，返回
                            return true;
                        } else if (mutex == null) {
                            return isOwner();
                        }

                        SortedSet<LockNode> lessThanMe = sortedNames.headSet(idName);
                        if (!lessThanMe.isEmpty()) {
                            // 关注一下排队在自己之前的最近的一个节点
                            LockNode lastChildName = lessThanMe.last();
                            lastChildId = lastChildName.getName();
                            // 异步watcher处理
                            IZkConnection connection = zookeeper.getConnection();
                            // zkclient包装的是一个持久化的zk，分布式lock只需要一次性的watcher，需要调用原始的zk链接进行操作
                            ZooKeeper orginZk = ((ZooKeeperx) connection).getZookeeper();
                            Stat stat = orginZk.exists(root + "/" + lastChildId, new AsyncWatcher() {

                                public void asyncProcess(WatchedEvent event) {
                                    if (!mutex.state()) { // 避免重复获取lock
                                        acquireLock(mutex);
                                    } else {
                                        logger.warn("locked successful.");
                                    }
                                }

                            });

                            if (stat == null) {
                                acquireLock(mutex);// 如果节点不存在，需要自己重新触发一下，watcher不会被挂上去
                            }
                        } else {
                            if (isOwner()) {
                                mutex.set(true);
                            } else {
                                logger.warn("lock lost with scene:no less ,id[] and node[]", id, idName);
                                unlock();// 可能自己的节点已超时挂了，所以id和ownerId不相同
                            }
                        }
                    }
                }
            } while (id == null);
        } catch (KeeperException e) {
            exception = e;
            if (mutex != null) {
                mutex.set(true);
            }
        } catch (InterruptedException e) {
            interrupt = e;
            if (mutex != null) {
                mutex.set(true);
            }
        } catch (Throwable e) {
            other = e;
            if (mutex != null) {
                mutex.set(true);
            }
        }

        if (isOwner() && mutex != null) {
            mutex.set(true);
        }
        return Boolean.FALSE;
    }
371077
def __checkLock(self):
        
        if self._isLocked:
            if (time.time() - self._timer) > TIMEOUT_TIMES[self._operation]:
                self.__unLock()
                return False
            return True
        return False
780989
def raise_or_lock(self, key, timeout):
        
        acquired = Lock(
            self.redis,
            key,
            timeout=timeout,
            blocking=self.blocking,
            blocking_timeout=self.blocking_timeout
        ).acquire()

        if not acquired:
            # Time remaining in milliseconds
            # https://redis.io/commands/pttl
            ttl = self.redis.pttl(key)
            raise AlreadyQueued(ttl / 1000.)
323556
private void setAcquired(HLock lock, Map<String, String> canBeEarlier) {
    // start the heartbeat
    Future<Void> heartbeat = scheduler.schedule(new Heartbeat(lock), lockTtl / 2, TimeUnit.MILLISECONDS);

    ((HLockImpl) lock).setHeartbeat(heartbeat);

    ((HLockImpl) lock).setAcquired(true);

    if (logger.isDebugEnabled()) {
      logLock(lock, canBeEarlier.keySet());
    }

  }
210485
def _add_watch_block(self, params, filterCb, capacity=500, cb=None):
		
		if (self._wbslock == None):
			self._wbslock = Lock()

		self._wbslock.acquire()
		wb = WatchBlock(params, filterCb, capacity, cb)  # Add a List of Watchers
		if ((wb != None) and (wb.cb == None)):
			wb.cb = wb._dequeue_default_cb

		self._wbs.append(wb)
		self._wbslock.release()

		if self._cookie == None:
			return None
		if wb != None and len(self._wbs) == 1 and wb.params["pollSec"] == None:
			self._start_enqueue_thread()
		if self._enqueueThread == None:
			return wb

		self._enqueueThreadSignal.acquire()
		self._enqueueThreadSignal.notify()  # Notify
		self._enqueueThreadSignal.release()  # Release the Lock

		return wb
398941
private void watch(WatcherCallback watcherCallback){
        try {
            while (true) {
                final WatchKey key = watchService.take();
                if(key == null){
                    continue;
                }
                for (WatchEvent<?> watchEvent : key.pollEvents()) {
                    final WatchEvent.Kind<?> kind = watchEvent.kind();
                    //忽略无效事件
                    if (kind == StandardWatchEventKinds.OVERFLOW) {
                        continue;
                    }
                    final WatchEvent<Path> watchEventPath = (WatchEvent<Path>) watchEvent;
                    //path是相对路径（相对于监控目录）
                    final Path contextPath = watchEventPath.context();
                    LOGGER.info("contextPath:"+contextPath);
                    //获取监控目录
                    final Path directoryPath = directories.get(key);
                    LOGGER.info("directoryPath:"+directoryPath);
                    //得到绝对路径
                    final Path absolutePath = directoryPath.resolve(contextPath);
                    LOGGER.info("absolutePath:"+absolutePath);
                    LOGGER.info("kind:"+kind);
                    //判断事件类别
                    switch (kind.name()) {
                        case "ENTRY_CREATE":
                            if (Files.isDirectory(absolutePath, LinkOption.NOFOLLOW_LINKS)) {
                                LOGGER.info("新增目录：" + absolutePath);
                                //为新增的目录及其所有子目录注册监控事件
                                registerDirectoryTree(absolutePath);
                            }else{
                                LOGGER.info("新增文件：" + absolutePath);
                            }
                            break;
                        case "ENTRY_DELETE":
                            LOGGER.info("删除：" + absolutePath);
                            break;
                        case "ENTRY_MODIFY":
                            LOGGER.info("修改：" + absolutePath);
                            break;
                    }
                    //业务逻辑
                    watcherCallback.execute(kind, absolutePath.toAbsolutePath().toString());
                }
                boolean valid = key.reset();
                if (!valid) {
                    if(directories.get(key) != null){
                        LOGGER.info("停止监控目录："+directories.get(key));
                        directories.remove(key);
                    }
                }
            }
        } catch (InterruptedException ex) {
            LOGGER.info("监控目录线程退出");
        } finally{
            try {
                watchService.close();
                LOGGER.info("关闭监控目录服务");
            } catch (IOException ex) {
                LOGGER.error("关闭监控目录服务出错", ex);
            }
        }
    }
876296
def write_lock(self, last_update=datetime.datetime.now()):
        
        data = {
            "last_update": last_update.strftime(AppCronLock.DATETIME_FORMAT)
        }

        with open(self.lock_file, "w") as writer:
            json.dump(data, writer)
559939
def handle_lock(handle):
    
    
    def wrapper(self, *args, **options):
        def on_interrupt(signum, frame):
            # It's necessary to release lockfile
            sys.exit()
        signal.signal(signal.SIGTERM, on_interrupt)

        start_time = time.time()
        try:
            verbosity = int(options.get('verbosity', 0))
        except ValueError:
            verbosity = 0
        logger = logging.getLogger(self.__module__)
        if verbosity == 0:
            logger.level = logging.WARNING
        elif verbosity == 1:
            logger.level = logging.INFO
        else:
            logger.level = logging.DEBUG
       
        logger.debug("-" * 72)
        
        lock_name = self.__module__.split('.').pop()
        lock = FileLock(os.path.join(LOCK_ROOT, lock_name))
        
        logger.debug("%s - acquiring lock..." % lock_name)
        try:
            lock.acquire(LOCK_WAIT_TIMEOUT)
        except AlreadyLocked:
            logger.debug("lock already in place. quitting.")
            return
        except LockTimeout:
            logger.debug("waiting for the lock timed out. quitting.")
            return
        logger.debug("acquired.")
        
        try:
            handle(self, logger, *args, **options)
        except (KeyboardInterrupt, SystemExit):
            pass
        except:
            import traceback
            logging.warn("Command Failed")
            logging.warn('=' * 72)
            logging.warn(traceback.format_exc())
            logging.warn('=' * 72)
        
        logger.debug("releasing lock...")
        lock.release()
        logger.debug("released.")
        
        logger.info("done in %.2f seconds" % (time.time() - start_time))
        return
        
    return wrapper
972850
def execute_watch(self, id=None, body=None, params=None):
        
        return self.transport.perform_request(
            "PUT",
            _make_path("_watcher", "watch", id, "_execute"),
            params=params,
            body=body,
        )
724262
def lock(self):
        """"""
        try:
            self._do_lock()
            return
        except LockError:
            time.sleep(self.TIMEOUT)

        self._do_lock()
1074300
public static function ulock($lock, $timeout = 30, &$wait = null) {
        try {
            $redis = RedisClient::getRedis();
            if ($redis) {
                $start = time();
                $cnt   = $redis->incr($lock);
                while ($cnt != 1) {
                    $wait = true;//等了
                    usleep(200);
                    if ((time() - $start) > $timeout) {
                        break;//超时
                    }
                    if (!$redis->exists($lock)) {//锁释放方式为删除
                        $cnt = $redis->incr($lock);
                    }
                }
                if ($cnt == 1) {
                    $redis->setTimeout($lock, $timeout * 2);//设置超时，防止死锁

                    return true;
                }
            }
        } catch (\Exception $e) {
            log_warn($e->getMessage(), 'redis_lock');
        }

        return false;
    }
844338
function() {
			if(index < count) {
				inspectors[index++].inspect(this); // 执行递归，进行下一轮检查
			} else if(index++ === count) {
				ctrler.execute(); // 当inspector 被执行完时，执行 controller
				end();
			}
		}
127305
function Lock(target, name, descriptor) {
    let fn = descriptor.value;
    let $$LockIsDoing = false;

    let reset = ()=>$$LockIsDoing=false;
    descriptor.value = function(...args) {
        if ($$LockIsDoing) return;
        $$LockIsDoing = true;

        let ret = fn.apply(this, args);

        if (ret && ret.then) {
            // is promise
            return ret.then((succ)=>{
                reset();
                return Promise.resolve(succ);
            }, (fail)=>{
                reset();
                return Promise.reject(fail);
            });
        } else {
            reset();
            return ret;
        }
    };

    return descriptor;
}
913894
def run(self):
        

        stop_watcher = threading.Event()
        watcher = None
        if self.watch_func:
            watcher = threading.Thread(
                target=self.watch_func,
                args=(self.stack, stop_watcher)
            )
            watcher.start()

        try:
            while not self.done:
                self._run_once()
        finally:
            if watcher:
                stop_watcher.set()
                watcher.join()
        return self.ok
962339
function lock(from, to) {
  from = fis.file.wrap(from).realpath;
  to = fis.file.wrap(to).realpath;

  lockedMap[to] = from;
}
1279573
def start(self, timer=True):
        
        # 将引擎设为启动
        self.__active = True
        
        # 启动事件处理线程
        self.__thread.start()
        
        # 启动计时器，计时器事件间隔默认设定为1秒
        if timer:
            self.__timer.start(1000)
777111
public final void incrementUnlockCount(long tick)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "incrementUnlockCount", Long.valueOf(tick));

    _targetStream.setCursor(tick);
    TickRange tickRange = _targetStream.getNext();
    if (tickRange.type == TickRange.Value)
    {
      AIValueTick valueTick = (AIValueTick) tickRange.value;
      if (valueTick != null)
        valueTick.incRMEUnlockCount();
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "incrementUnlockCount", tickRange);
  }