使用Spring Data Redis与Kotlin协程实现Redis Pub/Sub时订阅者收不到消息

问题原因 #

你遇到的订阅失败是两个核心问题共同导致的：

1. Redis序列化配置缺失Pub/Sub专属规则
   Spring Data Redis的RedisSerializationContext中，普通KV操作和Pub/Sub操作的序列化器是分开配置的，你当前只配置了KV、Hash相关的序列化器，Pub/Sub的消息和Channel序列化用的是默认值：

• Channel默认序列化器不是StringRedisSerializer，可能出现通道名匹配异常
• Pub/Sub消息默认序列化器是JDK序列化器，你发布时用Jackson序列化的Uuid对象，订阅时用JDK反序列化直接失败，异常被Flow吞掉导致没有报错日志，直接返回null

2. 测试中订阅就绪等待逻辑不可靠
   yield()只会让出当前协程的CPU时间片，不能保证订阅协程已经完成和Redis的SUBSCRIBE命令交互，大概率出现你还没订阅完成就已经发布消息，自然收不到。

解决方案 #

1. 修正Redis序列化配置 #

补全RedisSerializationContext的Pub/Sub序列化规则：

@Bean
fun reactiveRedisTemplate(
    reactiveRedisConnectionFactory: ReactiveRedisConnectionFactory,
    objectMapper: ObjectMapper
): ReactiveRedisTemplate<String, Any> {
    val stringRedisSerializer = StringRedisSerializer()
    val jacksonSerializer = GenericJackson2JsonRedisSerializer(objectMapper)
    val serializationContext =
        RedisSerializationContext.newSerializationContext<String, Any>()
            .key(stringRedisSerializer)
            .value(jacksonSerializer)
            .hashKey(stringRedisSerializer)
            .hashValue(jacksonSerializer)
            .channel(stringRedisSerializer) // 新增Channel序列化规则
            .message(jacksonSerializer) // 新增Pub/Sub消息序列化规则
            .build()
    return ReactiveRedisTemplate(reactiveRedisConnectionFactory, serializationContext)
}

2. 优化订阅就绪等待逻辑 #

用CountDownLatch确保订阅完成后再发布消息，同时可以加异常捕获方便排查：

// UuidRepository中简化订阅逻辑，无需二次反序列化
suspend fun subscribe(): Flow<Uuid> =
    redisTemplate.listenToChannelAsFlow(UUID_CHANNEL)
        .map { it.message as Uuid }
        .catch { log.error("订阅消息异常", it) } // 按需添加异常日志

// 测试用例修改
@Test
fun testPubSub() {
    val uuid = UUID.randomUUID().toString()
    val id = UUID.randomUUID().toString()
    // 新增订阅就绪计数器
    val subscribeReadyLatch = CountDownLatch(1)
    val receiveLatch = CountDownLatch(1)
    var uuid1: Uuid? = null
    runBlocking {
        launch(Dispatchers.IO) {
            log.info("Going to subscribe")
            uuid1 = repo.subscribe()
                .onStart { subscribeReadyLatch.countDown() } // 订阅启动后标记就绪
                .firstOrNull()
            receiveLatch.countDown()
        }
        // 等待订阅就绪后再发消息
        assertThat(subscribeReadyLatch.await(1, TimeUnit.SECONDS)).isTrue
        log.info("Going to publish")
        repo.publish(Uuid(id, uuid))
        assertThat(receiveLatch.await(2L, TimeUnit.SECONDS)).isTrue
        assertThat(uuid1).isNotNull
        assertThat(uuid1!!.userId).isEqualTo(id)
        assertThat(uuid1!!.uuid).isEqualTo(uuid)
    }
}

修改完成后即可正常接收订阅消息。

内容的提问来源于stack exchange，提问作者Abhijit Sarkar