|
|
|
|
@ -20,13 +20,13 @@ import java.net.URI;
|
|
|
|
|
import java.util.concurrent.ExecutorService;
|
|
|
|
|
import java.util.concurrent.Executors;
|
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
|
|
|
|
|
|
|
|
import org.redisson.api.RFuture;
|
|
|
|
|
import org.redisson.client.handler.RedisChannelInitializer;
|
|
|
|
|
import org.redisson.client.handler.RedisChannelInitializer.Type;
|
|
|
|
|
import org.redisson.misc.RPromise;
|
|
|
|
|
import org.redisson.misc.RedissonPromise;
|
|
|
|
|
import org.redisson.misc.URIBuilder;
|
|
|
|
|
|
|
|
|
|
import io.netty.bootstrap.Bootstrap;
|
|
|
|
|
import io.netty.channel.Channel;
|
|
|
|
|
@ -34,12 +34,16 @@ import io.netty.channel.ChannelFuture;
|
|
|
|
|
import io.netty.channel.ChannelFutureListener;
|
|
|
|
|
import io.netty.channel.ChannelOption;
|
|
|
|
|
import io.netty.channel.EventLoopGroup;
|
|
|
|
|
import io.netty.channel.epoll.EpollDatagramChannel;
|
|
|
|
|
import io.netty.channel.epoll.EpollSocketChannel;
|
|
|
|
|
import io.netty.channel.group.ChannelGroup;
|
|
|
|
|
import io.netty.channel.group.ChannelGroupFuture;
|
|
|
|
|
import io.netty.channel.group.DefaultChannelGroup;
|
|
|
|
|
import io.netty.channel.nio.NioEventLoopGroup;
|
|
|
|
|
import io.netty.channel.socket.SocketChannel;
|
|
|
|
|
import io.netty.channel.socket.nio.NioSocketChannel;
|
|
|
|
|
import io.netty.channel.socket.nio.NioDatagramChannel;
|
|
|
|
|
import io.netty.resolver.AddressResolver;
|
|
|
|
|
import io.netty.resolver.dns.DnsAddressResolverGroup;
|
|
|
|
|
import io.netty.resolver.dns.DnsServerAddressStreamProviders;
|
|
|
|
|
import io.netty.util.HashedWheelTimer;
|
|
|
|
|
import io.netty.util.Timer;
|
|
|
|
|
import io.netty.util.concurrent.Future;
|
|
|
|
|
@ -53,9 +57,11 @@ import io.netty.util.concurrent.FutureListener;
|
|
|
|
|
*/
|
|
|
|
|
public class RedisClient {
|
|
|
|
|
|
|
|
|
|
private final AtomicReference<RFuture<InetSocketAddress>> resolveFuture = new AtomicReference<RFuture<InetSocketAddress>>();
|
|
|
|
|
private final Bootstrap bootstrap;
|
|
|
|
|
private final Bootstrap pubSubBootstrap;
|
|
|
|
|
private final InetSocketAddress addr;
|
|
|
|
|
private final URI addr;
|
|
|
|
|
private InetSocketAddress resolvedAddr;
|
|
|
|
|
private final ChannelGroup channels;
|
|
|
|
|
|
|
|
|
|
private ExecutorService executor;
|
|
|
|
|
@ -66,6 +72,7 @@ public class RedisClient {
|
|
|
|
|
private boolean hasOwnTimer;
|
|
|
|
|
private boolean hasOwnExecutor;
|
|
|
|
|
private boolean hasOwnGroup;
|
|
|
|
|
private boolean hasOwnResolver;
|
|
|
|
|
|
|
|
|
|
public static RedisClient create(RedisClientConfig config) {
|
|
|
|
|
return new RedisClient(config);
|
|
|
|
|
@ -85,12 +92,20 @@ public class RedisClient {
|
|
|
|
|
copy.setExecutor(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2));
|
|
|
|
|
hasOwnExecutor = true;
|
|
|
|
|
}
|
|
|
|
|
if (copy.getResolverGroup() == null) {
|
|
|
|
|
if (config.getSocketChannelClass() == EpollSocketChannel.class) {
|
|
|
|
|
copy.setResolverGroup(new DnsAddressResolverGroup(EpollDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault()));
|
|
|
|
|
} else {
|
|
|
|
|
copy.setResolverGroup(new DnsAddressResolverGroup(NioDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault()));
|
|
|
|
|
}
|
|
|
|
|
hasOwnResolver = true;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
this.config = copy;
|
|
|
|
|
this.executor = copy.getExecutor();
|
|
|
|
|
this.timer = copy.getTimer();
|
|
|
|
|
|
|
|
|
|
addr = new InetSocketAddress(copy.getAddress().getHost(), copy.getAddress().getPort());
|
|
|
|
|
addr = copy.getAddress();
|
|
|
|
|
|
|
|
|
|
channels = new DefaultChannelGroup(copy.getGroup().next());
|
|
|
|
|
bootstrap = createBootstrap(copy, Type.PLAIN);
|
|
|
|
|
@ -101,9 +116,9 @@ public class RedisClient {
|
|
|
|
|
|
|
|
|
|
private Bootstrap createBootstrap(RedisClientConfig config, Type type) {
|
|
|
|
|
Bootstrap bootstrap = new Bootstrap()
|
|
|
|
|
.resolver(config.getResolverGroup())
|
|
|
|
|
.channel(config.getSocketChannelClass())
|
|
|
|
|
.group(config.getGroup())
|
|
|
|
|
.remoteAddress(addr);
|
|
|
|
|
.group(config.getGroup());
|
|
|
|
|
|
|
|
|
|
bootstrap.handler(new RedisChannelInitializer(bootstrap, config, this, channels, type));
|
|
|
|
|
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout());
|
|
|
|
|
@ -112,92 +127,8 @@ public class RedisClient {
|
|
|
|
|
return bootstrap;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(String address) {
|
|
|
|
|
this(URIBuilder.create(address));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(URI address) {
|
|
|
|
|
this(new HashedWheelTimer(), Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2), new NioEventLoopGroup(), address);
|
|
|
|
|
hasOwnGroup = true;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(Timer timer, ExecutorService executor, EventLoopGroup group, URI address) {
|
|
|
|
|
this(timer, executor, group, address.getHost(), address.getPort());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(String host, int port) {
|
|
|
|
|
this(new HashedWheelTimer(), Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2), new NioEventLoopGroup(), NioSocketChannel.class, host, port, 10000, 10000);
|
|
|
|
|
hasOwnGroup = true;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(Timer timer, ExecutorService executor, EventLoopGroup group, String host, int port) {
|
|
|
|
|
this(timer, executor, group, NioSocketChannel.class, host, port, 10000, 10000);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(String host, int port, int connectTimeout, int commandTimeout) {
|
|
|
|
|
this(new HashedWheelTimer(), Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2), new NioEventLoopGroup(), NioSocketChannel.class, host, port, connectTimeout, commandTimeout);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/*
|
|
|
|
|
* Use {@link #create(RedisClientConfig)}
|
|
|
|
|
*
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
public RedisClient(final Timer timer, ExecutorService executor, EventLoopGroup group, Class<? extends SocketChannel> socketChannelClass, String host, int port,
|
|
|
|
|
int connectTimeout, int commandTimeout) {
|
|
|
|
|
RedisClientConfig config = new RedisClientConfig();
|
|
|
|
|
config.setTimer(timer).setExecutor(executor).setGroup(group).setSocketChannelClass(socketChannelClass)
|
|
|
|
|
.setAddress(host, port).setConnectTimeout(connectTimeout).setCommandTimeout(commandTimeout);
|
|
|
|
|
|
|
|
|
|
this.config = config;
|
|
|
|
|
this.executor = config.getExecutor();
|
|
|
|
|
this.timer = config.getTimer();
|
|
|
|
|
|
|
|
|
|
addr = new InetSocketAddress(config.getAddress().getHost(), config.getAddress().getPort());
|
|
|
|
|
|
|
|
|
|
channels = new DefaultChannelGroup(config.getGroup().next());
|
|
|
|
|
bootstrap = createBootstrap(config, Type.PLAIN);
|
|
|
|
|
pubSubBootstrap = createBootstrap(config, Type.PUBSUB);
|
|
|
|
|
|
|
|
|
|
this.commandTimeout = config.getCommandTimeout();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public String getIpAddr() {
|
|
|
|
|
return addr.getAddress().getHostAddress() + ":" + addr.getPort();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public InetSocketAddress getAddr() {
|
|
|
|
|
return addr;
|
|
|
|
|
return resolvedAddr;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public long getCommandTimeout() {
|
|
|
|
|
@ -219,42 +150,78 @@ public class RedisClient {
|
|
|
|
|
throw new RedisConnectionException("Unable to connect to: " + addr, e);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public RFuture<InetSocketAddress> resolveAddr() {
|
|
|
|
|
final RPromise<InetSocketAddress> promise = new RedissonPromise<InetSocketAddress>();
|
|
|
|
|
if (!resolveFuture.compareAndSet(null, promise)) {
|
|
|
|
|
return resolveFuture.get();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
AddressResolver<InetSocketAddress> resolver = (AddressResolver<InetSocketAddress>) bootstrap.config().resolver().getResolver(bootstrap.config().group().next());
|
|
|
|
|
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(addr.getHost(), addr.getPort()));
|
|
|
|
|
resolveFuture.addListener(new FutureListener<InetSocketAddress>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
|
|
|
|
|
if (!future.isSuccess()) {
|
|
|
|
|
promise.tryFailure(future.cause());
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
resolvedAddr = future.getNow();
|
|
|
|
|
promise.trySuccess(future.getNow());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
return promise;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public RFuture<RedisConnection> connectAsync() {
|
|
|
|
|
final RPromise<RedisConnection> f = new RedissonPromise<RedisConnection>();
|
|
|
|
|
ChannelFuture channelFuture = bootstrap.connect();
|
|
|
|
|
channelFuture.addListener(new ChannelFutureListener() {
|
|
|
|
|
|
|
|
|
|
RFuture<InetSocketAddress> addrFuture = resolveAddr();
|
|
|
|
|
addrFuture.addListener(new FutureListener<InetSocketAddress>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final ChannelFuture future) throws Exception {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
final RedisConnection c = RedisConnection.getFrom(future.channel());
|
|
|
|
|
c.getConnectionPromise().addListener(new FutureListener<RedisConnection>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final Future<RedisConnection> future) throws Exception {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
|
|
|
|
|
if (!future.isSuccess()) {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ChannelFuture channelFuture = bootstrap.connect(future.getNow());
|
|
|
|
|
channelFuture.addListener(new ChannelFutureListener() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final ChannelFuture future) throws Exception {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
final RedisConnection c = RedisConnection.getFrom(future.channel());
|
|
|
|
|
c.getConnectionPromise().addListener(new FutureListener<RedisConnection>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void run() {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
if (!f.trySuccess(c)) {
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
public void operationComplete(final Future<RedisConnection> future) throws Exception {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
@Override
|
|
|
|
|
public void run() {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
if (!f.trySuccess(c)) {
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
} else {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void run() {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
} else {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void run() {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
return f;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -268,39 +235,52 @@ public class RedisClient {
|
|
|
|
|
|
|
|
|
|
public RFuture<RedisPubSubConnection> connectPubSubAsync() {
|
|
|
|
|
final RPromise<RedisPubSubConnection> f = new RedissonPromise<RedisPubSubConnection>();
|
|
|
|
|
ChannelFuture channelFuture = pubSubBootstrap.connect();
|
|
|
|
|
channelFuture.addListener(new ChannelFutureListener() {
|
|
|
|
|
|
|
|
|
|
RFuture<InetSocketAddress> nameFuture = resolveAddr();
|
|
|
|
|
nameFuture.addListener(new FutureListener<InetSocketAddress>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final ChannelFuture future) throws Exception {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
final RedisPubSubConnection c = RedisPubSubConnection.getFrom(future.channel());
|
|
|
|
|
c.<RedisPubSubConnection>getConnectionPromise().addListener(new FutureListener<RedisPubSubConnection>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final Future<RedisPubSubConnection> future) throws Exception {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
|
|
|
|
|
if (!future.isSuccess()) {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ChannelFuture channelFuture = pubSubBootstrap.connect(future.getNow());
|
|
|
|
|
channelFuture.addListener(new ChannelFutureListener() {
|
|
|
|
|
@Override
|
|
|
|
|
public void operationComplete(final ChannelFuture future) throws Exception {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
final RedisPubSubConnection c = RedisPubSubConnection.getFrom(future.channel());
|
|
|
|
|
c.<RedisPubSubConnection>getConnectionPromise().addListener(new FutureListener<RedisPubSubConnection>() {
|
|
|
|
|
@Override
|
|
|
|
|
public void run() {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
if (!f.trySuccess(c)) {
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
public void operationComplete(final Future<RedisPubSubConnection> future) throws Exception {
|
|
|
|
|
pubSubBootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
@Override
|
|
|
|
|
public void run() {
|
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
|
if (!f.trySuccess(c)) {
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
c.closeAsync();
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
} else {
|
|
|
|
|
pubSubBootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void run() {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
} else {
|
|
|
|
|
bootstrap.config().group().execute(new Runnable() {
|
|
|
|
|
public void run() {
|
|
|
|
|
f.tryFailure(future.cause());
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
return f;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -339,6 +319,9 @@ public class RedisClient {
|
|
|
|
|
executor.awaitTermination(15, TimeUnit.SECONDS);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (hasOwnResolver) {
|
|
|
|
|
bootstrap.config().resolver().close();
|
|
|
|
|
}
|
|
|
|
|
if (hasOwnGroup) {
|
|
|
|
|
bootstrap.config().group().shutdownGracefully();
|
|
|
|
|
}
|
|
|
|
|
|
Nikita
7 years ago