RedisClientFactory.java

/**
 * @author Johno Crawford (johno@sulake.com)
 */
@Component
class RedisClientFactory {

    private static final Logger logger = LogManager.getLogger(RedisClientFactory.class);

    private static final int RETRY_INTERVAL = 500;
    private static final int RETRY_ATTEMPTS = 1;
    private static final int TIMEOUT = 2000;

    @Value("${redis.enabled:false}")
    private boolean redisEnabled;

    @Autowired
    private RedisEventLoop redisEventLoop;

    private DnsAddressResolverGroup dnsAddressResolverGroup;

    private final Map<Integer, RedissonClientPair> redissonClientMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void initialize() {
        dnsAddressResolverGroup = new DnsAddressResolverGroup(new DnsNameResolverBuilder()
                .ttl(0, 10) // same value as networkaddress.cache.ttl
                .datagramChannelStrategy(DnsNameResolverChannelStrategy.ChannelPerResolution)
                .datagramChannelType(redisEventLoop.isNativeEpoll() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                .nameServerProvider(DnsServerAddressStreamProviders.platformDefault()));
    }

    RedissonReactiveClient getRedisReactiveClient(String redisNodes, int database) {
        if (!redisEnabled) {
            logger.info("Redis disabled");
            return null;
        }

        RedissonClientPair redissonClient = redissonClientMap.get(database);
        return redissonClient != null
                ? redissonClient.redissonReactiveClient
                : redissonClientMap.computeIfAbsent(database, db ->
                        createRedissonClient(redisNodes, db, TIMEOUT, RETRY_ATTEMPTS, RETRY_INTERVAL)).redissonReactiveClient;
    }

    RedissonClient getRedisClient(String redisNodes, int database) {
        if (!redisEnabled) {
            logger.info("Redis disabled");
            return null;
        }

        RedissonClientPair redissonClient = redissonClientMap.get(database);
        return redissonClient != null
            ? redissonClient.redissonClient
            : redissonClientMap.computeIfAbsent(database, db ->
                    createRedissonClient(redisNodes, db, TIMEOUT, RETRY_ATTEMPTS, RETRY_INTERVAL)).redissonClient;
    }

    RedissonClient getRedisClient(String redisNodes, int database, int timeout, int retryAttempts, int retryInterval) {
        if (!redisEnabled) {
            logger.info("Redis disabled");
            return null;
        }

        RedissonClientPair redissonClient = redissonClientMap.get(database);
        return redissonClient != null
                ? redissonClient.redissonClient
                : redissonClientMap.computeIfAbsent(database, db -> createRedissonClient(redisNodes, db,
                timeout, retryAttempts, retryInterval)).redissonClient;
    }

    private RedissonClientPair createRedissonClient(String redisNodes, int database, int timeout, int retryAttempts, int retryInterval) {
        final Config config = new Config();
        config.setCodec(new SnappyCodec());
        config.setEventLoopGroup(redisEventLoop.getWorkerGroup());
        config.setTransportMode(redisEventLoop.isNativeEpoll() ?
                                        TransportMode.EPOLL : TransportMode.NIO);
        // inject shared dns address resolver to save memory
        config.setAddressResolverGroupFactory((channelType, socketChannelType, nameServerProvider) -> dnsAddressResolverGroup);
        config.setReferenceEnabled(false);

        if (redisNodes.indexOf(',') > -1) {
            final ReplicatedServersConfig clusterConfig = new ReplicatedServersConfig();
            clusterConfig.setScanInterval(2500);
            clusterConfig.setDnsMonitoringInterval(2500);
            clusterConfig.setDatabase(database);
            clusterConfig.setTcpNoDelay(true);
            clusterConfig.setKeepAlive(true);
            for (String addr : StringUtils.tokenizeToStringArray(redisNodes, ",", true, true)) {
                clusterConfig.addNodeAddress(addSchema(addr));
            }
            clusterConfig.setTimeout(timeout);
            clusterConfig.setRetryAttempts(retryAttempts);
            clusterConfig.setRetryInterval(retryInterval);
            clusterConfig.setReadMode(ReadMode.SLAVE);
            config.useCustomServers(new ReplicatedConnectionManager(clusterConfig, config)); // shared connection manager
        } else {
            SingleServerConfig singleServerConfig = config.useSingleServer();
            singleServerConfig.setDatabase(database);
            singleServerConfig.setTimeout(timeout);
            singleServerConfig.setRetryAttempts(retryAttempts);
            singleServerConfig.setRetryInterval(retryInterval);
            singleServerConfig.setTcpNoDelay(true);
            singleServerConfig.setKeepAlive(true);
            singleServerConfig.setAddress(addSchema(redisNodes));
        }

        return new RedissonClientPair(Redisson.create(config), Redisson.createReactive(config));
    }

    static class RedissonClientPair {

        private final RedissonClient redissonClient;
        private final RedissonReactiveClient redissonReactiveClient;

        public RedissonClientPair(@NotNull RedissonClient redissonClient, @NotNull RedissonReactiveClient redissonReactiveClient) {
            this.redissonClient = redissonClient;
            this.redissonReactiveClient = redissonReactiveClient;
        }
    }

    @PreDestroy
    public void destroy() {
        logger.info("Shutting down Redisson clients");
        redissonClientMap.entrySet().parallelStream().forEach(entry -> {
            entry.getValue().redissonClient.shutdown(0, 15, TimeUnit.SECONDS);
            entry.getValue().redissonReactiveClient.shutdown();
            logger.info("Redisson client db{} shutdown", entry.getKey());
        });
        dnsAddressResolverGroup.close();
        logger.info("Redisson clients shutdown");
    }

    private static String addSchema(String address) {
        return address.startsWith("redis") ? address : "redis://" + address;
    }
}