Skip to content

Commit c5f6bc0

Browse files
KilemonnKilemonn
committed
For cache backed subqueues to improve performance and to allow us to add support for memcached, add a reserved set in the cache that holds all available keys in the subqueue and use it as the source of truth when managing keys.
1 parent 7836e3e commit c5f6bc0

12 files changed

Lines changed: 291 additions & 65 deletions

File tree

‎src/main/kotlin/au/kilemon/messagequeue/authentication/authenticator/cache/redis/RedisAuthenticator.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ import java.util.stream.Collectors
1313
*
1414
* @author github.com/Kilemonn
1515
*/
16-
class RedisAuthenticator: MultiQueueAuthenticator()
16+
class RedisAuthenticator(private val prefix: String): MultiQueueAuthenticator()
1717
{
1818
companion object
1919
{
@@ -33,7 +33,7 @@ class RedisAuthenticator: MultiQueueAuthenticator()
3333
{
3434
if (!isInNoneMode())
3535
{
36-
return setOf(RESTRICTED_KEY)
36+
return setOf("$prefix$RESTRICTED_KEY")
3737
}
3838
return setOf()
3939
}

‎src/main/kotlin/au/kilemon/messagequeue/configuration/QueueConfiguration.kt‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import au.kilemon.messagequeue.logging.HasLogger
1212
import au.kilemon.messagequeue.logging.Messages
1313
import au.kilemon.messagequeue.message.QueueMessage
1414
import au.kilemon.messagequeue.queue.MultiQueue
15+
import au.kilemon.messagequeue.queue.cache.CacheKeyManager
1516
import au.kilemon.messagequeue.queue.cache.redis.RedisMultiQueue
1617
import au.kilemon.messagequeue.queue.inmemory.InMemoryMultiQueue
1718
import au.kilemon.messagequeue.queue.nosql.mongo.MongoMultiQueue
@@ -45,10 +46,6 @@ class QueueConfiguration : HasLogger
4546
@Autowired
4647
private lateinit var messageSource: ReloadableResourceBundleMessageSource
4748

48-
@Autowired
49-
@Lazy
50-
private lateinit var redisTemplate: RedisTemplate<String, QueueMessage>
51-
5249
/**
5350
* Initialise the [MultiQueue] [Bean] based on the [MessageQueueSettings.storageMedium].
5451
*/
@@ -61,7 +58,7 @@ class QueueConfiguration : HasLogger
6158
var queue: MultiQueue = InMemoryMultiQueue()
6259
when (messageQueueSettings.storageMedium.uppercase()) {
6360
StorageMedium.REDIS.toString() -> {
64-
queue = RedisMultiQueue(messageQueueSettings.redisPrefix, redisTemplate)
61+
queue = RedisMultiQueue(messageQueueSettings.redisPrefix)
6562
}
6663
StorageMedium.SQL.toString() -> {
6764
queue = SqlMultiQueue()
@@ -118,7 +115,7 @@ class QueueConfiguration : HasLogger
118115
var authenticator: MultiQueueAuthenticator = InMemoryAuthenticator()
119116
when (messageQueueSettings.storageMedium.uppercase()) {
120117
StorageMedium.REDIS.toString() -> {
121-
authenticator = RedisAuthenticator()
118+
authenticator = RedisAuthenticator(messageQueueSettings.redisPrefix)
122119
}
123120
StorageMedium.SQL.toString() -> {
124121
authenticator = SqlAuthenticator()

‎src/main/kotlin/au/kilemon/messagequeue/configuration/cache/redis/RedisConfiguration.kt‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import au.kilemon.messagequeue.authentication.AuthenticationMatrix
44
import au.kilemon.messagequeue.authentication.RestrictionMode
55
import au.kilemon.messagequeue.logging.HasLogger
66
import au.kilemon.messagequeue.message.QueueMessage
7+
import au.kilemon.messagequeue.queue.cache.redis.RedisCacheKeyManager
78
import au.kilemon.messagequeue.settings.MessageQueueSettings
89
import io.lettuce.core.RedisURI
910
import org.slf4j.Logger
@@ -257,4 +258,21 @@ class RedisConfiguration: HasLogger
257258
template.keySerializer = StringRedisSerializer()
258259
return template
259260
}
261+
262+
@Bean
263+
@ConditionalOnProperty(name=[MessageQueueSettings.STORAGE_MEDIUM], havingValue="REDIS")
264+
fun getRedisCacheKeyManagerRedisTemplate(): RedisTemplate<String, String>
265+
{
266+
val template = RedisTemplate<String, String>()
267+
template.connectionFactory = getConnectionFactory()
268+
template.keySerializer = StringRedisSerializer()
269+
return template
270+
}
271+
272+
@Bean
273+
@ConditionalOnProperty(name=[MessageQueueSettings.STORAGE_MEDIUM], havingValue="REDIS")
274+
fun getRedisCacheKeyManager(): RedisCacheKeyManager
275+
{
276+
return RedisCacheKeyManager()
277+
}
260278
}

‎src/main/kotlin/au/kilemon/messagequeue/queue/MultiQueue.kt‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -356,11 +356,7 @@ abstract class MultiQueue: Queue<QueueMessage>, HasLogger
356356
*/
357357
fun keys(includeEmpty: Boolean = true): Set<String>
358358
{
359-
val keysSet = keysInternal(includeEmpty)
360-
361-
// Remove the reserved key(s)
362-
multiQueueAuthenticator.getReservedSubQueues().forEach { reservedSubQueue -> keysSet.remove(reservedSubQueue) }
363-
return keysSet
359+
return keysInternal(includeEmpty)
364360
}
365361

366362
/**

‎src/main/kotlin/au/kilemon/messagequeue/queue/cache/CacheKeyManager.kt‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
package au.kilemon.messagequeue.queue.cache
2+
3+
/**
4+
* To optimise how we determine the key list/sub queue list when using a cache, this class is used to store and
5+
* manage the subqueue list for the cache backed [au.kilemon.messagequeue.queue.MultiQueue].
6+
*
7+
* @author github.com/Kilemonn
8+
*/
9+
abstract class CacheKeyManager(protected val prefix: String = "")
10+
{
11+
companion object
12+
{
13+
const val CACHE_KEYS_KEY: String = "messagequeue-cache-keys"
14+
}
15+
16+
fun getReservedKeys(): Set<String>
17+
{
18+
return setOf("$prefix$CACHE_KEYS_KEY")
19+
}
20+
21+
abstract fun add(key: String)
22+
23+
abstract fun remove(key: String)
24+
25+
abstract fun getKeys(): HashSet<String>
26+
27+
/**
28+
* Used for tests.
29+
*/
30+
abstract fun contains(key: String): Boolean
31+
32+
/**
33+
* Used for tests.
34+
*/
35+
abstract fun clear()
36+
}

‎src/main/kotlin/au/kilemon/messagequeue/queue/cache/CacheMultiQueue.kt‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
package au.kilemon.messagequeue.queue.cache
2+
3+
import java.util.stream.Collectors
4+
5+
/**
6+
* A marker interface for Cache backed [au.kilemon.messagequeue.queue.MultiQueue].
7+
*
8+
* @author github.com/Kilemonn
9+
*/
10+
interface CacheMultiQueue
11+
{
12+
/**
13+
* Append the [getPrefix] to the provided [subQueue] [String].
14+
*
15+
* @param subQueue the [String] to add the prefix to
16+
* @return a [String] with the provided [subQueue] with the [getPrefix] appended to the beginning.
17+
*/
18+
fun appendPrefix(subQueue: String): String
19+
{
20+
if (hasPrefix() && !subQueue.startsWith(getPrefix()))
21+
{
22+
return "${getPrefix()}$subQueue"
23+
}
24+
return subQueue
25+
}
26+
27+
/**
28+
* @return whether the [getPrefix] is [String.isNotBlank]
29+
*/
30+
fun hasPrefix(): Boolean
31+
{
32+
return getPrefix().isNotBlank()
33+
}
34+
35+
/**
36+
* If [getPrefix] is set, removes this from all provided [keys].
37+
* If [getPrefix] is null or blank, then the provided [keys] [Set] is immediately returned.
38+
*
39+
* @param keys the [Set] of [String] to remove the [getPrefix] from
40+
* @return the updated [Set] of [String] with the [getPrefix] removed
41+
*/
42+
fun removePrefix(keys: Set<String>): Set<String>
43+
{
44+
if (!hasPrefix())
45+
{
46+
return keys
47+
}
48+
49+
val prefixLength = getPrefix().length
50+
return keys.stream().filter { key -> key.startsWith(getPrefix()) }
51+
.map { key -> key.substring(prefixLength) }
52+
.collect(Collectors.toSet())
53+
}
54+
55+
fun getPrefix(): String
56+
}

‎src/main/kotlin/au/kilemon/messagequeue/queue/cache/redis/RedisCacheKeyManager.kt‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package au.kilemon.messagequeue.queue.cache.redis
2+
3+
import au.kilemon.messagequeue.queue.cache.CacheKeyManager
4+
import org.springframework.beans.factory.annotation.Autowired
5+
import org.springframework.data.redis.core.RedisTemplate
6+
7+
/**
8+
* To optimise how we determine the key list/sub queue list when using a cache, this class is used to store and
9+
* manage the subqueue list for the [RedisMultiQueue].
10+
*
11+
* @author github.com/Kilemonn
12+
*/
13+
class RedisCacheKeyManager: CacheKeyManager()
14+
{
15+
@Autowired
16+
private lateinit var redisTemplate: RedisTemplate<String, String>
17+
18+
override fun add(key: String)
19+
{
20+
redisTemplate.opsForSet().add(CACHE_KEYS_KEY, key)
21+
}
22+
23+
override fun remove(key: String)
24+
{
25+
redisTemplate.opsForSet().remove(CACHE_KEYS_KEY, key)
26+
}
27+
28+
override fun contains(key: String): Boolean
29+
{
30+
return redisTemplate.opsForSet().isMember(CACHE_KEYS_KEY, key)
31+
}
32+
33+
override fun getKeys(): HashSet<String>
34+
{
35+
return HashSet(redisTemplate.opsForSet().members(CACHE_KEYS_KEY))
36+
}
37+
38+
override fun clear()
39+
{
40+
redisTemplate.delete(CACHE_KEYS_KEY)
41+
}
42+
}

‎src/main/kotlin/au/kilemon/messagequeue/queue/cache/redis/RedisMultiQueue.kt‎

Lines changed: 20 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -3,17 +3,20 @@ package au.kilemon.messagequeue.queue.cache.redis
33
import au.kilemon.messagequeue.logging.HasLogger
44
import au.kilemon.messagequeue.message.QueueMessage
55
import au.kilemon.messagequeue.queue.MultiQueue
6+
import au.kilemon.messagequeue.queue.cache.CacheMultiQueue
7+
import au.kilemon.messagequeue.queue.exception.IllegalSubQueueIdentifierException
68
import au.kilemon.messagequeue.queue.exception.MessageUpdateException
79
import au.kilemon.messagequeue.settings.MessageQueueSettings
810
import org.slf4j.Logger
11+
import org.springframework.beans.factory.annotation.Autowired
912
import org.springframework.data.redis.core.RedisTemplate
10-
import org.springframework.data.redis.core.ScanOptions
1113
import java.util.Optional
1214
import java.util.Queue
1315

1416
import java.util.concurrent.ConcurrentLinkedQueue
1517
import java.util.stream.Collectors
1618
import kotlin.collections.HashSet
19+
import kotlin.jvm.Throws
1720

1821
/**
1922
* A `Redis` specific implementation of the [MultiQueue].
@@ -22,59 +25,24 @@ import kotlin.collections.HashSet
2225
*
2326
* @author github.com/Kilemonn
2427
*/
25-
class RedisMultiQueue(private val prefix: String = "", private val redisTemplate: RedisTemplate<String, QueueMessage>) : MultiQueue(), HasLogger
28+
class RedisMultiQueue(private val prefix: String) : MultiQueue(), HasLogger, CacheMultiQueue
2629
{
2730
override val LOG: Logger = this.initialiseLogger()
2831

29-
/**
30-
* Append the [MessageQueueSettings.redisPrefix] to the provided [subQueue] [String].
31-
*
32-
* @param subQueue the [String] to add the prefix to
33-
* @return a [String] with the provided [subQueue] with the [MessageQueueSettings.redisPrefix] appended to the beginning.
34-
*/
35-
private fun appendPrefix(subQueue: String): String
36-
{
37-
if (hasPrefix() && !subQueue.startsWith(getPrefix()))
38-
{
39-
return "${getPrefix()}$subQueue"
40-
}
41-
return subQueue
42-
}
32+
@Autowired
33+
private lateinit var redisTemplate: RedisTemplate<String, QueueMessage>
4334

44-
/**
45-
* @return whether the [prefix] is [String.isNotBlank]
46-
*/
47-
internal fun hasPrefix(): Boolean
48-
{
49-
return getPrefix().isNotBlank()
50-
}
35+
@Autowired
36+
private lateinit var cacheKeyManager: RedisCacheKeyManager
5137

5238
/**
5339
* @return [prefix]
5440
*/
55-
internal fun getPrefix(): String
41+
override fun getPrefix(): String
5642
{
5743
return prefix
5844
}
5945

60-
/**
61-
* If [prefix] is set, removes this from all provided [keys].
62-
* If [prefix] is null or blank, then the provided [keys] [Set] is immediately returned.
63-
*
64-
* @param keys the [Set] of [String] to remove the [prefix] from
65-
* @return the updated [Set] of [String] with the [prefix] removed
66-
*/
67-
fun removePrefix(keys: Set<String>): Set<String>
68-
{
69-
if (!hasPrefix())
70-
{
71-
return keys
72-
}
73-
74-
val prefixLength = getPrefix().length
75-
return keys.stream().filter { key -> key.startsWith(getPrefix()) }.map { key -> key.substring(prefixLength) }.collect(Collectors.toSet())
76-
}
77-
7846
/**
7947
* Attempts to append the prefix before requesting the underlying redis entry if the provided [subQueue] is not prefixed with [MessageQueueSettings.redisPrefix].
8048
*/
@@ -125,8 +93,16 @@ class RedisMultiQueue(private val prefix: String = "", private val redisTemplate
12593
return Optional.empty()
12694
}
12795

96+
@Throws(IllegalSubQueueIdentifierException::class)
12897
override fun addInternal(element: QueueMessage): Boolean
12998
{
99+
if (cacheKeyManager.getReservedKeys().contains(element.subQueue)
100+
|| cacheKeyManager.getReservedKeys().contains(appendPrefix(element.subQueue)))
101+
{
102+
throw IllegalSubQueueIdentifierException(element.subQueue)
103+
}
104+
105+
cacheKeyManager.add(appendPrefix(element.subQueue))
130106
val result = redisTemplate.opsForSet().add(appendPrefix(element.subQueue), element)
131107
return result != null && result > 0
132108
}
@@ -151,6 +127,7 @@ class RedisMultiQueue(private val prefix: String = "", private val redisTemplate
151127
{
152128
LOG.debug("Attempting to clear non-existent sub-queue [{}]. No messages cleared.", subQueue)
153129
}
130+
cacheKeyManager.remove(appendPrefix(subQueue))
154131
return amountRemoved
155132
}
156133

@@ -171,10 +148,7 @@ class RedisMultiQueue(private val prefix: String = "", private val redisTemplate
171148

172149
override fun keysInternal(includeEmpty: Boolean): HashSet<String>
173150
{
174-
val scanOptions = ScanOptions.scanOptions().match(appendPrefix("*")).build()
175-
val cursor = redisTemplate.scan(scanOptions)
176-
val keys = HashSet<String>()
177-
cursor.forEach { element -> keys.add(element) }
151+
val keys = cacheKeyManager.getKeys()
178152
if (includeEmpty)
179153
{
180154
LOG.debug("Including all empty queue keys in call to keys(). Total queue keys [{}].", keys.size)

0 commit comments

Comments
 (0)