kotlin - 使用 Redis Streams 和 Spring Data 获取待处理消息
问题描述
我在我的 Spring Boot 应用程序中使用 Redis Streams。在调度程序中,我经常想要获取所有待处理的消息并检查它们已经处理了多长时间,并在必要时重新触发它们。
我现在的问题是我可以获取待处理的消息,但我不确定如何获取有效负载。
我的第一种方法使用了pending
andrange
操作。这里的缺点是 totalDeliveryCount 没有增加range
- 所以我不能使用 range 方法
val pendingMessages = stringRedisTemplate.opsForStream<String, Any>().pending(redisStreamName, Consumer.from(redisConsumerGroup, instanceName))
return pendingMessages.filter { pendingMessage ->
if (pendingMessage.totalDeliveryCount < maxDeliveryAttempts && pendingMessage.elapsedTimeSinceLastDelivery > Duration.ofMillis(pendingTimeout.toLong())) {
return@filter true
} else {
...
return@filter false
}
}.map { //map from PendingMessage::class to a MapRecord with the content
val map = stringRedisTemplate.opsForStream().range(redisStreamName, Range.just(it.idAsString)) // does not increase totalDeliveryCount !!!
if (map != null && map.size > 0) {
return@map map[0]
} else {
return@map null
}
}.filterNotNull().toList()
我的第二种方法使用了pending
andread
操作。对于读取操作,我可以使用当前 ID 指定偏移量。问题是我只能得到高于指定 ID 的 ID。
val pendingMessages = stringRedisTemplate.opsForStream().pending(redisStreamName, Consumer.from(redisConsumerGroup, instanceName))
return pendingMessages.filter { pendingMessage ->
if (pendingMessage.totalDeliveryCount < maxDeliveryAttempts && pendingMessage.elapsedTimeSinceLastDelivery > Duration.ofMillis(pendingTimeout.toLong())) {
return@filter true
} else {
...
return@filter false
}
}.map { //map from PendingMessage::class to a MapRecord with the content
val map = stringRedisTemplate.opsForStream<String, Any>()
.read(it.consumer, StreamReadOptions.empty().count(1),
StreamOffset.create(redisStreamName, ReadOffset.from(it.id)))
if (map != null && map.size > 0 && map[0].id.value == it.idAsString) { // map[0].id.value == it.idAsString does not match
return@map map[0]
} else {
return@map null
}
}.filterNotNull().toList()
因此,当我使用ReadOffset.from('1234-0')
该消息时,我不会收到消息,1234-0
而是收到消息之后的所有内容。有没有办法获得确切的信息并尊重totalDeliveryCount
和elapsedTimeSinceLastDelivery
统计?
我正在使用 spring-data-redis 2.3.1.RELEASE
解决方案
我现在正在使用以下解决方法,这对于大多数情况应该是好的:
return if (id.sequence > 0) {
"${id.timestamp}-${id.sequence - 1}"
} else {
"${id.timestamp - 1}-99999"
}
它依赖于这样一个事实,即每毫秒插入的消息不超过 99999 条。
推荐阅读
- python - Selenium 无法点击网页 python 的按钮
- excel - IsEmpty 对于空范围和单元格内容为空白的范围均返回 True,如何区分?
- firebase - 如何使用 Firebase Firestore 进行一次性简单查询?
- reactjs - React 状态依赖于其他状态
- google-chrome-extension - 是否可以将自定义身份验证库注入 chrome 扩展后台脚本?
- r - 在 R 中循环以从数据帧中读取数据
- python - Python 多处理共享变量
- android - Android使用adb跳过chrome欢迎屏幕
- python - 如何将函数应用于熊猫中每一行的对象属性
- c# - 计算位置的唯一组合