DATAREDIS-369 - Avoid resource leak in RedisCache.
Introduce dedicated types RedisCache is supposed to work upon. We still stick to the contract defined by previous versions to assure source and binary compatibility. In a next step RedisCache should be decoupled from o.s.cache.Cache by an additional indirection via RedisCacheCache. Original Pull Request: #122
This commit is contained in:
Christoph Strobl
@@ -16,6 +16,9 @@
|
||||
|
||||
package org.springframework.data.redis.cache;
|
||||
|
||||
import static org.springframework.util.Assert.*;
|
||||
import static org.springframework.util.ObjectUtils.*;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Set;
|
||||
|
||||
@@ -28,7 +31,6 @@ import org.springframework.data.redis.core.RedisCallback;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Cache implementation on top of Redis.
|
||||
@@ -40,16 +42,10 @@ import org.springframework.util.Assert;
|
||||
@SuppressWarnings("unchecked")
|
||||
public class RedisCache implements Cache {
|
||||
|
||||
private static final byte[] REMOVE_KEYS_BY_PATTERN_LUA = "local keys = redis.call('KEYS', ARGV[1]); local keysCount = table.getn(keys); if(keysCount > 0) then for _, key in ipairs(keys) do redis.call('del', key); end; end; return keysCount;"
|
||||
.getBytes();
|
||||
private static final int PAGE_SIZE = 128;
|
||||
private final String name;
|
||||
@SuppressWarnings("rawtypes") private final RedisTemplate template;
|
||||
private final byte[] prefix;
|
||||
private final byte[] setName;
|
||||
private final byte[] cacheLockName;
|
||||
private long WAIT_FOR_LOCK = 300;
|
||||
private final long expiration;
|
||||
@SuppressWarnings("rawtypes")//
|
||||
private final RedisTemplate template;
|
||||
private final RedisCacheMetadata cacheMetadata;
|
||||
private final CacheValueAccessor cacheValueAccessor;
|
||||
|
||||
/**
|
||||
* Constructs a new <code>RedisCache</code> instance.
|
||||
@@ -62,41 +58,12 @@ public class RedisCache implements Cache {
|
||||
public RedisCache(String name, byte[] prefix, RedisTemplate<? extends Object, ? extends Object> template,
|
||||
long expiration) {
|
||||
|
||||
Assert.hasText(name, "non-empty cache name is required");
|
||||
this.name = name;
|
||||
hasText(name, "non-empty cache name is required");
|
||||
this.cacheMetadata = new RedisCacheMetadata(name, prefix);
|
||||
this.cacheMetadata.setDefaultExpiration(expiration);
|
||||
|
||||
this.template = template;
|
||||
this.prefix = prefix;
|
||||
this.expiration = expiration;
|
||||
|
||||
StringRedisSerializer stringSerializer = new StringRedisSerializer();
|
||||
|
||||
// name of the set holding the keys
|
||||
this.setName = stringSerializer.serialize(name + "~keys");
|
||||
this.cacheLockName = stringSerializer.serialize(name + "~lock");
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc} This implementation simply returns the RedisTemplate used for configuring the cache, giving access to
|
||||
* the underlying Redis store.
|
||||
*/
|
||||
public Object getNativeCache() {
|
||||
return template;
|
||||
}
|
||||
|
||||
public ValueWrapper get(final Object key) {
|
||||
return (ValueWrapper) template.execute(new RedisCallback<ValueWrapper>() {
|
||||
|
||||
public ValueWrapper doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
waitForLock(connection);
|
||||
byte[] bs = connection.get(computeKey(key));
|
||||
Object value = template.getValueSerializer() != null ? template.getValueSerializer().deserialize(bs) : bs;
|
||||
return (bs == null ? null : new SimpleValueWrapper(value));
|
||||
}
|
||||
}, true);
|
||||
this.cacheValueAccessor = new CacheValueAccessor(template.getValueSerializer());
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -114,196 +81,538 @@ public class RedisCache implements Cache {
|
||||
return wrapper == null ? null : (T) wrapper.get();
|
||||
}
|
||||
|
||||
public void put(final Object key, final Object value) {
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#get(java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public ValueWrapper get(Object key) {
|
||||
return get(new RedisCacheKey(key).usePrefix(this.cacheMetadata.getKeyPrefix()).withKeySerializer(
|
||||
template.getKeySerializer()));
|
||||
}
|
||||
|
||||
final byte[] keyBytes = computeKey(key);
|
||||
final byte[] valueBytes = convertToBytesIfNecessary(template.getValueSerializer(), value);
|
||||
/**
|
||||
* Return the value to which this cache maps the specified key.
|
||||
*
|
||||
* @param cacheKey the key whose associated value is to be returned via its binary representation.
|
||||
* @return the {@link RedisCacheElement} stored at given key or {@literal null} if no value found for key.
|
||||
* @since 1.5
|
||||
*/
|
||||
public RedisCacheElement get(final RedisCacheKey cacheKey) {
|
||||
|
||||
template.execute(new RedisCallback<Object>() {
|
||||
public Object doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
notNull(cacheKey, "CacheKey must not be null!");
|
||||
return (RedisCacheElement) template.execute(new AbstractRedisCacheCallback<RedisCacheElement>(
|
||||
new RedisCacheElement(cacheKey, null), cacheMetadata) {
|
||||
|
||||
waitForLock(connection);
|
||||
@Override
|
||||
public RedisCacheElement doInRedis(RedisCacheElement element, RedisConnection connection)
|
||||
throws DataAccessException {
|
||||
|
||||
connection.multi();
|
||||
|
||||
connection.set(keyBytes, valueBytes);
|
||||
|
||||
if (shouldTrackKeys()) {
|
||||
connection.zAdd(setName, 0, keyBytes);
|
||||
}
|
||||
if (expiration > 0) {
|
||||
connection.expire(keyBytes, expiration);
|
||||
// update the expiration of the set of keys as well
|
||||
|
||||
if (shouldTrackKeys()) {
|
||||
connection.expire(setName, expiration);
|
||||
}
|
||||
}
|
||||
connection.exec();
|
||||
|
||||
return null;
|
||||
byte[] bs = connection.get(element.getKeyBytes());
|
||||
Object value = template.getValueSerializer() != null ? template.getValueSerializer().deserialize(bs) : bs;
|
||||
return (bs == null ? null : new RedisCacheElement(element.getKey(), value));
|
||||
}
|
||||
}, true);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#put(java.lang.Object, java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public void put(final Object key, final Object value) {
|
||||
|
||||
put(new RedisCacheElement(new RedisCacheKey(key).usePrefix(cacheMetadata.getKeyPrefix()).withKeySerializer(
|
||||
template.getKeySerializer()), value).expireAfter(cacheMetadata.getDefaultExpiration()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Add the element by adding {@link RedisCacheElement#get()} at {@link RedisCacheElement#getKeyBytes()}. If the cache
|
||||
* previously contained a mapping for this {@link RedisCacheElement#getKeyBytes()}, the old value is replaced by
|
||||
* {@link RedisCacheElement#get()}.
|
||||
*
|
||||
* @param element must not be {@literal null}.
|
||||
* @since 1.5
|
||||
*/
|
||||
public void put(RedisCacheElement element) {
|
||||
|
||||
notNull(element, "Element must not be null!");
|
||||
template.execute(new RedisCachePutCallback(element, cacheValueAccessor, cacheMetadata), true);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#putIfAbsent(java.lang.Object, java.lang.Object)
|
||||
*/
|
||||
public ValueWrapper putIfAbsent(Object key, final Object value) {
|
||||
|
||||
final byte[] keyBytes = computeKey(key);
|
||||
final byte[] valueBytes = convertToBytesIfNecessary(template.getValueSerializer(), value);
|
||||
return putIfAbsent(new RedisCacheElement(new RedisCacheKey(key).usePrefix(cacheMetadata.getKeyPrefix())
|
||||
.withKeySerializer(template.getKeySerializer()), value).expireAfter(cacheMetadata.getDefaultExpiration()));
|
||||
}
|
||||
|
||||
return toWrapper(template.execute(new RedisCallback<Object>() {
|
||||
public Object doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
/**
|
||||
* Add the element as long as no element exists at {@link RedisCacheElement#getKeyBytes()}. If a value is present for
|
||||
* {@link RedisCacheElement#getKeyBytes()} this one is returned.
|
||||
*
|
||||
* @param element must not be {@literal null}.
|
||||
* @return
|
||||
* @since 1.5
|
||||
*/
|
||||
public ValueWrapper putIfAbsent(RedisCacheElement element) {
|
||||
|
||||
waitForLock(connection);
|
||||
notNull(element, "Element must not be null!");
|
||||
return toWrapper(template.execute(new RedisCachePutIfAbsentCallback(element, cacheValueAccessor, cacheMetadata),
|
||||
true));
|
||||
}
|
||||
|
||||
Object resultValue = value;
|
||||
boolean valueWasSet = connection.setNX(keyBytes, valueBytes);
|
||||
if (valueWasSet) {
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#evict(java.lang.Object)
|
||||
*/
|
||||
public void evict(Object key) {
|
||||
evict(new RedisCacheElement(new RedisCacheKey(key).usePrefix(cacheMetadata.getKeyPrefix()).withKeySerializer(
|
||||
template.getKeySerializer()), null));
|
||||
}
|
||||
|
||||
if (shouldTrackKeys()) {
|
||||
connection.zAdd(setName, 0, keyBytes);
|
||||
}
|
||||
if (expiration > 0) {
|
||||
/**
|
||||
* @param element {@link RedisCacheElement#getKeyBytes()}
|
||||
* @since 1.5
|
||||
*/
|
||||
public void evict(final RedisCacheElement element) {
|
||||
|
||||
connection.expire(keyBytes, expiration);
|
||||
// update the expiration of the set of keys as well
|
||||
notNull(element, "Element must not be null!");
|
||||
template.execute(new RedisCacheEvictCallback(element, cacheMetadata));
|
||||
}
|
||||
|
||||
if (shouldTrackKeys()) {
|
||||
connection.expire(setName, expiration);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
resultValue = deserializeIfNecessary(template.getValueSerializer(), connection.get(keyBytes));
|
||||
}
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#clear()
|
||||
*/
|
||||
public void clear() {
|
||||
template.execute(cacheMetadata.usesKeyPrefix() ? new RedisCacheCleanByPrefixCallback(cacheMetadata)
|
||||
: new RedisCacheCleanByKeysCallback(cacheMetadata), true);
|
||||
}
|
||||
|
||||
return resultValue;
|
||||
}
|
||||
}, true));
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.cache.Cache#getName()
|
||||
*/
|
||||
public String getName() {
|
||||
return cacheMetadata.getCacheName();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc} This implementation simply returns the RedisTemplate used for configuring the cache, giving access to
|
||||
* the underlying Redis store.
|
||||
*/
|
||||
public Object getNativeCache() {
|
||||
return template;
|
||||
}
|
||||
|
||||
private ValueWrapper toWrapper(Object value) {
|
||||
return (value != null ? new SimpleValueWrapper(value) : null);
|
||||
}
|
||||
|
||||
public void evict(Object key) {
|
||||
final byte[] k = computeKey(key);
|
||||
/**
|
||||
* Metadata required to maintain {@link RedisCache}. Keeps track of additional data structures required for processing
|
||||
* cache operations.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCacheMetadata {
|
||||
|
||||
template.execute(new RedisCallback<Object>() {
|
||||
public Object doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
connection.del(k);
|
||||
// remove key from set
|
||||
if (shouldTrackKeys()) {
|
||||
connection.zRem(setName, k);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}, true);
|
||||
}
|
||||
private final String cacheName;
|
||||
private final byte[] keyPrefix;
|
||||
private final byte[] setOfKnownKeys;
|
||||
private final byte[] cacheLockName;
|
||||
private long defaultExpiration = 0;
|
||||
|
||||
public void clear() {
|
||||
// need to del each key individually
|
||||
template.execute(new RedisCallback<Object>() {
|
||||
public Object doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
// another clear is on-going
|
||||
if (connection.exists(cacheLockName)) {
|
||||
return null;
|
||||
}
|
||||
/**
|
||||
* @param cacheName must not be {@literal null} or empty.
|
||||
* @param keyPrefix can be {@literal null}.
|
||||
*/
|
||||
public RedisCacheMetadata(String cacheName, byte[] keyPrefix) {
|
||||
|
||||
try {
|
||||
connection.set(cacheLockName, cacheLockName);
|
||||
hasText(cacheName, "CacheName must not be null or empty!");
|
||||
this.cacheName = cacheName;
|
||||
this.keyPrefix = keyPrefix;
|
||||
|
||||
int offset = 0;
|
||||
boolean finished = false;
|
||||
StringRedisSerializer stringSerializer = new StringRedisSerializer();
|
||||
|
||||
if (shouldTrackKeys()) {
|
||||
do {
|
||||
// need to paginate the keys
|
||||
Set<byte[]> keys = connection.zRange(setName, (offset) * PAGE_SIZE, (offset + 1) * PAGE_SIZE - 1);
|
||||
finished = keys.size() < PAGE_SIZE;
|
||||
offset++;
|
||||
if (!keys.isEmpty()) {
|
||||
connection.del(keys.toArray(new byte[keys.size()][]));
|
||||
}
|
||||
} while (!finished);
|
||||
|
||||
connection.del(setName);
|
||||
} else {
|
||||
|
||||
byte[] wildCard = "*".getBytes();
|
||||
byte[] prefixToUse = Arrays.copyOf(prefix, prefix.length + wildCard.length);
|
||||
System.arraycopy(wildCard, 0, prefixToUse, prefix.length, wildCard.length);
|
||||
|
||||
connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
|
||||
}
|
||||
return null;
|
||||
} finally {
|
||||
connection.del(cacheLockName);
|
||||
}
|
||||
}
|
||||
}, true);
|
||||
}
|
||||
|
||||
private byte[] computeKey(Object key) {
|
||||
|
||||
byte[] keyBytes = convertToBytesIfNecessary(template.getKeySerializer(), key);
|
||||
|
||||
if (prefix == null || prefix.length == 0) {
|
||||
return keyBytes;
|
||||
// name of the set holding the keys
|
||||
this.setOfKnownKeys = usesKeyPrefix() ? new byte[] {} : stringSerializer.serialize(cacheName + "~keys");
|
||||
this.cacheLockName = stringSerializer.serialize(cacheName + "~lock");
|
||||
}
|
||||
|
||||
byte[] result = Arrays.copyOf(prefix, prefix.length + keyBytes.length);
|
||||
System.arraycopy(keyBytes, 0, result, prefix.length, keyBytes.length);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private boolean waitForLock(RedisConnection connection) {
|
||||
|
||||
boolean retry;
|
||||
boolean foundLock = false;
|
||||
do {
|
||||
retry = false;
|
||||
if (connection.exists(cacheLockName)) {
|
||||
foundLock = true;
|
||||
try {
|
||||
Thread.sleep(WAIT_FOR_LOCK);
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
retry = true;
|
||||
}
|
||||
} while (retry);
|
||||
|
||||
return foundLock;
|
||||
}
|
||||
|
||||
private byte[] convertToBytesIfNecessary(RedisSerializer<Object> serializer, Object value) {
|
||||
|
||||
if (serializer == null && value instanceof byte[]) {
|
||||
return (byte[]) value;
|
||||
/**
|
||||
* @return true if the {@link RedisCache} uses a prefix for key ranges.
|
||||
*/
|
||||
public boolean usesKeyPrefix() {
|
||||
return (keyPrefix != null && keyPrefix.length > 0);
|
||||
}
|
||||
|
||||
return serializer.serialize(value);
|
||||
}
|
||||
|
||||
private Object deserializeIfNecessary(RedisSerializer<byte[]> serializer, byte[] value) {
|
||||
|
||||
if (serializer != null) {
|
||||
return serializer.deserialize(value);
|
||||
/**
|
||||
* Get the binary representation of the key prefix.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
public byte[] getKeyPrefix() {
|
||||
return this.keyPrefix;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the binary representation of the key identifying the data structure used to maintain known keys.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
public byte[] getSetOfKnownKeysKey() {
|
||||
return setOfKnownKeys;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the binary representation of the key identifying the data structure used to lock the cache.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
public byte[] getCacheLockKey() {
|
||||
return cacheLockName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the name of the cache.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public String getCacheName() {
|
||||
return cacheName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the default expiration time in seconds
|
||||
*
|
||||
* @param defaultExpiration
|
||||
*/
|
||||
public void setDefaultExpiration(long seconds) {
|
||||
this.defaultExpiration = seconds;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the default expiration time in seconds.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public long getDefaultExpiration() {
|
||||
return defaultExpiration;
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* A set of keys needs to be maintained in case no prefix is provided. Clear cache depends on either a pattern to
|
||||
* identify keys to remove or a maintained set of known keys, otherwise the entire db would be flushed potentially
|
||||
* removing non cache entry values.
|
||||
*
|
||||
* @return
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
private boolean shouldTrackKeys() {
|
||||
return (prefix == null || prefix.length == 0);
|
||||
static class CacheValueAccessor {
|
||||
|
||||
@SuppressWarnings("rawtypes")//
|
||||
private final RedisSerializer valueSerializer;
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
CacheValueAccessor(RedisSerializer valueRedisSerializer) {
|
||||
valueSerializer = valueRedisSerializer;
|
||||
}
|
||||
|
||||
byte[] convertToBytesIfNecessary(Object value) {
|
||||
|
||||
if (valueSerializer == null && value instanceof byte[]) {
|
||||
return (byte[]) value;
|
||||
}
|
||||
|
||||
return valueSerializer.serialize(value);
|
||||
}
|
||||
|
||||
Object deserializeIfNecessary(byte[] value) {
|
||||
|
||||
if (valueSerializer != null) {
|
||||
return valueSerializer.deserialize(value);
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
* @param <T>
|
||||
*/
|
||||
static abstract class AbstractRedisCacheCallback<T> implements RedisCallback<T> {
|
||||
|
||||
private long WAIT_FOR_LOCK_TIMEOUT = 300;
|
||||
private final RedisCacheElement element;
|
||||
private final RedisCacheMetadata cacheMetadata;
|
||||
|
||||
public AbstractRedisCacheCallback(RedisCacheElement element, RedisCacheMetadata metadata) {
|
||||
this.element = element;
|
||||
this.cacheMetadata = metadata;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.core.RedisCallback#doInRedis(org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public T doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
waitForLock(connection);
|
||||
return doInRedis(element, connection);
|
||||
}
|
||||
|
||||
public abstract T doInRedis(RedisCacheElement element, RedisConnection connection) throws DataAccessException;
|
||||
|
||||
protected void processKeyExpiration(RedisCacheElement element, RedisConnection connection) {
|
||||
if (!element.isEternal()) {
|
||||
connection.expire(element.getKeyBytes(), element.getTimeToLive());
|
||||
}
|
||||
}
|
||||
|
||||
protected void maintainKnownKeys(RedisCacheElement element, RedisConnection connection) {
|
||||
|
||||
if (!element.hasKeyPrefix()) {
|
||||
connection.zAdd(cacheMetadata.getSetOfKnownKeysKey(), 0, element.getKeyBytes());
|
||||
connection.expire(cacheMetadata.getSetOfKnownKeysKey(), element.getTimeToLive());
|
||||
}
|
||||
}
|
||||
|
||||
protected void cleanKnownKeys(RedisCacheElement element, RedisConnection connection) {
|
||||
|
||||
if (!element.hasKeyPrefix()) {
|
||||
connection.zRem(cacheMetadata.getSetOfKnownKeysKey(), element.getKeyBytes());
|
||||
}
|
||||
}
|
||||
|
||||
protected boolean waitForLock(RedisConnection connection) {
|
||||
|
||||
boolean retry;
|
||||
boolean foundLock = false;
|
||||
do {
|
||||
retry = false;
|
||||
if (connection.exists(cacheMetadata.getCacheLockKey())) {
|
||||
foundLock = true;
|
||||
try {
|
||||
Thread.sleep(WAIT_FOR_LOCK_TIMEOUT);
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
retry = true;
|
||||
}
|
||||
} while (retry);
|
||||
|
||||
return foundLock;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @param <T>
|
||||
* @since 1.5
|
||||
*/
|
||||
static abstract class LockingRedisCacheCallback<T> implements RedisCallback<T> {
|
||||
|
||||
private final RedisCacheMetadata metadata;
|
||||
|
||||
public LockingRedisCacheCallback(RedisCacheMetadata metadata) {
|
||||
this.metadata = metadata;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.core.RedisCallback#doInRedis(org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public T doInRedis(RedisConnection connection) throws DataAccessException {
|
||||
|
||||
if (connection.exists(metadata.getCacheLockKey())) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
connection.set(metadata.getCacheLockKey(), metadata.getCacheLockKey());
|
||||
return doInLock(connection);
|
||||
} finally {
|
||||
connection.del(metadata.getCacheLockKey());
|
||||
}
|
||||
}
|
||||
|
||||
public abstract T doInLock(RedisConnection connection);
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCacheCleanByKeysCallback extends LockingRedisCacheCallback<Void> {
|
||||
|
||||
private static final int PAGE_SIZE = 128;
|
||||
private final RedisCacheMetadata metadata;
|
||||
|
||||
RedisCacheCleanByKeysCallback(RedisCacheMetadata metadata) {
|
||||
super(metadata);
|
||||
this.metadata = metadata;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.cache.RedisCache.LockingRedisCacheCallback#doInLock(org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public Void doInLock(RedisConnection connection) {
|
||||
|
||||
int offset = 0;
|
||||
boolean finished = false;
|
||||
|
||||
do {
|
||||
// need to paginate the keys
|
||||
Set<byte[]> keys = connection.zRange(metadata.getSetOfKnownKeysKey(), (offset) * PAGE_SIZE, (offset + 1)
|
||||
* PAGE_SIZE - 1);
|
||||
finished = keys.size() < PAGE_SIZE;
|
||||
offset++;
|
||||
if (!keys.isEmpty()) {
|
||||
connection.del(keys.toArray(new byte[keys.size()][]));
|
||||
}
|
||||
} while (!finished);
|
||||
|
||||
connection.del(metadata.getSetOfKnownKeysKey());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCacheCleanByPrefixCallback extends LockingRedisCacheCallback<Void> {
|
||||
|
||||
private static final byte[] REMOVE_KEYS_BY_PATTERN_LUA = new StringRedisSerializer()
|
||||
.serialize("local keys = redis.call('KEYS', ARGV[1]); local keysCount = table.getn(keys); if(keysCount > 0) then for _, key in ipairs(keys) do redis.call('del', key); end; end; return keysCount;");
|
||||
private static final byte[] WILD_CARD = new StringRedisSerializer().serialize("*");
|
||||
private final RedisCacheMetadata metadata;
|
||||
|
||||
public RedisCacheCleanByPrefixCallback(RedisCacheMetadata metadata) {
|
||||
super(metadata);
|
||||
this.metadata = metadata;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.cache.RedisCache.LockingRedisCacheCallback#doInLock(org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public Void doInLock(RedisConnection connection) throws DataAccessException {
|
||||
|
||||
byte[] prefixToUse = Arrays.copyOf(metadata.getKeyPrefix(), metadata.getKeyPrefix().length + WILD_CARD.length);
|
||||
System.arraycopy(WILD_CARD, 0, prefixToUse, metadata.getKeyPrefix().length, WILD_CARD.length);
|
||||
|
||||
connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
|
||||
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCacheEvictCallback extends AbstractRedisCacheCallback<Void> {
|
||||
|
||||
public RedisCacheEvictCallback(RedisCacheElement element, RedisCacheMetadata metadata) {
|
||||
super(element, metadata);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.cache.RedisCache.AbstractRedisCacheCallback#doInRedis(org.springframework.data.redis.cache.RedisCacheElement, org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public Void doInRedis(RedisCacheElement element, RedisConnection connection) throws DataAccessException {
|
||||
|
||||
connection.del(element.getKeyBytes());
|
||||
cleanKnownKeys(element, connection);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCachePutCallback extends AbstractRedisCacheCallback<Void> {
|
||||
|
||||
private final CacheValueAccessor valueAccessor;
|
||||
|
||||
public RedisCachePutCallback(RedisCacheElement element, CacheValueAccessor valueAccessor,
|
||||
RedisCacheMetadata metadata) {
|
||||
|
||||
super(element, metadata);
|
||||
this.valueAccessor = valueAccessor;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.cache.RedisCache.AbstractRedisPutCallback#doInRedis(org.springframework.data.redis.cache.RedisCache.RedisCacheElement, org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public Void doInRedis(RedisCacheElement element, RedisConnection connection) throws DataAccessException {
|
||||
|
||||
connection.multi();
|
||||
|
||||
connection.set(element.getKeyBytes(), valueAccessor.convertToBytesIfNecessary(element.get()));
|
||||
|
||||
processKeyExpiration(element, connection);
|
||||
maintainKnownKeys(element, connection);
|
||||
|
||||
connection.exec();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
static class RedisCachePutIfAbsentCallback extends AbstractRedisCacheCallback<Object> {
|
||||
|
||||
private final CacheValueAccessor valueAccessor;
|
||||
|
||||
public RedisCachePutIfAbsentCallback(RedisCacheElement element, CacheValueAccessor valueAccessor,
|
||||
RedisCacheMetadata metadata) {
|
||||
|
||||
super(element, metadata);
|
||||
this.valueAccessor = valueAccessor;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.redis.cache.RedisCache.AbstractRedisPutCallback#doInRedis(org.springframework.data.redis.cache.RedisCache.RedisCacheElement, org.springframework.data.redis.connection.RedisConnection)
|
||||
*/
|
||||
@Override
|
||||
public Object doInRedis(RedisCacheElement element, RedisConnection connection) throws DataAccessException {
|
||||
|
||||
waitForLock(connection);
|
||||
Object resultValue = put(element, connection);
|
||||
|
||||
if (nullSafeEquals(element.get(), resultValue)) {
|
||||
processKeyExpiration(element, connection);
|
||||
maintainKnownKeys(element, connection);
|
||||
}
|
||||
|
||||
return resultValue;
|
||||
}
|
||||
|
||||
private Object put(RedisCacheElement element, RedisConnection connection) {
|
||||
|
||||
boolean valueWasSet = connection.setNX(element.getKeyBytes(),
|
||||
valueAccessor.convertToBytesIfNecessary(element.get()));
|
||||
return valueWasSet ? element.get() : valueAccessor.deserializeIfNecessary(connection.get(element.getKeyBytes()));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
101
src/main/java/org/springframework/data/redis/cache/RedisCacheElement.java
vendored
Normal file
101
src/main/java/org/springframework/data/redis/cache/RedisCacheElement.java
vendored
Normal file
@@ -0,0 +1,101 @@
|
||||
/*
|
||||
* Copyright 2015 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.cache;
|
||||
|
||||
import static org.springframework.util.Assert.*;
|
||||
|
||||
import org.springframework.cache.support.SimpleValueWrapper;
|
||||
|
||||
/**
|
||||
* Element to be stored inside {@link RedisCache}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
public class RedisCacheElement extends SimpleValueWrapper {
|
||||
|
||||
private final RedisCacheKey cacheKey;
|
||||
private long timeToLive;
|
||||
|
||||
/**
|
||||
* @param cacheKey the key to be used for storing value in {@link RedisCache}. Must not be {@literal null}.
|
||||
* @param value
|
||||
*/
|
||||
public RedisCacheElement(RedisCacheKey cacheKey, Object value) {
|
||||
super(value);
|
||||
|
||||
notNull(cacheKey, "CacheKey must not be null!");
|
||||
this.cacheKey = cacheKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the binary key representation.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public byte[] getKeyBytes() {
|
||||
return cacheKey.getKeyBytes();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return
|
||||
*/
|
||||
public RedisCacheKey getKey() {
|
||||
return cacheKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the elements time to live. Use {@literal zero} to store eternally.
|
||||
*
|
||||
* @param timeToLive
|
||||
*/
|
||||
public void setTimeToLive(long timeToLive) {
|
||||
this.timeToLive = timeToLive;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return
|
||||
*/
|
||||
public long getTimeToLive() {
|
||||
return timeToLive;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true in case {@link RedisCacheKey} is prefixed.
|
||||
*/
|
||||
public boolean hasKeyPrefix() {
|
||||
return cacheKey.hasPrefix();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true if timeToLive is 0
|
||||
*/
|
||||
public boolean isEternal() {
|
||||
return 0 == timeToLive;
|
||||
}
|
||||
|
||||
/**
|
||||
* Expire the element after given seconds.
|
||||
*
|
||||
* @param seconds
|
||||
* @return
|
||||
*/
|
||||
public RedisCacheElement expireAfter(long seconds) {
|
||||
|
||||
setTimeToLive(seconds);
|
||||
return this;
|
||||
}
|
||||
}
|
||||
116
src/main/java/org/springframework/data/redis/cache/RedisCacheKey.java
vendored
Normal file
116
src/main/java/org/springframework/data/redis/cache/RedisCacheKey.java
vendored
Normal file
@@ -0,0 +1,116 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.data.redis.cache;
|
||||
|
||||
import static org.springframework.util.Assert.*;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 1.5
|
||||
*/
|
||||
public class RedisCacheKey {
|
||||
|
||||
private final Object keyElement;
|
||||
private byte[] prefix;
|
||||
@SuppressWarnings("rawtypes")//
|
||||
private RedisSerializer serializer;
|
||||
|
||||
/**
|
||||
* @param keyElement must not be {@literal null}.
|
||||
*/
|
||||
public RedisCacheKey(Object keyElement) {
|
||||
|
||||
notNull(keyElement, "KeyElement must not be null!");
|
||||
this.keyElement = keyElement;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the {@link Byte} representation of the given key element using prefix if available.
|
||||
*/
|
||||
public byte[] getKeyBytes() {
|
||||
|
||||
byte[] rawKey = serializeKeyElement();
|
||||
if (!hasPrefix()) {
|
||||
return rawKey;
|
||||
}
|
||||
|
||||
byte[] prefixedKey = Arrays.copyOf(prefix, prefix.length + rawKey.length);
|
||||
System.arraycopy(rawKey, 0, prefixedKey, prefix.length, rawKey.length);
|
||||
|
||||
return prefixedKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return
|
||||
*/
|
||||
public Object getKeyElement() {
|
||||
return keyElement;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private byte[] serializeKeyElement() {
|
||||
|
||||
if (serializer == null && keyElement instanceof byte[]) {
|
||||
return (byte[]) keyElement;
|
||||
}
|
||||
|
||||
return serializer.serialize(keyElement);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link RedisSerializer} used for converting the key into its {@link Byte} representation.
|
||||
*
|
||||
* @param serializer can be {@literal null}.
|
||||
*/
|
||||
public void setSerializer(RedisSerializer<?> serializer) {
|
||||
this.serializer = serializer;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true if prefix is not empty.
|
||||
*/
|
||||
public boolean hasPrefix() {
|
||||
return (prefix != null && prefix.length > 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Use the given prefix when generating key.
|
||||
*
|
||||
* @param prefix can be {@literal null}.
|
||||
* @return
|
||||
*/
|
||||
public RedisCacheKey usePrefix(byte[] prefix) {
|
||||
this.prefix = prefix;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Use {@link RedisSerializer} for converting the key into its {@link Byte} representation.
|
||||
*
|
||||
* @param serializer can be {@literal null}.
|
||||
* @return
|
||||
*/
|
||||
public RedisCacheKey withKeySerializer(RedisSerializer serializer) {
|
||||
|
||||
this.serializer = serializer;
|
||||
return this;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user