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

refactoring

Browse Source
pull/2041/head
Nikita Koksharov 6 years ago
parent b88765b8a6
commit 1f0459fef1
5 changed files with 76 additions and 71 deletions
Show Stats Download Patch File Download Diff File

31
redisson/src/main/java/org/redisson/BaseRemoteService.java
Unescape Escape View File

@ -31,6 +31,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RFuture;
@ -47,6 +48,7 @@ import org.redisson.client.protocol.RedisCommands;
import org.redisson.codec.CompositeCodec;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.executor.RemotePromise;
import org.redisson.misc.Hash;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
import org.redisson.remote.RRemoteServiceResponse;
@ -64,6 +66,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.ScheduledFuture;
@ -78,7 +82,7 @@ public abstract class BaseRemoteService {
private static final Logger log = LoggerFactory.getLogger(BaseRemoteService.class);
private final Map<Class<?>, String> requestQueueNameCache = new ConcurrentHashMap<>();
private final ConcurrentMap<Method, List<String>> methodSignaturesCache = new ConcurrentHashMap<>();
private final ConcurrentMap<Method, long[]> methodSignaturesCache = new ConcurrentHashMap<>();
protected final Codec codec;
protected final RedissonClient redisson;
@ -200,7 +204,7 @@ public abstract class BaseRemoteService {
Long ackTimeout = optionsCopy.getAckTimeoutInMillis();
RemoteServiceRequest request = new RemoteServiceRequest(executorId, requestId.toString(), method.getName(), getMethodSignatures(method), args,
RemoteServiceRequest request = new RemoteServiceRequest(executorId, requestId.toString(), method.getName(), getMethodSignature(method), args,
optionsCopy, System.currentTimeMillis());
final RFuture<RemoteServiceAck> ackFuture;
@ -507,7 +511,7 @@ public abstract class BaseRemoteService {
RequestId requestId = generateRequestId();
String requestQueueName = getRequestQueueName(remoteInterface);
RemoteServiceRequest request = new RemoteServiceRequest(executorId, requestId.toString(), method.getName(), getMethodSignatures(method), args, optionsCopy,
RemoteServiceRequest request = new RemoteServiceRequest(executorId, requestId.toString(), method.getName(), getMethodSignature(method), args, optionsCopy,
System.currentTimeMillis());
final RFuture<RemoteServiceAck> ackFuture;
@ -676,16 +680,19 @@ public abstract class BaseRemoteService {
}
}
protected List<String> getMethodSignatures(Method method) {
List<String> result = methodSignaturesCache.get(method);
protected long[] getMethodSignature(Method method) {
long[] result = methodSignaturesCache.get(method);
if (result == null) {
result = new ArrayList<>(method.getParameterTypes().length);
for (Class<?> t : method.getParameterTypes()) {
result.add(t.getName());
}
List<String> oldList = methodSignaturesCache.putIfAbsent(method, result);
if (oldList != null) {
result = oldList;
String str = Arrays.stream(method.getParameterTypes())
.map(c -> c.getName())
.collect(Collectors.joining());
ByteBuf buf = Unpooled.copiedBuffer(str, CharsetUtil.UTF_8);
result = Hash.hash128(buf);
buf.release();
long[] oldResult = methodSignaturesCache.putIfAbsent(method, result);
if (oldResult != null) {
return oldResult;
}
}

6
redisson/src/main/java/org/redisson/RedissonRemoteService.java
Unescape Escape View File

@ -125,7 +125,7 @@ public class RedissonRemoteService extends BaseRemoteService implements RRemoteS
@Override
public <T> void deregister(Class<T> remoteInterface) {
for (Method method : remoteInterface.getMethods()) {
RemoteServiceKey key = new RemoteServiceKey(remoteInterface, method.getName(), getMethodSignatures(method));
RemoteServiceKey key = new RemoteServiceKey(remoteInterface, method.getName(), getMethodSignature(method));
beans.remove(key);
}
@ -156,7 +156,7 @@ public class RedissonRemoteService extends BaseRemoteService implements RRemoteS
}
for (Method method : remoteInterface.getMethods()) {
RemoteServiceMethod value = new RemoteServiceMethod(method, object);
RemoteServiceKey key = new RemoteServiceKey(remoteInterface, method.getName(), getMethodSignatures(method));
RemoteServiceKey key = new RemoteServiceKey(remoteInterface, method.getName(), getMethodSignature(method));
if (beans.put(key, value) != null) {
return;
}
@ -307,7 +307,7 @@ public class RedissonRemoteService extends BaseRemoteService implements RRemoteS
private <T> void executeMethod(Class<T> remoteInterface, RBlockingQueue<String> requestQueue,
ExecutorService executor, RemoteServiceRequest request) {
RemoteServiceMethod method = beans.get(new RemoteServiceKey(remoteInterface, request.getMethodName(), request.getSignatures()));
RemoteServiceMethod method = beans.get(new RemoteServiceKey(remoteInterface, request.getMethodName(), request.getSignature()));
String responseName = getResponseQueueName(request.getExecutorId());

66
redisson/src/main/java/org/redisson/pubsub/PublishSubscribeService.java
Unescape Escape View File

@ -62,7 +62,7 @@ public class PublishSubscribeService {
private final ConcurrentMap<ChannelName, PubSubConnectionEntry> name2PubSubConnection = new ConcurrentHashMap<>();
private final Queue<PubSubConnectionEntry> freePubSubConnections = new ConcurrentLinkedQueue<PubSubConnectionEntry>();
private final Queue<PubSubConnectionEntry> freePubSubConnections = new ConcurrentLinkedQueue<>();
private final SemaphorePubSub semaphorePubSub = new SemaphorePubSub(this);
@ -109,9 +109,9 @@ public class PublishSubscribeService {
return subscribe(PubSubType.SUBSCRIBE, codec, channelName, new RedissonPromise<PubSubConnectionEntry>(), listeners);
}
private RFuture<PubSubConnectionEntry> subscribe(final PubSubType type, final Codec codec, final ChannelName channelName,
final RPromise<PubSubConnectionEntry> promise, final RedisPubSubListener<?>... listeners) {
final AsyncSemaphore lock = getSemaphore(channelName);
private RFuture<PubSubConnectionEntry> subscribe(PubSubType type, Codec codec, ChannelName channelName,
RPromise<PubSubConnectionEntry> promise, RedisPubSubListener<?>... listeners) {
AsyncSemaphore lock = getSemaphore(channelName);
lock.acquire(new Runnable() {
@Override
public void run() {
@ -120,7 +120,7 @@ public class PublishSubscribeService {
return;
}
final RPromise<PubSubConnectionEntry> result = new RedissonPromise<PubSubConnectionEntry>();
RPromise<PubSubConnectionEntry> result = new RedissonPromise<PubSubConnectionEntry>();
promise.onComplete((res, e) -> {
if (e != null) {
result.tryFailure(e);
@ -150,11 +150,11 @@ public class PublishSubscribeService {
return locks[Math.abs(channelName.hashCode() % locks.length)];
}
private void subscribe(final Codec codec, final ChannelName channelName,
final RPromise<PubSubConnectionEntry> promise, final PubSubType type, final AsyncSemaphore lock, final RedisPubSubListener<?>... listeners) {
final PubSubConnectionEntry connEntry = name2PubSubConnection.get(channelName);
private void subscribe(Codec codec, ChannelName channelName,
RPromise<PubSubConnectionEntry> promise, PubSubType type, AsyncSemaphore lock, RedisPubSubListener<?>... listeners) {
PubSubConnectionEntry connEntry = name2PubSubConnection.get(channelName);
if (connEntry != null) {
subscribe(channelName, promise, type, lock, connEntry, listeners);
addListeners(channelName, promise, type, lock, connEntry, listeners);
return;
}
@ -168,7 +168,7 @@ public class PublishSubscribeService {
return;
}
final PubSubConnectionEntry freeEntry = freePubSubConnections.peek();
PubSubConnectionEntry freeEntry = freePubSubConnections.peek();
if (freeEntry == null) {
connect(codec, channelName, promise, type, lock, listeners);
return;
@ -179,12 +179,12 @@ public class PublishSubscribeService {
throw new IllegalStateException();
}
final PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, freeEntry);
PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, freeEntry);
if (oldEntry != null) {
freeEntry.release();
freePubSubLock.release();
subscribe(channelName, promise, type, lock, oldEntry, listeners);
addListeners(channelName, promise, type, lock, oldEntry, listeners);
return;
}
@ -193,7 +193,7 @@ public class PublishSubscribeService {
}
freePubSubLock.release();
subscribe(channelName, promise, type, lock, freeEntry, listeners);
addListeners(channelName, promise, type, lock, freeEntry, listeners);
if (PubSubType.PSUBSCRIBE == type) {
freeEntry.psubscribe(codec, channelName);
@ -205,14 +205,14 @@ public class PublishSubscribeService {
});
}
private void subscribe(final ChannelName channelName, final RPromise<PubSubConnectionEntry> promise,
final PubSubType type, final AsyncSemaphore lock, final PubSubConnectionEntry connEntry,
final RedisPubSubListener<?>... listeners) {
private void addListeners(ChannelName channelName, RPromise<PubSubConnectionEntry> promise,
PubSubType type, AsyncSemaphore lock, PubSubConnectionEntry connEntry,
RedisPubSubListener<?>... listeners) {
for (RedisPubSubListener<?> listener : listeners) {
connEntry.addListener(channelName, listener);
}
SubscribeListener list = connEntry.getSubscribeFuture(channelName, type);
final RFuture<Void> subscribeFuture = list.getSuccessFuture();
RFuture<Void> subscribeFuture = list.getSuccessFuture();
subscribeFuture.onComplete((res, e) -> {
if (!promise.trySuccess(connEntry)) {
@ -283,14 +283,14 @@ public class PublishSubscribeService {
freePubSubLock.release();
subscribe(channelName, promise, type, lock, oldEntry, listeners);
addListeners(channelName, promise, type, lock, oldEntry, listeners);
return;
}
freePubSubConnections.add(entry);
freePubSubLock.release();
subscribe(channelName, promise, type, lock, entry, listeners);
addListeners(channelName, promise, type, lock, entry, listeners);
if (PubSubType.PSUBSCRIBE == type) {
entry.psubscribe(codec, channelName);
@ -300,14 +300,14 @@ public class PublishSubscribeService {
});
}
public RFuture<Void> unsubscribe(final ChannelName channelName, final AsyncSemaphore lock) {
final PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
public RFuture<Void> unsubscribe(ChannelName channelName, AsyncSemaphore lock) {
PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
if (entry == null || connectionManager.isShuttingDown()) {
lock.release();
return RedissonPromise.newSucceededFuture(null);
}
final RedissonPromise<Void> result = new RedissonPromise<Void>();
RedissonPromise<Void> result = new RedissonPromise<Void>();
entry.unsubscribe(channelName, new BaseRedisPubSubListener() {
@Override
@ -329,17 +329,17 @@ public class PublishSubscribeService {
return result;
}
public RFuture<Codec> unsubscribe(final ChannelName channelName, final PubSubType topicType) {
public RFuture<Codec> unsubscribe(ChannelName channelName, PubSubType topicType) {
if (connectionManager.isShuttingDown()) {
return RedissonPromise.newSucceededFuture(null);
}
final RPromise<Codec> result = new RedissonPromise<Codec>();
final AsyncSemaphore lock = getSemaphore(channelName);
RPromise<Codec> result = new RedissonPromise<>();
AsyncSemaphore lock = getSemaphore(channelName);
lock.acquire(new Runnable() {
@Override
public void run() {
final PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
if (entry == null) {
lock.release();
result.trySuccess(null);
@ -352,7 +352,7 @@ public class PublishSubscribeService {
freePubSubConnections.remove(entry);
freePubSubLock.release();
final Codec entryCodec;
Codec entryCodec;
if (topicType == PubSubType.PUNSUBSCRIBE) {
entryCodec = entry.getConnection().getPatternChannels().get(channelName);
} else {
@ -385,8 +385,8 @@ public class PublishSubscribeService {
return result;
}
public void punsubscribe(final ChannelName channelName, final AsyncSemaphore lock) {
final PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
public void punsubscribe(ChannelName channelName, AsyncSemaphore lock) {
PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
if (entry == null) {
lock.release();
return;
@ -444,8 +444,8 @@ public class PublishSubscribeService {
});
}
private void subscribe(final ChannelName channelName, final Collection<RedisPubSubListener<?>> listeners,
final Codec subscribeCodec) {
private void subscribe(ChannelName channelName, Collection<RedisPubSubListener<?>> listeners,
Codec subscribeCodec) {
RFuture<PubSubConnectionEntry> subscribeFuture = subscribe(subscribeCodec, channelName, listeners.toArray(new RedisPubSubListener[listeners.size()]));
subscribeFuture.onComplete((res, e) -> {
if (e != null) {
@ -457,8 +457,8 @@ public class PublishSubscribeService {
});
}
private void psubscribe(final ChannelName channelName, final Collection<RedisPubSubListener<?>> listeners,
final Codec subscribeCodec) {
private void psubscribe(ChannelName channelName, Collection<RedisPubSubListener<?>> listeners,
Codec subscribeCodec) {
RFuture<PubSubConnectionEntry> subscribeFuture = psubscribe(channelName, subscribeCodec, listeners.toArray(new RedisPubSubListener[listeners.size()]));
subscribeFuture.onComplete((res, e) -> {
if (e != null) {

27
redisson/src/main/java/org/redisson/remote/RemoteServiceKey.java
Unescape Escape View File

@ -15,8 +15,7 @@
*/
package org.redisson.remote;
import java.util.Collections;
import java.util.List;
import java.util.Arrays;
/**
*
@ -27,23 +26,23 @@ public class RemoteServiceKey {
private final Class<?> serviceInterface;
private final String methodName;
private final List<String> signatures;
private final long[] signature;
public RemoteServiceKey(Class<?> serviceInterface, String method, List<String> signatures) {
public RemoteServiceKey(Class<?> serviceInterface, String method, long[] signature) {
super();
this.serviceInterface = serviceInterface;
this.methodName = method;
this.signatures = Collections.unmodifiableList(signatures);
this.signature = signature;
}
public String getMethodName() {
return methodName;
}
public List<String> getSignatures() {
return signatures;
public long[] getSignature() {
return signature;
}
public Class<?> getServiceInterface() {
return serviceInterface;
}
@ -54,8 +53,8 @@ public class RemoteServiceKey {
final int prime = 31;
int result = 1;
result = prime * result + ((methodName == null) ? 0 : methodName.hashCode());
result = prime * result + ((signatures == null) ? 0 : signatures.hashCode());
result = prime * result + ((serviceInterface == null) ? 0 : serviceInterface.getName().hashCode());
result = prime * result + Arrays.hashCode(signature);
return result;
}
@ -71,16 +70,16 @@ public class RemoteServiceKey {
if (methodName == null) {
if (other.methodName != null)
return false;
} else if (!methodName.equals(other.methodName)) {
return false;
} else if (!signatures.equals(other.signatures)) {
} else if (!methodName.equals(other.methodName))
return false;
} else if (serviceInterface == null) {
if (serviceInterface == null) {
if (other.serviceInterface != null)
return false;
} else if (!serviceInterface.equals(other.serviceInterface))
return false;
if (!Arrays.equals(signature, other.signature))
return false;
return true;
}
}

17
redisson/src/main/java/org/redisson/remote/RemoteServiceRequest.java
Unescape Escape View File

@ -17,7 +17,6 @@ package org.redisson.remote;
import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import org.redisson.api.RemoteInvocationOptions;
@ -33,7 +32,7 @@ public class RemoteServiceRequest implements Serializable {
private String id;
private String executorId;
private String methodName;
private List<String> signatures;
private long[] signature;
private Object[] args;
private RemoteInvocationOptions options;
private long date;
@ -46,12 +45,12 @@ public class RemoteServiceRequest implements Serializable {
this.id = id;
}
public RemoteServiceRequest(String executorId, String id, String methodName, List<String> signatures, Object[] args, RemoteInvocationOptions options, long date) {
public RemoteServiceRequest(String executorId, String id, String methodName, long[] signature, Object[] args, RemoteInvocationOptions options, long date) {
super();
this.id = id;
this.executorId = executorId;
this.methodName = methodName;
this.signatures = signatures;
this.signature = signature;
this.args = args;
this.options = options;
this.date = date;
@ -73,10 +72,10 @@ public class RemoteServiceRequest implements Serializable {
return args;
}
public List<String> getSignatures() {
return signatures;
public long[] getSignature() {
return signature;
}
public RemoteInvocationOptions getOptions() {
return options;
}
@ -87,8 +86,8 @@ public class RemoteServiceRequest implements Serializable {
@Override
public String toString() {
return "RemoteServiceRequest [requestId=" + id + ", methodName=" + methodName + ", signatures=["
+ Arrays.toString(signatures.toArray()) + "], args="
return "RemoteServiceRequest [requestId=" + id + ", methodName=" + methodName + ", signature="
+ signature + ", args="
+ Arrays.toString(args) + ", options=" + options + ", date=" + date + "]";
}

Write Preview
Loading…
Cancel
Save