FeedEntryMessageSource.doReceive()方法阻塞：设置超时仍无效

环境信息 #

• OpenJDK 17
• Spring Integration 5.5.7

问题现象 #

已为UrlResource设置connectTimeout（10秒）和readTimeout（5秒），但线程依然出现阻塞情况，线程快照显示处于锁等待状态：

代码实现 #

private FeedEntryMessageSource feedEntrySource(URL feedURL, String metadataKey) {
    UrlResource urlResource =
        new UrlResource(feedURL) {

          @Override
          protected void customizeConnection(HttpURLConnection connection) throws IOException {
            super.customizeConnection(connection);
            connection.setConnectTimeout(10000);
            connection.setReadTimeout(5000);
          }
        };
    return new FeedEntryMessageSource(urlResource, metadataKey);
  }

线程快照 #

stackTrace:
java.lang.Thread.State: BLOCKED (on object monitor)
at org.springframework.integration.feed.inbound.FeedEntryMessageSource.doReceive(FeedEntryMessageSource.java:172)
- waiting to lock <0x000000043297b4e8> (a java.lang.Object)

问题分析与解决 #

阻塞原因拆解 #

1. 锁竞争导致阻塞：FeedEntryMessageSource.doReceive()是同步方法，同一时刻仅能有一个线程执行，其他线程会进入BLOCKED状态等待锁释放。
2. 超时设置未生效：UrlResource的customizeConnection()方法在部分场景下不会被调用，导致网络请求没有超时限制——持有锁的线程可能因网络长时间挂起无法释放锁，进一步加剧锁竞争。

解决方案 #

1. 消除锁竞争根源

不要让多个线程共享同一个FeedEntryMessageSource实例，为每个消费线程或任务创建独立实例，从根本上避免锁等待。

2. 确保超时设置生效

替换UrlResource的自定义逻辑，改用SimpleClientHttpRequestFactory明确配置网络超时，确保参数被正确应用：

private FeedEntryMessageSource feedEntrySource(URL feedURL, String metadataKey) {
    SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    requestFactory.setConnectTimeout(10000);
    requestFactory.setReadTimeout(5000);

    try (InputStream inputStream = requestFactory.createRequest(feedURL, HttpMethod.GET).getBody()) {
        byte[] content = StreamUtils.copyToByteArray(inputStream);
        Resource resource = new ByteArrayResource(content);
        return new FeedEntryMessageSource(resource, metadataKey);
    } catch (IOException e) {
        throw new RuntimeException("Failed to fetch feed resource", e);
    }
}

3. 增加线程超时管控

若必须共享实例，可在调用receive()方法时配合线程池的超时机制（如ExecutorService.invokeAll()设置超时时间），避免线程无限期等待锁释放。

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