Java 类com.hazelcast.core.ILock 实例源码

public void disposeLock(Lock lock) {
    switch (Settings.INSTANCE.clusteringMode()) {
    case HAZELCAST:
        ((ILock) lock).destroy();
        break;

    case IGNITE:
        break;

    case SINGLE:
        break;

    default:
        break;
    }
}

@Override
protected void doRunInitialization(final String beanName) {
    LockKey lockKey = new LockKey(KeyType.INITIALIZATION, beanName);
    // Try to get the initialization lock
    ILock lock = hazelcastInstance.getLock(lockKey.toString());
    if (lock.tryLock()) {
        // No one else is trying to run this initialization right now. Check if it was already ran by someone else
        if (!initializationControl.containsKey(beanName)) {
            try {
                // This initialization was never executed. Run it and mark it as executed
                super.doRunInitialization(beanName);
                initializationControl.put(beanName, beanName);
            } finally {
                HazelcastHelper.release(lock);
            }
        }
    } else {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Not running initialization for bean " + beanName + " because some other node is currently running it");
        }
    }
}

@Override
protected boolean doRunPollingTask(final String key, final Callable<Boolean> task) {
    LockKey lockKey = new LockKey(KeyType.POLLING_TASK, key);
    ILock lock = hazelcastInstance.getLock(lockKey.toString());
    // Ensure multiple nodes can't run a polling task simultaneously
    if (lock.tryLock()) {
        try {
            return super.doRunPollingTask(key, task);
        } finally {
            HazelcastHelper.release(lock);
        }
    } else {
        // Force a sleep, as couldn't get the lock for this polling task
        if (LOG.isDebugEnabled()) {
            LOG.debug("Some other cluster node is running the " + key + " polling task. Leaving.");
        }
        return false;
    }
}

private void acquire(final LockKey key) {
    if (acquiredLocks.containsKey(key)) {
        // Already own the lock
        return;
    }
    ILock lock = hazelcastInstance.getLock(key.toString());
    try {
        if (lock.tryLock(timeoutSeconds, TimeUnit.SECONDS)) {
            acquiredLocks.put(key, lock);
        } else {
            throw new LockingException();
        }
    } catch (InterruptedException e) {
        throw new LockingException(e);
    }

}

@Test
public void testLockUnlock() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    final ILock lock = hClient.getLock("testLockUnlock");
    lock.lock();
    final CountDownLatch latch = new CountDownLatch(1);
    final CountDownLatch unlockLatch = new CountDownLatch(1);
    new Thread(new Runnable() {
        public void run() {
            assertFalse(lock.tryLock());
            unlockLatch.countDown();
            lock.lock();
            latch.countDown();
        }
    }).start();
    assertTrue(unlockLatch.await(10, TimeUnit.SECONDS));
    lock.unlock();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
}

@Test
public void testTryLock() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    final ILock lock = hClient.getLock("testTryLock");
    assertTrue(lock.tryLock());
    lock.lock();
    final CountDownLatch latch = new CountDownLatch(1);
    final CountDownLatch unlockLatch = new CountDownLatch(1);
    new Thread(new Runnable() {

        public void run() {
            assertFalse(lock.tryLock());
            unlockLatch.countDown();
            try {
                assertTrue(lock.tryLock(10, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            latch.countDown();
        }
    }).start();
    assertTrue(unlockLatch.await(10, TimeUnit.SECONDS));
    lock.unlock();
    lock.unlock();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
}

protected com.hazelcast.core.ILock acquireDistributedLock(Object object) {
    if (object == null) {
        if (log.isWarnEnabled()) {
            log.warn("Could not acquire distributed lock, object is null");
        }
        return null;
    }

    if (log.isDebugEnabled()) {
        log.debug(String.format("Acquiring distributed lock for %s...", object.getClass().getSimpleName()));
    }
    ILock lock = getHazelcastInstance().getLock(object);
    if (log.isDebugEnabled()) {
        log.debug(String.format("Distributed lock acquired for %s", object.getClass().getSimpleName()));
    }
    return lock;
}

protected void releaseDistributedLock(ILock lock) {
    if (lock == null) {
        if (log.isWarnEnabled()) {
            log.warn("Could not release distributed lock, lock is null");
        }
        return;
    }

    if (log.isDebugEnabled()) {
        log.debug(String.format("Releasing distributed lock for %s...", lock.getKey()));
    }
    lock.forceUnlock();
    if (log.isDebugEnabled()) {
        log.debug(String.format("Distributed lock released for %s", lock.getKey()));
    }
}

@Verify(global = true)
public void verify() {
    for (int i = 0; i < maxAccounts; i++) {
        ILock lock = targetInstance.getLock(name + i);
        assertFalse(name + ": Lock should be unlocked", lock.isLocked());
    }

    long totalValue = 0;
    IList<Long> accounts = targetInstance.getList(name);
    for (long value : accounts) {
        totalValue += value;
    }
    logger.info(": totalValue=" + totalValue);
    assertEquals(name + ": totalInitialValue != totalValue ", totalInitialValue, totalValue);

    Counter total = new Counter();
    IList<Counter> totals = targetInstance.getList(name + "count");
    for (Counter count : totals) {
        total.add(count);
    }
    logger.info("total count " + total);
}

@TimeStep
public void timeStep(BaseThreadState state) {
    int lockIndex = state.randomInt(lockCount);
    ILock lock = targetInstance.getLock(name + lockIndex);

    int leaseTime = 1 + state.randomInt(maxLeaseTimeMillis);
    int tryTime = 1 + state.randomInt(maxTryTimeMillis);

    if (state.randomBoolean()) {
        lock.lock(leaseTime, MILLISECONDS);
    } else {
        try {
            lock.tryLock(tryTime, MILLISECONDS, leaseTime, MILLISECONDS);
        } catch (InterruptedException e) {
            logger.info("tryLock() got exception: " + e.getMessage());
        }
    }
}

@Verify
public void verify() {
    for (int i = 0; i < lockCount; i++) {
        ILock lock = targetInstance.getLock(name + i);

        boolean isLocked = lock.isLocked();
        long remainingLeaseTime = lock.getRemainingLeaseTime();
        if (isLocked) {
            String message = format("%s is locked with remainingLeaseTime: %d ms", lock, remainingLeaseTime);
            if (allowZeroMillisRemainingLeaseLockTime && remainingLeaseTime == 0) {
                logger.warn(message);
            } else {
                fail(message);
            }
        }
        if (remainingLeaseTime > 0) {
            fail(format("%s has remainingLeaseTime: %d ms", lock, remainingLeaseTime));
        }
    }
}

@Override
protected void doRunInitialization(final String beanName) {
    LockKey lockKey = new LockKey(KeyType.INITIALIZATION, beanName);
    // Try to get the initialization lock
    ILock lock = hazelcastInstance.getLock(lockKey);
    if (lock.tryLock()) {
        // No one else is trying to run this initialization right now. Check if it was already ran by someone else
        if (!initializationControl.containsKey(beanName)) {
            try {
                // This initialization was never executed. Run it and mark it as executed
                super.doRunInitialization(beanName);
                initializationControl.put(beanName, beanName);
            } finally {
                HazelcastHelper.release(lock);
            }
        }
    } else {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Not running initialization for bean " + beanName + " because some other node is currently running it");
        }
    }
}

@Override
protected boolean doRunPollingTask(final String key, final Callable<Boolean> task) {
    LockKey lockKey = new LockKey(KeyType.POLLING_TASK, key);
    ILock lock = hazelcastInstance.getLock(lockKey);
    // Ensure multiple nodes can't run a polling task simultaneously
    if (lock.tryLock()) {
        try {
            return super.doRunPollingTask(key, task);
        } finally {
            HazelcastHelper.release(lock);
        }
    } else {
        // Force a sleep, as couldn't get the lock for this polling task
        if (LOG.isDebugEnabled()) {
            LOG.debug("Some other cluster node is running the " + key + " polling task. Leaving.");
        }
        return false;
    }
}

private void acquire(final LockKey key) {
    if (acquiredLocks.containsKey(key)) {
        // Already own the lock
        return;
    }
    ILock lock = hazelcastInstance.getLock(key);
    try {
        if (lock.tryLock(timeoutSeconds, TimeUnit.SECONDS)) {
            acquiredLocks.put(key, lock);
        } else {
            throw new LockingException();
        }
    } catch (InterruptedException e) {
        throw new LockingException(e);
    }

}

@Test
public void testLockUnlock() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    final ILock lock = hClient.getLock("testLockUnlock");
    lock.lock();
    final CountDownLatch latch = new CountDownLatch(1);
    final CountDownLatch unlockLatch = new CountDownLatch(1);
    new Thread(new Runnable() {
        public void run() {
            assertFalse(lock.tryLock());
            unlockLatch.countDown();
            lock.lock();
            latch.countDown();
        }
    }).start();
    assertTrue(unlockLatch.await(10, TimeUnit.SECONDS));
    lock.unlock();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
}

@Test
public void testTryLock() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    final ILock lock = hClient.getLock("testTryLock");
    assertTrue(lock.tryLock());
    lock.lock();
    final CountDownLatch latch = new CountDownLatch(1);
    final CountDownLatch unlockLatch = new CountDownLatch(1);
    new Thread(new Runnable() {

        public void run() {
            assertFalse(lock.tryLock());
            unlockLatch.countDown();
            try {
                assertTrue(lock.tryLock(10, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            latch.countDown();
        }
    }).start();
    assertTrue(unlockLatch.await(10, TimeUnit.SECONDS));
    lock.unlock();
    lock.unlock();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
}

/**
 * Gets a lock
 * @param name
 * @return
 */
 ILock getLock(String name)
{
    if(isRunning())
    {
        return hazelcast.getLock(name);
    }
    throw new IllegalStateException("Hazelcast not running");
}

public Object synchronizePut(Object key, Object value, String map) {
 if(isRunning()){
    ILock lock = hazelcast.getLock(map);

    try
    {
      if(lock.tryLock(10, TimeUnit.SECONDS))
      {
        return hazelcast.getMap(map).put(key, value);
      }
      else
      {
        log.warn("[synchronizePut] Operation did not synchroznize in 10 secs");
        return hazelcast.getMap(map).put(key, value);

      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      log.debug("", e);
    }
    finally
    {
      lock.unlock();
    }

  }
  return null;

}

public void synchronizeSet(Object key, Object value, String map) {
  if(isRunning()){
    ILock lock = hazelcast.getLock(map);

    try
    {
      if(lock.tryLock(10, TimeUnit.SECONDS))
      {
        hazelcast.getMap(map).set(key, value);
      }
      else
      {
        hazelcast.getMap(map).set(key, value);
        log.warn("[synchronizeSet] Operation did not synchroznize in 10 secs");
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      log.debug("", e);
    }
    finally
    {
      lock.unlock();
    }

  }

}

@Override
protected void doHandleDatabaseInitialization(final Runnable runnable) {
    LockKey lockKey = new LockKey(KeyType.DB_INIT, StringUtils.EMPTY);
    ILock lock = hazelcastInstance.getLock(lockKey.toString());
    // Sleep until the lock is acquired
    lock.lock();
    try {
        super.doHandleDatabaseInitialization(runnable);
    } finally {
        HazelcastHelper.release(lock);
    }
}

@Override
public ILock getILock(String name)
{
    name = Objects.requireNonNull(name);

    final ILock valu = getBeanSafely(name, ILock.class);

    if (null != valu)
    {
        return valu;
    }
    return hz().getLock(name);
}

private static ILock getLock(){
    if(lock==null){
        synchronized (Locks.class){
            if(lock==null){
                lock = HazelcastWrapper.getInstance().getLock("s1.locks");
            }
        }
    }
    return lock;
}

@Override
public boolean equals(Object o) {
    if (o != null && o instanceof ILock) {
        return getId().equals(((ILock) o).getId());
    } else {
        return false;
    }
}

/**
 * Releases a given distributed/local lock.
 *
 * @param lock
 */
@Override
public void releaseLock(Lock lock) {
    if (isClustered()) {
        releaseDistributedLock((ILock) lock);
    } else {
        lock.unlock();
    }
}

private void loadSegment(int segmentNumber) throws Exception {
    indexLock.writeLock().lock();
    try {
        if (!segmentMap.containsKey(segmentNumber)) {
            String lockName = indexName + "-" + segmentNumber;
            ILock hzLock = hazelcastManager.getLock(lockName);
            hazelLockMap.put(segmentNumber, hzLock);
            log.info("Waiting for lock for index <" + indexName + "> segment <" + segmentNumber + ">");
            hzLock.lock();
            log.info("Obtained lock for index <" + indexName + "> segment <" + segmentNumber + ">");

            //Just for clarity
            IndexSegmentInterface indexSegmentInterface = this;

            //doesnt need to be done each time and it is done in StartNode but helps with test cases that take different paths
            FacetsConfig.DEFAULT_DIM_CONFIG.multiValued = true;
            facetsConfig = new FacetsConfig();

            LumongoSegment s = new LumongoSegment(segmentNumber, indexSegmentInterface, indexConfig, facetsConfig, documentStorage);
            segmentMap.put(segmentNumber, s);

            log.info("Loaded segment <" + segmentNumber + "> for index <" + indexName + ">");
            log.info("Current segments <" + (new TreeSet<>(segmentMap.keySet())) + "> for index <" + indexName + ">");

        }
    }
    finally {
        indexLock.writeLock().unlock();
    }
}

public void unloadSegment(int segmentNumber, boolean terminate) throws IOException {
    indexLock.writeLock().lock();
    try {
        ILock hzLock = hazelLockMap.get(segmentNumber);
        try {
            if (segmentMap.containsKey(segmentNumber)) {
                LumongoSegment s = segmentMap.remove(segmentNumber);
                if (s != null) {
                    log.info("Closing segment <" + segmentNumber + "> for index <" + indexName + ">");
                    s.close(terminate);
                    log.info("Removed segment <" + segmentNumber + "> for index <" + indexName + ">");
                    log.info("Current segments <" + (new TreeSet<>(segmentMap.keySet())) + "> for index <" + indexName + ">");
                }
            }

        }
        finally {
            try {
                hzLock.forceUnlock();
                log.info("Unlocked lock for index <" + indexName + "> segment <" + segmentNumber + ">");
            }
            catch (Exception e) {
                log.error("Failed to unlock <" + segmentNumber + ">: ", e);
            }
        }
    }
    finally {
        indexLock.writeLock().unlock();
    }

}

@TimeStep
public void timeStep(BaseThreadState state) {
    int key1 = state.randomInt(maxAccounts);
    int key2;
    do {
        key2 = state.randomInt(maxAccounts);
    } while (key1 == key2);

    ILock lock1 = targetInstance.getLock(name + key1);
    if (lock1.tryLock()) {
        try {
            ILock lock2 = targetInstance.getLock(name + key2);
            if (lock2.tryLock()) {
                try {
                    IAtomicLong account1 = targetInstance.getAtomicLong(name + key1);
                    IAtomicLong account2 = targetInstance.getAtomicLong(name + key2);

                    int delta = state.randomInt(100);
                    if (account1.get() >= delta) {
                        account1.set(account1.get() - delta);
                        account2.set(account2.get() + delta);
                    }
                } finally {
                    lock2.unlock();
                }
            }
        } finally {
            lock1.unlock();
        }
    }
}

@Verify
public void verify() {
    int value = 0;
    for (int i = 0; i < maxAccounts; i++) {
        ILock lock = targetInstance.getLock(name + i);
        IAtomicLong account = targetInstance.getAtomicLong(name + i);

        logger.info(format("%s %d", account, account.get()));

        assertFalse(name + ": Lock should be unlocked", lock.isLocked());
        assertTrue(name + ": Amount is < 0 ", account.get() >= 0);
        value += account.get();
    }
    assertEquals(name + " totals not adding up ", totalValue, value);
}

@TimeStep
public void timeStep(ThreadState state) {
    long key1 = state.getRandomAccountKey();
    long key2 = state.getRandomAccountKey();
    int randomAmount = state.randomInt(amount);

    ILock lock1 = targetInstance.getLock(getLockId(key1));
    ILock lock2 = targetInstance.getLock(getLockId(key2));
    IAtomicLong account1 = targetInstance.getAtomicLong(getAccountId(key1));
    IAtomicLong account2 = targetInstance.getAtomicLong(getAccountId(key2));

    if (!lock1.tryLock()) {
        return;
    }
    try {
        if (!lock2.tryLock()) {
            return;
        }
        try {
            if (account1.get() < 0 || account2.get() < 0) {
                throw new TestException("Amount on account can't be smaller than 0");
            }
            if (account1.get() < randomAmount) {
                return;
            }
            account1.set(account1.get() - randomAmount);
            account2.set(account2.get() + randomAmount);
        } finally {
            lock2.unlock();
        }
    } finally {
        lock1.unlock();
    }
}

@Verify
public void verify() {
    long actual = 0;
    for (long i = 0; i < lockCounter.get(); i++) {
        ILock lock = targetInstance.getLock(getLockId(i));
        assertFalse("Lock should be unlocked", lock.isLocked());

        long accountAmount = targetInstance.getAtomicLong(getAccountId(i)).get();
        assertTrue("Amount on account can't be smaller than 0", accountAmount >= 0);
        actual += accountAmount;
    }

    long expected = initialAmount * lockCounter.get();
    assertEquals(format("%s: Money was lost or created (%d)", name, expected - actual), expected, actual);
}

@Override
protected void doHandleDatabaseInitialization(final Runnable runnable) {
    LockKey lockKey = new LockKey(KeyType.DB_INIT, StringUtils.EMPTY);
    ILock lock = hazelcastInstance.getLock(lockKey);
    // Sleep until the lock is acquired
    lock.lock();
    try {
        super.doHandleDatabaseInitialization(runnable);
    } finally {
        HazelcastHelper.release(lock);
    }
}

@Override
public boolean equals(Object o) {
    if (o != null && o instanceof ILock) {
        return getId().equals(((ILock) o).getId());
    } else {
        return false;
    }
}

/**
 * Test for issue #39
 */
@Test
public void testIsLocked() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    final ILock lock = hClient.getLock("testIsLocked");
    assertFalse(lock.isLocked());
    lock.lock();
    assertTrue(lock.isLocked());

    final CountDownLatch latch = new CountDownLatch(1);
    Thread thread = new Thread(new Runnable() {
        public void run() {
            assertTrue(lock.isLocked());
            try {
                while (lock.isLocked()) {
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            latch.countDown();
        }
    });
    thread.start();
    Thread.sleep(100);
    lock.unlock();
    assertTrue(latch.await(3, TimeUnit.SECONDS));
}

@Override
public ILock getLock(String s) {
    return null;
}

protected ILock getLock(String key) {
    return substance.getLock(key);
}

@Override
protected void doRunScheduledTask(final String taskName, final Calendar time) {
    // Scheduled tasks won't run twice for the same hour in the entire cluster.
    LockKey lockKey = new LockKey(KeyType.SCHEDULED_TASK, taskName);
    ILock lock = hazelcastInstance.getLock(lockKey.toString());
    if (lock.tryLock()) {
        // No other node is trying to execute this scheduled task
        try {
            // Determine whether the task is daily
            ScheduledTask scheduledTask = getSchedulingHandler().getTask(taskName);
            boolean daily = !scheduledTask.isEveryHour();
            int field = daily ? Calendar.DAY_OF_MONTH : Calendar.HOUR_OF_DAY;

            // Check the last hour this task was performed
            Calendar lastRun = scheduledTaskControl.get(taskName);
            if (lastRun != null) {
                lastRun = DateUtils.truncate(lastRun, field);
            }
            Calendar thisRun = DateUtils.truncate(time, field);

            // Fill all the gaps between the last run and this run.
            // In normal execution, this loop will be evaluated only once.
            while (lastRun == null || lastRun.before(thisRun)) {
                if (lastRun == null) {
                    // Never executed: run as this time
                    lastRun = thisRun;
                } else {
                    // Increment the field (either hour or day)
                    lastRun.add(field, 1);
                }

                // Run the task
                super.doRunScheduledTask(taskName, lastRun);

                // Store the task hour, no other node will run it on this hour again
                scheduledTaskControl.put(taskName, lastRun);
            }
        } finally {
            HazelcastHelper.release(lock);
        }
    }
}

@Override
public void release() {
    for (ILock lock : acquiredLocks.values()) {
        HazelcastHelper.release(lock);
    }
}

@Override
public ILock getLock(final String key) {
  return getHazelcastInstance().getLock(key);
}

@Override
public ILock getLock(final Object key) {
  throw new DeprecatedError();
}

public LockMBean(ILock lock, ManagementService managementService) {
    super(lock, managementService);
}