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

MasterSlaveConnectionManager#getEntry(java.net.InetSocketAddress) optimization. #1081

Browse Source
pull/1085/head
Nikita 8 years ago
parent 0be813da07
commit 063dcabc8c
5 changed files with 45 additions and 47 deletions
Show Stats Download Patch File Download Diff File

16
redisson/src/main/java/org/redisson/RedissonKeys.java
Unescape Escape View File

@ -23,7 +23,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@ -207,7 +206,7 @@ public class RedissonKeys implements RKeys {
final RPromise<Long> result = commandExecutor.getConnectionManager().newPromise();
final AtomicReference<Throwable> failed = new AtomicReference<Throwable>();
final AtomicLong count = new AtomicLong();
Set<MasterSlaveEntry> entries = commandExecutor.getConnectionManager().getEntrySet();
Collection<MasterSlaveEntry> entries = commandExecutor.getConnectionManager().getEntrySet();
final AtomicLong executed = new AtomicLong(entries.size());
final FutureListener<Long> listener = new FutureListener<Long>() {
@Override
@ -301,14 +300,13 @@ public class RedissonKeys implements RKeys {
Map<MasterSlaveEntry, List<String>> range2key = new HashMap<MasterSlaveEntry, List<String>>();
for (String key : keys) {
int slot = commandExecutor.getConnectionManager().calcSlot(key);
for (MasterSlaveEntry entry : commandExecutor.getConnectionManager().getEntrySet()) {
List<String> list = range2key.get(entry);
if (list == null) {
list = new ArrayList<String>();
range2key.put(entry, list);
}
list.add(key);
MasterSlaveEntry entry = commandExecutor.getConnectionManager().getEntry(slot);
List<String> list = range2key.get(entry);
if (list == null) {
list = new ArrayList<String>();
range2key.put(entry, list);
}
list.add(key);
}
final RPromise<Long> result = commandExecutor.getConnectionManager().newPromise();

17
redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java
Unescape Escape View File

@ -568,8 +568,8 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
for (Integer slot : removedSlots) {
MasterSlaveEntry entry = removeMaster(slot);
entry.removeSlotRange(slot);
MasterSlaveEntry entry = getEntry(slot);
removeMaster(slot);
if (entry.getSlotRanges().isEmpty()) {
entry.shutdownMasterAsync();
log.info("{} master and slaves for it removed", entry.getClient().getAddr());
@ -584,12 +584,11 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
for (final Integer slot : addedSlots) {
ClusterPartition partition = find(newPartitions, slot);
for (MasterSlaveEntry entry : getEntrySet()) {
if (entry.getClient().getAddr().equals(partition.getMasterAddr())) {
addEntry(slot, entry);
lastPartitions.put(slot, partition);
break;
}
MasterSlaveEntry entry = getEntry(partition.getMasterAddr());
if (entry != null && entry.getClient().getAddr().equals(partition.getMasterAddr())) {
addEntry(slot, entry);
lastPartitions.put(slot, partition);
break;
}
}
}
@ -611,7 +610,6 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
MasterSlaveEntry entry = getEntry(currentPartition.getMasterAddr());
for (Integer slot : addedSlots) {
entry.addSlotRange(slot);
addEntry(slot, entry);
lastPartitions.put(slot, currentPartition);
}
@ -623,7 +621,6 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
removedSlots.removeAll(newPartition.getSlots());
for (Integer removeSlot : removedSlots) {
if (lastPartitions.remove(removeSlot, currentPartition)) {
entry.removeSlotRange(removeSlot);
removeMaster(removeSlot);
}
}

13
redisson/src/main/java/org/redisson/command/CommandAsyncService.java
Unescape Escape View File

@ -16,11 +16,13 @@
package org.redisson.command;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -71,9 +73,6 @@ import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import java.util.AbstractMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
/**
*
@ -198,7 +197,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
@Override
public <T, R> RFuture<Collection<R>> readAllAsync(RedisCommand<T> command, Object... params) {
final RPromise<Collection<R>> mainPromise = connectionManager.newPromise();
final Set<MasterSlaveEntry> nodes = connectionManager.getEntrySet();
final Collection<MasterSlaveEntry> nodes = connectionManager.getEntrySet();
final List<R> results = new ArrayList<R>();
final AtomicInteger counter = new AtomicInteger(nodes.size());
FutureListener<R> listener = new FutureListener<R>() {
@ -288,7 +287,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
private <T, R> RFuture<R> allAsync(boolean readOnlyMode, RedisCommand<T> command, final SlotCallback<T, R> callback, Object... params) {
final RPromise<R> mainPromise = connectionManager.newPromise();
final Set<MasterSlaveEntry> nodes = connectionManager.getEntrySet();
final Collection<MasterSlaveEntry> nodes = connectionManager.getEntrySet();
final AtomicInteger counter = new AtomicInteger(nodes.size());
FutureListener<T> listener = new FutureListener<T>() {
@Override
@ -413,7 +412,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
public <T, R> RFuture<R> evalAllAsync(boolean readOnlyMode, RedisCommand<T> command, final SlotCallback<T, R> callback, String script, List<Object> keys, Object... params) {
final RPromise<R> mainPromise = connectionManager.newPromise();
final Set<MasterSlaveEntry> entries = connectionManager.getEntrySet();
final Collection<MasterSlaveEntry> entries = connectionManager.getEntrySet();
final AtomicInteger counter = new AtomicInteger(entries.size());
FutureListener<T> listener = new FutureListener<T>() {

3
redisson/src/main/java/org/redisson/connection/ConnectionManager.java
Unescape Escape View File

@ -18,7 +18,6 @@ package org.redisson.connection;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@ -85,7 +84,7 @@ public interface ConnectionManager {
Codec getCodec();
Set<MasterSlaveEntry> getEntrySet();
Collection<MasterSlaveEntry> getEntrySet();
MasterSlaveEntry getEntry(int slot);

43
redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
Unescape Escape View File

@ -24,9 +24,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
@ -130,7 +128,8 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
protected MasterSlaveServersConfig config;
private final Map<Integer, MasterSlaveEntry> entries = PlatformDependent.newConcurrentHashMap();
private final Map<Integer, MasterSlaveEntry> slot2entry = PlatformDependent.newConcurrentHashMap();
private final Map<InetSocketAddress, MasterSlaveEntry> addr2entry = PlatformDependent.newConcurrentHashMap();
private final RPromise<Boolean> shutdownPromise;
@ -234,8 +233,9 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
return codec;
}
public Set<MasterSlaveEntry> getEntrySet() {
return new HashSet<MasterSlaveEntry>(entries.values());
@Override
public Collection<MasterSlaveEntry> getEntrySet() {
return addr2entry.values();
}
protected void initTimer(MasterSlaveServersConfig config) {
@ -672,29 +672,34 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public MasterSlaveEntry getEntry(InetSocketAddress addr) {
// TODO optimize
for (Entry<Integer, MasterSlaveEntry> entry : entries.entrySet()) {
if (entry.getValue().getClient().getAddr().equals(addr)) {
return entry.getValue();
}
}
return null;
return addr2entry.get(addr);
}
@Override
public MasterSlaveEntry getEntry(int slot) {
return entries.get(slot);
return slot2entry.get(slot);
}
protected void changeMaster(int slot, URI address) {
getEntry(slot).changeMaster(address);
protected final void changeMaster(int slot, URI address) {
MasterSlaveEntry entry = getEntry(slot);
addr2entry.remove(entry.getClient().getAddr());
entry.changeMaster(address);
addr2entry.put(entry.getClient().getAddr(), entry);
}
protected void addEntry(Integer slot, MasterSlaveEntry entry) {
entries.put(slot, entry);
protected final void addEntry(Integer slot, MasterSlaveEntry entry) {
slot2entry.put(slot, entry);
entry.addSlotRange(slot);
addr2entry.put(entry.getClient().getAddr(), entry);
}
protected MasterSlaveEntry removeMaster(Integer slot) {
return entries.remove(slot);
MasterSlaveEntry entry = slot2entry.remove(slot);
entry.removeSlotRange(slot);
if (entry.getSlotRanges().isEmpty()) {
addr2entry.remove(entry.getClient().getAddr());
}
return entry;
}
@Override
@ -812,7 +817,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
shutdownPromise.trySuccess(true);
shutdownLatch.awaitUninterruptibly();
for (MasterSlaveEntry entry : entries.values()) {
for (MasterSlaveEntry entry : getEntrySet()) {
entry.shutdown();
}

Write Preview
Loading…
Cancel
Save