如何使用Spring Flux实现数据库变更时主动推送候选人点赞数更新

核心思路 #

• 用Sinks.Many作为事件发射器，按候选人ID分组维护独立的发射器实例，仅在点赞更新时主动推送最新数据
• 订阅时直接返回发射器对应的Flux，无需定时轮询数据库
• 点赞更新操作执行完成后，主动向对应ID的发射器发送最新的候选人数据

完整代码实现 #

首先改造CandidateService：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;

public class CandidateService {
    // 按候选人ID存储对应的多播Sink，保证多线程安全
    private final Map<String, Sinks.Many<Candidate>> candidateSinkMap = new ConcurrentHashMap<>();

    public Candidate getCandidateDetail(String id) {
        // 原有数据库查询逻辑保持不变
        // select * from candidates where id = :id
        // and return it
    }

    public void updateLikes(String id, int likesCount) {
        // 原有更新数据库逻辑保持不变
        // update candidates set likes_count = :likesCount where id = :id
        // ...
        
        // 新增逻辑：更新完成后推送最新数据
        Candidate latestCandidate = getCandidateDetail(id);
        // 仅当有客户端订阅该候选人时才执行推送
        Sinks.Many<Candidate> sink = candidateSinkMap.get(id);
        if (sink != null) {
            // 用onBackpressureLatest处理背压，只保留最新的点赞数据，丢弃过期数据
            sink.emitNext(latestCandidate, Sinks.EmitFailureHandler.FAIL_FAST);
        }
    }

    // 订阅候选人变更的方法，替换原有轮询实现
    public Flux<Candidate> subscribeItemChange(String id) {
        // 不存在则自动创建对应ID的Sink
        Sinks.Many<Candidate> sink = candidateSinkMap.computeIfAbsent(id, 
            k -> Sinks.many().multicast().onBackpressureLatest());
        
        // 转成Flux返回，订阅取消时自动清理无订阅的Sink避免内存泄漏
        return sink.asFlux()
            .doFinally(signalType -> {
                if (sink.currentSubscriberCount() == 0) {
                    candidateSinkMap.remove(id);
                }
            });
    }
}

注意事项 #

• 上述方案适用于单机部署场景，如果是多节点集群部署，需要引入消息中间件（如Kafka、RocketMQ）做跨节点的变更事件广播：点赞更新后先发送事件到消息队列，所有服务节点消费到事件后再向自己本地的Sink推送数据
• 如果需要支持订阅时立即返回一次当前最新的点赞数，可以在subscribeItemChange返回Flux的时候拼接startWith(getCandidateDetail(id))即可
• 可以根据业务需要添加Sink的超时清理逻辑，避免长时间无订阅的Sink残留

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