github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'mrniko/master'

Browse Source
pull/282/head
jackygurui 10 years ago
parent e31ac9c60e 4c483c6bc0
commit 8b587cf16b
4 changed files with 103 additions and 184 deletions
Show Stats Download Patch File Download Diff File

95
src/main/java/org/redisson/RedissonCountDownLatch.java
Unescape Escape View File

@ -54,24 +54,32 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown
this.id = id; this.id = id;
} }
private Future<Boolean> subscribe() { private Future<RedissonCountDownLatchEntry> subscribe() {
Promise<Boolean> promise = aquire(); synchronized (ENTRIES) {
if (promise != null) { RedissonCountDownLatchEntry entry = ENTRIES.get(getEntryName());
return promise; if (entry != null) {
} entry.aquire();
return entry.getPromise();
}
Promise<RedissonCountDownLatchEntry> newPromise = newPromise();
final RedissonCountDownLatchEntry value = new RedissonCountDownLatchEntry(newPromise);
value.aquire();
Promise<Boolean> newPromise = newPromise(); RedissonCountDownLatchEntry oldValue = ENTRIES.putIfAbsent(getEntryName(), value);
final RedissonCountDownLatchEntry value = new RedissonCountDownLatchEntry(newPromise); if (oldValue != null) {
value.aquire(); oldValue.aquire();
RedissonCountDownLatchEntry oldValue = ENTRIES.putIfAbsent(getEntryName(), value); return oldValue.getPromise();
if (oldValue != null) {
Promise<Boolean> oldPromise = aquire();
if (oldPromise == null) {
return subscribe();
} }
return oldPromise;
RedisPubSubListener<Integer> listener = createListener(value);
commandExecutor.getConnectionManager().subscribe(listener, getChannelName());
return newPromise;
} }
}
private RedisPubSubListener<Integer> createListener(final RedissonCountDownLatchEntry value) {
RedisPubSubListener<Integer> listener = new BaseRedisPubSubListener<Integer>() { RedisPubSubListener<Integer> listener = new BaseRedisPubSubListener<Integer>() {
@Override @Override
@ -89,61 +97,32 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown
@Override @Override
public boolean onStatus(PubSubType type, String channel) { public boolean onStatus(PubSubType type, String channel) {
if (channel.equals(getChannelName()) && !value.getPromise().isSuccess()) { if (channel.equals(getChannelName()) && !value.getPromise().isSuccess()
value.getPromise().setSuccess(true); && type == PubSubType.SUBSCRIBE) {
value.getPromise().setSuccess(value);
return true; return true;
} }
return false; return false;
} }
}; };
return listener;
synchronized (ENTRIES) {
commandExecutor.getConnectionManager().subscribe(listener, getChannelName());
}
return newPromise;
}
private void unsubscribe() {
while (true) {
RedissonCountDownLatchEntry entry = ENTRIES.get(getEntryName());
if (entry == null) {
return;
}
RedissonCountDownLatchEntry newEntry = new RedissonCountDownLatchEntry(entry);
newEntry.release();
if (ENTRIES.replace(getEntryName(), entry, newEntry)) {
if (newEntry.isFree()
&& ENTRIES.remove(getEntryName(), newEntry)) {
synchronized (ENTRIES) {
// maybe added during subscription
if (!ENTRIES.containsKey(getEntryName())) {
commandExecutor.getConnectionManager().unsubscribe(getChannelName());
}
}
}
return;
}
}
} }
private Promise<Boolean> aquire() { private void unsubscribe(RedissonCountDownLatchEntry entry) {
while (true) { synchronized (ENTRIES) {
RedissonCountDownLatchEntry entry = ENTRIES.get(getEntryName()); if (entry.release() == 0) {
if (entry != null) { // just an assertion
RedissonCountDownLatchEntry newEntry = new RedissonCountDownLatchEntry(entry); boolean removed = ENTRIES.remove(getEntryName()) == entry;
newEntry.aquire(); if (removed) {
if (ENTRIES.replace(getEntryName(), entry, newEntry)) { commandExecutor.getConnectionManager().unsubscribe(getChannelName());
return newEntry.getPromise();
} }
} else {
return null;
} }
} }
} }
public void await() throws InterruptedException { public void await() throws InterruptedException {
Future<Boolean> promise = subscribe(); Future<RedissonCountDownLatchEntry> promise = subscribe();
try { try {
promise.await(); promise.await();
@ -155,14 +134,14 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown
} }
} }
} finally { } finally {
unsubscribe(); unsubscribe(promise.getNow());
} }
} }
@Override @Override
public boolean await(long time, TimeUnit unit) throws InterruptedException { public boolean await(long time, TimeUnit unit) throws InterruptedException {
Future<Boolean> promise = subscribe(); Future<RedissonCountDownLatchEntry> promise = subscribe();
try { try {
if (!promise.await(time, unit)) { if (!promise.await(time, unit)) {
return false; return false;
@ -186,7 +165,7 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown
return true; return true;
} finally { } finally {
unsubscribe(); unsubscribe(promise.getNow());
} }
} }

56
src/main/java/org/redisson/RedissonCountDownLatchEntry.java
Unescape Escape View File

@ -15,69 +15,37 @@
*/ */
package org.redisson; package org.redisson;
import io.netty.util.concurrent.Promise;
import org.redisson.misc.ReclosableLatch; import org.redisson.misc.ReclosableLatch;
import io.netty.util.concurrent.Promise;
public class RedissonCountDownLatchEntry { public class RedissonCountDownLatchEntry {
private int counter; private int counter;
private final ReclosableLatch latch; private final ReclosableLatch latch;
private final Promise<Boolean> promise; private final Promise<RedissonCountDownLatchEntry> promise;
public RedissonCountDownLatchEntry(RedissonCountDownLatchEntry source) { public RedissonCountDownLatchEntry(Promise<RedissonCountDownLatchEntry> promise) {
counter = source.counter;
latch = source.latch;
promise = source.promise;
}
public RedissonCountDownLatchEntry(Promise<Boolean> promise) {
super(); super();
this.latch = new ReclosableLatch(); this.latch = new ReclosableLatch();
this.promise = promise; this.promise = promise;
} }
public boolean isFree() {
return counter == 0;
}
public void aquire() { public void aquire() {
counter++; counter++;
} }
public void release() { public int release() {
counter--; return --counter;
} }
public Promise<Boolean> getPromise() { public Promise<RedissonCountDownLatchEntry> getPromise() {
return promise; return promise;
} }
public ReclosableLatch getLatch() { public ReclosableLatch getLatch() {
return latch; return latch;
} }
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + counter;
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
RedissonCountDownLatchEntry other = (RedissonCountDownLatchEntry) obj;
if (counter != other.counter)
return false;
return true;
}
} }

108
src/main/java/org/redisson/RedissonLock.java
Unescape Escape View File

@ -58,26 +58,14 @@ public class RedissonLock extends RedissonExpirable implements RLock {
this.id = id; this.id = id;
} }
private void unsubscribe() { private void unsubscribe(RedissonLockEntry entry) {
while (true) { synchronized (ENTRIES) {
RedissonLockEntry entry = ENTRIES.get(getEntryName()); if (entry.release() == 0) {
if (entry == null) { // just an assertion
return; boolean removed = ENTRIES.remove(getEntryName()) == entry;
} if (removed) {
commandExecutor.getConnectionManager().unsubscribe(getChannelName());
RedissonLockEntry newEntry = new RedissonLockEntry(entry);
newEntry.release();
if (ENTRIES.replace(getEntryName(), entry, newEntry)) {
if (newEntry.isFree()
&& ENTRIES.remove(getEntryName(), newEntry)) {
synchronized (ENTRIES) {
// maybe added during subscription
if (!ENTRIES.containsKey(getEntryName())) {
commandExecutor.getConnectionManager().unsubscribe(getChannelName());
}
}
} }
return;
} }
} }
} }
@ -86,63 +74,49 @@ public class RedissonLock extends RedissonExpirable implements RLock {
return id + ":" + getName(); return id + ":" + getName();
} }
private Promise<Boolean> aquire() { private Future<RedissonLockEntry> subscribe() {
while (true) { synchronized (ENTRIES) {
RedissonLockEntry entry = ENTRIES.get(getEntryName()); RedissonLockEntry entry = ENTRIES.get(getEntryName());
if (entry == null) { if (entry != null) {
return null; entry.aquire();
} return entry.getPromise();
RedissonLockEntry newEntry = new RedissonLockEntry(entry);
newEntry.aquire();
if (ENTRIES.replace(getEntryName(), entry, newEntry)) {
return newEntry.getPromise();
} }
}
}
private Future<Boolean> subscribe() { Promise<RedissonLockEntry> newPromise = newPromise();
Promise<Boolean> promise = aquire(); final RedissonLockEntry value = new RedissonLockEntry(newPromise);
if (promise != null) { value.aquire();
return promise;
}
Promise<Boolean> newPromise = newPromise(); RedissonLockEntry oldValue = ENTRIES.putIfAbsent(getEntryName(), value);
final RedissonLockEntry value = new RedissonLockEntry(newPromise); if (oldValue != null) {
value.aquire(); oldValue.aquire();
RedissonLockEntry oldValue = ENTRIES.putIfAbsent(getEntryName(), value); return oldValue.getPromise();
if (oldValue != null) {
Promise<Boolean> oldPromise = aquire();
if (oldPromise == null) {
return subscribe();
} }
return oldPromise;
}
RedisPubSubListener<Integer> listener = new BaseRedisPubSubListener<Integer>() { RedisPubSubListener<Integer> listener = new BaseRedisPubSubListener<Integer>() {
@Override @Override
public void onMessage(String channel, Integer message) { public void onMessage(String channel, Integer message) {
if (message.equals(unlockMessage) && getChannelName().equals(channel)) { if (message.equals(unlockMessage) && getChannelName().equals(channel)) {
value.getLatch().release(); value.getLatch().release();
}
} }
}
@Override @Override
public boolean onStatus(PubSubType type, String channel) { public boolean onStatus(PubSubType type, String channel) {
if (channel.equals(getChannelName()) && !value.getPromise().isSuccess()) { if (channel.equals(getChannelName())
value.getPromise().setSuccess(true); && !value.getPromise().isSuccess()
return true; && type == PubSubType.SUBSCRIBE) {
value.getPromise().setSuccess(value);
return true;
}
return false;
} }
return false;
}
}; };
synchronized (ENTRIES) {
commandExecutor.getConnectionManager().subscribe(listener, getChannelName()); commandExecutor.getConnectionManager().subscribe(listener, getChannelName());
return newPromise;
} }
return newPromise;
} }
private String getChannelName() { private String getChannelName() {
@ -186,7 +160,8 @@ public class RedissonLock extends RedissonExpirable implements RLock {
return; return;
} }
subscribe().awaitUninterruptibly(); Future<RedissonLockEntry> future = subscribe();
future.awaitUninterruptibly();
try { try {
while (true) { while (true) {
@ -209,7 +184,7 @@ public class RedissonLock extends RedissonExpirable implements RLock {
} }
} }
} finally { } finally {
unsubscribe(); unsubscribe(future.getNow());
} }
} }
@ -290,7 +265,8 @@ public class RedissonLock extends RedissonExpirable implements RLock {
return true; return true;
} }
if (!subscribe().awaitUninterruptibly(time, TimeUnit.MILLISECONDS)) { Future<RedissonLockEntry> future = subscribe();
if (!future.awaitUninterruptibly(time, TimeUnit.MILLISECONDS)) {
return false; return false;
} }
@ -325,7 +301,7 @@ public class RedissonLock extends RedissonExpirable implements RLock {
} }
return true; return true;
} finally { } finally {
unsubscribe(); unsubscribe(future.getNow());
} }
} }

28
src/main/java/org/redisson/RedissonLockEntry.java
Unescape Escape View File

@ -24,36 +24,32 @@ public class RedissonLockEntry {
private int counter; private int counter;
private final Semaphore latch; private final Semaphore latch;
private final Promise<Boolean> promise; private final Promise<RedissonLockEntry> promise;
public RedissonLockEntry(RedissonLockEntry source) { public RedissonLockEntry(RedissonLockEntry source) {
counter = source.counter; counter = source.counter;
latch = source.latch; latch = source.latch;
promise = source.promise; promise = source.promise;
} }
public RedissonLockEntry(Promise<Boolean> promise) { public RedissonLockEntry(Promise<RedissonLockEntry> promise) {
super(); super();
this.latch = new Semaphore(0); this.latch = new Semaphore(0);
this.promise = promise; this.promise = promise;
} }
public boolean isFree() {
return counter == 0;
}
public void aquire() { public void aquire() {
counter++; counter++;
} }
public void release() { public int release() {
counter--; return --counter;
} }
public Promise<Boolean> getPromise() { public Promise<RedissonLockEntry> getPromise() {
return promise; return promise;
} }
public Semaphore getLatch() { public Semaphore getLatch() {
return latch; return latch;
} }
@ -79,5 +75,5 @@ public class RedissonLockEntry {
return false; return false;
return true; return true;
} }
} }

Write Preview
Loading…
Cancel
Save