RedissonRedLock implemented. #533
Nikita
9 years ago
parent
fe3f3dfb0f
commit
db4fba6533
@ -0,0 +1,100 @@
|
||||
/**
|
||||
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.redisson.core;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import io.netty.util.concurrent.Future;
|
||||
|
||||
/**
|
||||
* RedLock locking algorithm implementation for multiple locks.
|
||||
* It manages all locks as one.
|
||||
*
|
||||
* @see <a href="http://redis.io/topics/distlock">http://redis.io/topics/distlock</a>
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class RedissonRedLock extends RedissonMultiLock {
|
||||
|
||||
/**
|
||||
* Creates instance with multiple {@link RLock} objects.
|
||||
* Each RLock object could be created by own Redisson instance.
|
||||
*
|
||||
* @param locks
|
||||
*/
|
||||
public RedissonRedLock(RLock... locks) {
|
||||
super(locks);
|
||||
}
|
||||
|
||||
protected boolean sync(Map<RLock, Future<Boolean>> tryLockFutures) {
|
||||
Queue<RLock> lockedLocks = new ConcurrentLinkedQueue<RLock>();
|
||||
RuntimeException latestException = null;
|
||||
for (Entry<RLock, Future<Boolean>> entry : tryLockFutures.entrySet()) {
|
||||
try {
|
||||
if (entry.getValue().syncUninterruptibly().getNow()) {
|
||||
lockedLocks.add(entry.getKey());
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
latestException = e;
|
||||
}
|
||||
}
|
||||
|
||||
if (lockedLocks.size() < minLocksAmount(locks)) {
|
||||
unlock();
|
||||
lockedLocks.clear();
|
||||
if (latestException != null) {
|
||||
throw latestException;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
public void unlock() {
|
||||
List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(locks.size());
|
||||
|
||||
for (RLock lock : locks) {
|
||||
futures.add(lock.forceUnlockAsync());
|
||||
}
|
||||
|
||||
for (Future<Boolean> future : futures) {
|
||||
future.awaitUninterruptibly();
|
||||
}
|
||||
}
|
||||
|
||||
protected int minLocksAmount(final List<RLock> locks) {
|
||||
return locks.size()/2 + 1;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean isLockFailed(Future<Boolean> future) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean isAllLocksAcquired(AtomicReference<RLock> lockedLockHolder, AtomicReference<Throwable> failed, Queue<RLock> lockedLocks) {
|
||||
return (lockedLockHolder.get() == null && failed.get() == null) || lockedLocks.size() >= minLocksAmount(locks);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,204 @@
|
||||
package org.redisson;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.redisson.core.RLock;
|
||||
import org.redisson.core.RedissonMultiLock;
|
||||
import org.redisson.core.RedissonRedLock;
|
||||
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import org.redisson.RedisRunner.RedisProcess;
|
||||
import static com.jayway.awaitility.Awaitility.await;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
public class RedissonRedLockTest {
|
||||
|
||||
@Test
|
||||
public void testLockFailed() throws IOException, InterruptedException {
|
||||
RedisProcess redis1 = redisTestMultilockInstance(6320);
|
||||
RedisProcess redis2 = redisTestMultilockInstance(6321);
|
||||
|
||||
RedissonClient client1 = createClient("127.0.0.1:6320");
|
||||
RedissonClient client2 = createClient("127.0.0.1:6321");
|
||||
|
||||
RLock lock1 = client1.getLock("lock1");
|
||||
RLock lock2 = client1.getLock("lock2");
|
||||
RLock lock3 = client2.getLock("lock3");
|
||||
|
||||
Thread t1 = new Thread() {
|
||||
public void run() {
|
||||
lock3.lock();
|
||||
};
|
||||
};
|
||||
t1.start();
|
||||
t1.join();
|
||||
|
||||
Thread t = new Thread() {
|
||||
public void run() {
|
||||
RedissonMultiLock lock = new RedissonRedLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
|
||||
System.out.println("123");
|
||||
|
||||
try {
|
||||
Thread.sleep(3000);
|
||||
} catch (InterruptedException e) {
|
||||
}
|
||||
|
||||
lock.unlock();
|
||||
};
|
||||
};
|
||||
t.start();
|
||||
t.join(1000);
|
||||
|
||||
RedissonMultiLock lock = new RedissonRedLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
System.out.println("1234");
|
||||
lock.unlock();
|
||||
|
||||
assertThat(redis1.stop()).isEqualTo(0);
|
||||
assertThat(redis2.stop()).isEqualTo(0);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testConnectionFailed() throws IOException, InterruptedException {
|
||||
RedisProcess redis1 = redisTestMultilockInstance(6320);
|
||||
RedisProcess redis2 = redisTestMultilockInstance(6321);
|
||||
|
||||
RedissonClient client1 = createClient("127.0.0.1:6320");
|
||||
RedissonClient client2 = createClient("127.0.0.1:6321");
|
||||
|
||||
RLock lock1 = client1.getLock("lock1");
|
||||
RLock lock2 = client1.getLock("lock2");
|
||||
assertThat(redis2.stop()).isEqualTo(0);
|
||||
RLock lock3 = client2.getLock("lock3");
|
||||
|
||||
Thread t = new Thread() {
|
||||
public void run() {
|
||||
RedissonMultiLock lock = new RedissonRedLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
|
||||
try {
|
||||
Thread.sleep(3000);
|
||||
} catch (InterruptedException e) {
|
||||
}
|
||||
|
||||
lock.unlock();
|
||||
};
|
||||
};
|
||||
t.start();
|
||||
t.join(1000);
|
||||
|
||||
RedissonMultiLock lock = new RedissonRedLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
lock.unlock();
|
||||
|
||||
assertThat(redis1.stop()).isEqualTo(0);
|
||||
}
|
||||
|
||||
|
||||
// @Test
|
||||
public void testMultiThreads() throws IOException, InterruptedException {
|
||||
RedisProcess redis1 = redisTestMultilockInstance(6320);
|
||||
|
||||
Config config1 = new Config();
|
||||
config1.useSingleServer().setAddress("127.0.0.1:6320");
|
||||
RedissonClient client = Redisson.create(config1);
|
||||
|
||||
RLock lock1 = client.getLock("lock1");
|
||||
RLock lock2 = client.getLock("lock2");
|
||||
RLock lock3 = client.getLock("lock3");
|
||||
|
||||
Thread t = new Thread() {
|
||||
public void run() {
|
||||
RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
|
||||
try {
|
||||
Thread.sleep(3000);
|
||||
} catch (InterruptedException e) {
|
||||
}
|
||||
|
||||
lock.unlock();
|
||||
};
|
||||
};
|
||||
t.start();
|
||||
t.join(1000);
|
||||
|
||||
RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
lock.unlock();
|
||||
|
||||
assertThat(redis1.stop()).isEqualTo(0);
|
||||
}
|
||||
|
||||
// @Test
|
||||
public void test() throws IOException, InterruptedException {
|
||||
RedisProcess redis1 = redisTestMultilockInstance(6320);
|
||||
RedisProcess redis2 = redisTestMultilockInstance(6321);
|
||||
RedisProcess redis3 = redisTestMultilockInstance(6322);
|
||||
|
||||
NioEventLoopGroup group = new NioEventLoopGroup();
|
||||
|
||||
RedissonClient client1 = createClient(group, "127.0.0.1:6320");
|
||||
RedissonClient client2 = createClient(group, "127.0.0.1:6321");
|
||||
RedissonClient client3 = createClient(group, "127.0.0.1:6322");
|
||||
|
||||
final RLock lock1 = client1.getLock("lock1");
|
||||
final RLock lock2 = client2.getLock("lock2");
|
||||
final RLock lock3 = client3.getLock("lock3");
|
||||
|
||||
RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
|
||||
lock.lock();
|
||||
|
||||
final AtomicBoolean executed = new AtomicBoolean();
|
||||
|
||||
Thread t = new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
|
||||
assertThat(lock.tryLock()).isFalse();
|
||||
assertThat(lock.tryLock()).isFalse();
|
||||
executed.set(true);
|
||||
}
|
||||
};
|
||||
t.start();
|
||||
t.join();
|
||||
|
||||
await().atMost(5, TimeUnit.SECONDS).until(() -> assertThat(executed.get()).isTrue());
|
||||
|
||||
lock.unlock();
|
||||
|
||||
assertThat(redis1.stop()).isEqualTo(0);
|
||||
|
||||
assertThat(redis2.stop()).isEqualTo(0);
|
||||
|
||||
assertThat(redis3.stop()).isEqualTo(0);
|
||||
}
|
||||
|
||||
private RedissonClient createClient(String host) {
|
||||
return createClient(null, host);
|
||||
}
|
||||
|
||||
private RedissonClient createClient(NioEventLoopGroup group, String host) {
|
||||
Config config1 = new Config();
|
||||
config1.useSingleServer().setAddress(host);
|
||||
config1.setEventLoopGroup(group);
|
||||
RedissonClient client1 = Redisson.create(config1);
|
||||
client1.getKeys().flushdb();
|
||||
return client1;
|
||||
}
|
||||
|
||||
private RedisProcess redisTestMultilockInstance(int port) throws IOException, InterruptedException {
|
||||
return new RedisRunner()
|
||||
.nosave()
|
||||
.randomDir()
|
||||
.port(port)
|
||||
.run();
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue