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/4640/merge
Nikita Koksharov 2 years ago
parent eeba9d0488
commit 9a2090664e
9 changed files with 82 additions and 69 deletions
Show Stats Download Patch File Download Diff File

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

@ -19,7 +19,6 @@ import io.netty.resolver.AddressResolver;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import org.redisson.api.NatMapper;
import org.redisson.api.NodeType;
import org.redisson.api.RFuture;
import org.redisson.client.*;
@ -27,10 +26,7 @@ import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.cluster.ClusterNodeInfo.Flag;
import org.redisson.cluster.ClusterPartition.Type;
import org.redisson.config.ClusterServersConfig;
import org.redisson.config.Config;
import org.redisson.config.MasterSlaveServersConfig;
import org.redisson.config.ReadMode;
import org.redisson.config.*;
import org.redisson.connection.*;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.RedisURI;
@ -63,25 +59,30 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
private String configEndpointHostName;
private final NatMapper natMapper;
private final AtomicReferenceArray<MasterSlaveEntry> slot2entry = new AtomicReferenceArray<>(MAX_SLOT);
private final Map<RedisClient, MasterSlaveEntry> client2entry = new ConcurrentHashMap<>();
private ClusterServersConfig cfg;
public ClusterConnectionManager(ClusterServersConfig cfg, Config config, UUID id) {
super(config, id);
super(cfg, config, id);
}
@Override
protected MasterSlaveServersConfig create(BaseMasterSlaveServersConfig<?> cfg) {
this.cfg = (ClusterServersConfig) cfg;
return super.create(cfg);
}
@Override
public void connect() {
if (cfg.getNodeAddresses().isEmpty()) {
throw new IllegalArgumentException("At least one cluster node should be defined!");
}
this.natMapper = cfg.getNatMapper();
this.config = create(cfg);
initTimer(this.config);
Throwable lastException = null;
List<String> failedMasters = new ArrayList<String>();
List<String> failedMasters = new ArrayList<>();
for (String address : cfg.getNodeAddresses()) {
RedisURI addr = new RedisURI(address);
CompletionStage<RedisConnection> connectionFuture = connectToNode(cfg, addr, addr.getHost());
@ -91,14 +92,14 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
if (cfg.getNodeAddresses().size() == 1 && !addr.isIP()) {
configEndpointHostName = addr.getHost();
}
clusterNodesCommand = RedisCommands.CLUSTER_NODES;
if (addr.isSsl()) {
clusterNodesCommand = RedisCommands.CLUSTER_NODES_SSL;
}
List<ClusterNodeInfo> nodes = connection.sync(clusterNodesCommand);
StringBuilder nodesValue = new StringBuilder();
for (ClusterNodeInfo clusterNodeInfo : nodes) {
nodesValue.append(clusterNodeInfo.getNodeInfo()).append("\n");
@ -106,7 +107,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
log.info("Redis cluster nodes configuration got from {}:\n{}", connection.getRedisClient().getAddr(), nodesValue);
lastClusterNode = addr;
CompletableFuture<Collection<ClusterPartition>> partitionsFuture = parsePartitions(nodes);
Collection<ClusterPartition> partitions;
try {
@ -163,7 +164,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
throw new RedisConnectionException("Not all slots covered! Only " + lastPartitions.size() + " slots are available. Set checkSlotsCoverage = false to avoid this check. Failed masters according to cluster status: " + failedMasters, lastException);
}
}
scheduleClusterChangeCheck(cfg);
}
@ -818,7 +819,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
@Override
public RedisURI applyNatMap(RedisURI address) {
return natMapper.map(address);
return cfg.getNatMapper().map(address);
}
private CompletableFuture<Collection<ClusterPartition>> parsePartitions(List<ClusterNodeInfo> nodes) {

19
redisson/src/main/java/org/redisson/config/ConfigSupport.java
Unescape Escape View File

@ -182,26 +182,31 @@ public class ConfigSupport {
public static ConnectionManager createConnectionManager(Config configCopy) {
UUID id = UUID.randomUUID();
ConnectionManager cm = null;
if (configCopy.getMasterSlaveServersConfig() != null) {
validate(configCopy.getMasterSlaveServersConfig());
return new MasterSlaveConnectionManager(configCopy.getMasterSlaveServersConfig(), configCopy, id);
cm = new MasterSlaveConnectionManager(configCopy.getMasterSlaveServersConfig(), configCopy, id);
} else if (configCopy.getSingleServerConfig() != null) {
validate(configCopy.getSingleServerConfig());
return new SingleConnectionManager(configCopy.getSingleServerConfig(), configCopy, id);
cm = new SingleConnectionManager(configCopy.getSingleServerConfig(), configCopy, id);
} else if (configCopy.getSentinelServersConfig() != null) {
validate(configCopy.getSentinelServersConfig());
return new SentinelConnectionManager(configCopy.getSentinelServersConfig(), configCopy, id);
cm = new SentinelConnectionManager(configCopy.getSentinelServersConfig(), configCopy, id);
} else if (configCopy.getClusterServersConfig() != null) {
validate(configCopy.getClusterServersConfig());
return new ClusterConnectionManager(configCopy.getClusterServersConfig(), configCopy, id);
cm = new ClusterConnectionManager(configCopy.getClusterServersConfig(), configCopy, id);
} else if (configCopy.getReplicatedServersConfig() != null) {
validate(configCopy.getReplicatedServersConfig());
return new ReplicatedConnectionManager(configCopy.getReplicatedServersConfig(), configCopy, id);
cm = new ReplicatedConnectionManager(configCopy.getReplicatedServersConfig(), configCopy, id);
} else if (configCopy.getConnectionManager() != null) {
return configCopy.getConnectionManager();
}else {
cm = configCopy.getConnectionManager();
}
if (cm == null) {
throw new IllegalArgumentException("server(s) address(es) not defined!");
}
cm.connect();
return cm;
}
private static void validate(SingleServerConfig config) {

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

@ -44,7 +44,9 @@ import java.util.concurrent.TimeUnit;
*
*/
public interface ConnectionManager {
void connect();
RedisURI applyNatMap(RedisURI address);
CompletableFuture<RedisURI> resolveIP(RedisURI address);

7
redisson/src/main/java/org/redisson/connection/IdleConnectionWatcher.java
Unescape Escape View File

@ -21,7 +21,6 @@ import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import org.redisson.client.RedisConnection;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.config.MasterSlaveServersConfig;
import org.redisson.misc.AsyncSemaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -62,7 +61,7 @@ public class IdleConnectionWatcher {
private final Map<ClientConnectionsEntry, List<Entry>> entries = new ConcurrentHashMap<>();
private final ScheduledFuture<?> monitorFuture;
public IdleConnectionWatcher(ConnectionManager manager, MasterSlaveServersConfig config) {
public IdleConnectionWatcher(ConnectionManager manager) {
monitorFuture = manager.getGroup().scheduleWithFixedDelay(() -> {
long currTime = System.nanoTime();
for (Entry entry : entries.values().stream().flatMap(m -> m.stream()).collect(Collectors.toList())) {
@ -80,7 +79,7 @@ public class IdleConnectionWatcher {
continue;
}
if (timeInPool > config.getIdleConnectionTimeout()
if (timeInPool > manager.getConfig().getIdleConnectionTimeout()
&& validateAmount(entry)
&& entry.deleteHandler.apply(c)) {
ChannelFuture future = c.closeAsync();
@ -93,7 +92,7 @@ public class IdleConnectionWatcher {
}
}
}
}, config.getIdleConnectionTimeout(), config.getIdleConnectionTimeout(), TimeUnit.MILLISECONDS);
}, manager.getConfig().getIdleConnectionTimeout(), manager.getConfig().getIdleConnectionTimeout(), TimeUnit.MILLISECONDS);
}
private boolean validateAmount(Entry entry) {

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

@ -133,20 +133,23 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
private final Map<RedisURI, RedisConnection> nodeConnections = new ConcurrentHashMap<>();
public MasterSlaveConnectionManager(MasterSlaveServersConfig cfg, Config config, UUID id) {
public MasterSlaveConnectionManager(BaseMasterSlaveServersConfig<?> cfg, Config config, UUID id) {
this(config, id);
this.config = cfg;
if (cfg.getSlaveAddresses().isEmpty()
&& (cfg.getReadMode() == ReadMode.SLAVE || cfg.getReadMode() == ReadMode.MASTER_SLAVE)) {
throw new IllegalArgumentException("Slaves aren't defined. readMode can't be SLAVE or MASTER_SLAVE");
if (cfg instanceof MasterSlaveServersConfig) {
this.config = (MasterSlaveServersConfig) cfg;
if (this.config.getSlaveAddresses().isEmpty()
&& (this.config.getReadMode() == ReadMode.SLAVE || this.config.getReadMode() == ReadMode.MASTER_SLAVE)) {
throw new IllegalArgumentException("Slaves aren't defined. readMode can't be SLAVE or MASTER_SLAVE");
}
} else {
this.config = create(cfg);
}
initTimer(cfg);
initSingleEntry();
initTimer();
}
protected MasterSlaveConnectionManager(Config cfg, UUID id) {
private MasterSlaveConnectionManager(Config cfg, UUID id) {
this.id = id.toString();
Version.logVersion();
@ -302,7 +305,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
return Collections.emptyList();
}
protected void initTimer(MasterSlaveServersConfig config) {
private void initTimer() {
int[] timeouts = new int[]{config.getRetryInterval(), config.getTimeout()};
Arrays.sort(timeouts);
int minTimeout = timeouts[0];
@ -316,11 +319,11 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
timer = new HashedWheelTimer(new DefaultThreadFactory("redisson-timer"), minTimeout, TimeUnit.MILLISECONDS, 1024, false);
connectionWatcher = new IdleConnectionWatcher(this, config);
subscribeService = new PublishSubscribeService(this, config);
connectionWatcher = new IdleConnectionWatcher(this);
subscribeService = new PublishSubscribeService(this);
}
protected void initSingleEntry() {
public void connect() {
try {
if (config.checkSkipSlavesInit()) {
masterSlaveEntry = new SingleEntry(this, config);

13
redisson/src/main/java/org/redisson/connection/ReplicatedConnectionManager.java
Unescape Escape View File

@ -63,12 +63,14 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
slave
}
public ReplicatedConnectionManager(ReplicatedServersConfig cfg, Config config, UUID id) {
super(config, id);
private ReplicatedServersConfig cfg;
this.config = create(cfg);
initTimer(this.config);
public ReplicatedConnectionManager(ReplicatedServersConfig cfg, Config config, UUID id) {
super(cfg, config, id);
}
@Override
public void connect() {
for (String address : cfg.getNodeAddresses()) {
RedisURI addr = new RedisURI(address);
CompletionStage<RedisConnection> connectionFuture = connectToNode(cfg, addr, addr.getHost());
@ -101,7 +103,7 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
log.warn("ReadMode = {}, but slave nodes are not found! Please specify all nodes in replicated mode.", this.config.getReadMode());
}
initSingleEntry();
super.connect();
scheduleMasterChangeCheck(cfg);
}
@ -113,6 +115,7 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
@Override
protected MasterSlaveServersConfig create(BaseMasterSlaveServersConfig<?> cfg) {
this.cfg = (ReplicatedServersConfig) cfg;
MasterSlaveServersConfig res = super.create(cfg);
res.setDatabase(((ReplicatedServersConfig) cfg).getDatabase());
return res;

36
redisson/src/main/java/org/redisson/connection/SentinelConnectionManager.java
Unescape Escape View File

@ -58,25 +58,14 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
private final AddressResolver<InetSocketAddress> sentinelResolver;
private final Set<RedisURI> disconnectedSentinels = Collections.newSetFromMap(new ConcurrentHashMap<>());
private final RedisStrictCommand<RedisURI> masterHostCommand;
private RedisStrictCommand<RedisURI> masterHostCommand;
private boolean usePassword = false;
private String scheme;
private final SentinelServersConfig cfg;
private SentinelServersConfig cfg;
public SentinelConnectionManager(SentinelServersConfig cfg, Config config, UUID id) {
super(config, id);
if (cfg.getMasterName() == null) {
throw new IllegalArgumentException("masterName parameter is not defined!");
}
if (cfg.getSentinelAddresses().isEmpty()) {
throw new IllegalArgumentException("At least one sentinel node should be defined!");
}
this.config = create(cfg);
this.cfg = cfg;
initTimer(this.config);
super(cfg, config, id);
this.sentinelResolver = resolverGroup.getResolver(getGroup().next());
@ -88,7 +77,10 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
sentinelHosts.add(addr);
}
}
}
@Override
public void connect() {
checkAuth(cfg);
if ("redis".equals(scheme)) {
@ -205,9 +197,9 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
if (this.config.getReadMode() != ReadMode.MASTER && this.config.getSlaveAddresses().isEmpty()) {
log.warn("ReadMode = {}, but slave nodes are not found!", this.config.getReadMode());
}
initSingleEntry();
super.connect();
scheduleChangeCheck(cfg, null);
}
@ -642,8 +634,16 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
@Override
protected MasterSlaveServersConfig create(BaseMasterSlaveServersConfig<?> cfg) {
this.cfg = (SentinelServersConfig) cfg;
if (this.cfg.getMasterName() == null) {
throw new IllegalArgumentException("masterName parameter is not defined!");
}
if (this.cfg.getSentinelAddresses().isEmpty()) {
throw new IllegalArgumentException("At least one sentinel node should be defined!");
}
MasterSlaveServersConfig res = super.create(cfg);
res.setDatabase(((SentinelServersConfig) cfg).getDatabase());
res.setDatabase(this.cfg.getDatabase());
return res;
}

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

@ -101,10 +101,10 @@ public class PublishSubscribeService {
private final LockPubSub lockPubSub = new LockPubSub(this);
public PublishSubscribeService(ConnectionManager connectionManager, MasterSlaveServersConfig config) {
public PublishSubscribeService(ConnectionManager connectionManager) {
super();
this.connectionManager = connectionManager;
this.config = config;
this.config = connectionManager.getConfig();
for (int i = 0; i < locks.length; i++) {
locks[i] = new AsyncSemaphore(1);
}

2
redisson/src/test/java/org/redisson/RedissonTest.java
Unescape Escape View File

@ -81,7 +81,7 @@ public class RedissonTest extends BaseTest {
}
ex.shutdown();
assertThat(ex.awaitTermination(7, TimeUnit.SECONDS)).isTrue();
assertThat(ex.awaitTermination(8, TimeUnit.SECONDS)).isTrue();
inst.shutdown();
}

Write Preview
Loading…
Cancel
Save