基于Spring REST实现带超时机制的长轮询功能需求咨询

嘿，这个需求用长轮询完全能搞定！结合Spring生态的话，实现起来其实挺顺畅的——我之前做过类似的实时数据推送场景，分享下具体的思路和代码实现：

核心思路 #

长轮询的核心逻辑就是：客户端发起GET请求后，服务器不立即返回响应，而是挂起这个请求，直到两个触发条件之一被满足：

1. 数据库发生了目标变更，服务器立即拉取最新数据返回；
2. 达到预设的最大超时时间，服务器返回当前数据（或超时提示）。

Spring的DeferredResult是处理这种异步挂起请求的绝佳工具，天然支持超时控制，搭配数据库变更监听就能完美实现需求。

具体实现步骤 #

1. 用DeferredResult处理异步请求与超时 #

首先，我们可以用DeferredResult来创建支持超时的异步接口：

@RestController
@RequestMapping("/api")
public class LongPollingController {
    @Autowired
    private DataChangeListener dataChangeListener;

    @GetMapping("/latest-data")
    public DeferredResult<ResponseEntity<Object>> getLatestData(
            // 允许客户端自定义超时，默认20秒
            @RequestParam(defaultValue = "20000") long timeoutMs) {
        
        // 初始化DeferredResult，设置超时时间
        DeferredResult<ResponseEntity<Object>> deferredResult = new DeferredResult<>(timeoutMs);

        // 超时回调：超时后返回当前数据
        deferredResult.onTimeout(() -> {
            Object currentData = fetchCurrentData();
            deferredResult.setResult(ResponseEntity.ok(currentData));
        });

        // 将请求绑定到数据库变更监听器
        dataChangeListener.bindPendingRequest(deferredResult);

        return deferredResult;
    }

    // 模拟获取当前数据的方法
    private Object fetchCurrentData() {
        return dataRepository.findLatest();
    }
}

2. 监听数据库变更事件 #

这里分两种场景给出方案：

场景一：单体应用 + Spring JPA

如果是单体应用且用Spring JPA，可以利用Spring的事件机制监听实体变更：

首先，定义一个自定义事件类，用来传递变更信息：

public class DataChangedEvent extends ApplicationEvent {
    public DataChangedEvent(Object source) {
        super(source);
    }
}

然后在数据更新的服务类中，发布变更事件：

@Service
public class DataService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    @Autowired
    private DataRepository dataRepository;

    public void updateData(Data data) {
        // 执行数据库更新操作
        dataRepository.save(data);
        // 发布数据变更事件
        eventPublisher.publishEvent(new DataChangedEvent(data));
    }
}

最后，编写一个监听器，维护所有挂起的请求，收到变更事件后立即返回最新数据：

@Component
public class DataChangeListener {
    // 用线程安全的CopyOnWriteArrayList存储挂起的请求
    private final CopyOnWriteArrayList<DeferredResult<ResponseEntity<Object>>> pendingRequests 
            = new CopyOnWriteArrayList<>();

    public void bindPendingRequest(DeferredResult<ResponseEntity<Object>> deferredResult) {
        pendingRequests.add(deferredResult);
        // 请求完成（超时/正常返回）后从集合移除，避免内存泄漏
        deferredResult.onCompletion(() -> pendingRequests.remove(deferredResult));
    }

    // 监听数据变更事件
    @TransactionalEventListener
    public void handleDataChanged(DataChangedEvent event) {
        Object latestData = fetchLatestData();
        // 遍历所有挂起的请求，返回最新数据
        for (DeferredResult<ResponseEntity<Object>> result : pendingRequests) {
            result.setResult(ResponseEntity.ok(latestData));
        }
    }

    private Object fetchLatestData() {
        return dataRepository.findLatest();
    }
}

场景二：分布式应用

如果是多实例部署的分布式应用，单实例的内存集合无法共享请求状态，这时候可以用Redis Pub/Sub或者**消息队列（如RabbitMQ）**来实现跨实例的变更通知：

以Redis Pub/Sub为例，先编写一个Redis订阅监听器：

@Component
public class RedisDataChangeListener {
    @Autowired
    private StringRedisTemplate redisTemplate;
    private final CopyOnWriteArrayList<DeferredResult<ResponseEntity<Object>>> pendingRequests 
            = new CopyOnWriteArrayList<>();

    @PostConstruct
    public void subscribeToDataChanges() {
        redisTemplate.execute((RedisConnection connection) -> {
            // 订阅"data-changed"频道
            connection.subscribe((message, pattern) -> {
                // 收到变更消息，返回最新数据
                Object latestData = fetchLatestData();
                for (DeferredResult<ResponseEntity<Object>> result : pendingRequests) {
                    result.setResult(ResponseEntity.ok(latestData));
                }
            }, "data-changed".getBytes());
            return null;
        });
    }

    public void bindPendingRequest(DeferredResult<ResponseEntity<Object>> deferredResult) {
        pendingRequests.add(deferredResult);
        deferredResult.onCompletion(() -> pendingRequests.remove(deferredResult));
    }

    private Object fetchLatestData() {
        return dataRepository.findLatest();
    }
}

然后在数据更新时，往Redis频道发送消息：

@Service
public class DataService {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private DataRepository dataRepository;

    public void updateData(Data data) {
        dataRepository.save(data);
        // 往Redis频道发送变更通知
        redisTemplate.convertAndSend("data-changed", "data-updated");
    }
}

关键注意事项 #

• 线程安全：必须用线程安全的集合（如CopyOnWriteArrayList）存储挂起的请求，避免并发修改导致的异常。
• 内存泄漏防护：一定要在DeferredResult的onCompletion回调中移除请求，防止无用请求长期占用内存。
• 超时灵活性：允许客户端通过请求参数自定义超时时间，适配不同场景的需求。
• 分布式兼容：多实例部署时，务必用分布式消息中间件来统一处理变更通知，不能依赖单实例的内存状态。

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