本文目录导读：
我将为您提供一个完整的Java长轮询实现案例,包括服务端和客户端。
基础概念
长轮询（Long Polling） 是Web应用中常用的实时通信方案：
- 客户端发起请求
- 服务端保持连接,直到有新数据或超时
- 数据到达后立即返回响应
- 客户端收到响应后立即发起新的请求
服务端实现
1 使用Spring MVC实现
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@RestController
public class LongPollingController {
// 存储等待响应的请求
private final Map<String, DeferredResult<String>> pendingRequests = new ConcurrentHashMap<>();
// 定时任务线程池
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
/**
* 长轮询接口
*/
@GetMapping("/api/long-polling")
public DeferredResult<String> longPolling(
@RequestParam String requestId,
@RequestParam(defaultValue = "30000") long timeout) {
// 创建DeferredResult，设置超时时间（毫秒）
DeferredResult<String> deferredResult = new DeferredResult<>(timeout);
// 存储请求
pendingRequests.put(requestId, deferredResult);
// 设置超时回调
deferredResult.onTimeout(() -> {
pendingRequests.remove(requestId);
deferredResult.setResult("timeout");
});
// 设置完成回调
deferredResult.onCompletion(() -> {
pendingRequests.remove(requestId);
});
System.out.println("收到请求: " + requestId + ", 当前等待数: " + pendingRequests.size());
return deferredResult;
}
/**
* 触发数据更新接口
*/
@GetMapping("/api/trigger")
public String trigger(@RequestParam String requestId, @RequestParam String message) {
DeferredResult<String> deferredResult = pendingRequests.remove(requestId);
if (deferredResult != null) {
deferredResult.setResult("消息: " + message);
return "已发送响应给: " + requestId;
}
return "未找到对应的请求: " + requestId;
}
/**
* 批量触发接口（用于测试）
*/
@GetMapping("/api/trigger-all")
public String triggerAll(@RequestParam String message) {
int count = 0;
for (Map.Entry<String, DeferredResult<String>> entry : pendingRequests.entrySet()) {
entry.getValue().setResult("批量消息: " + message);
count++;
}
return "已发送响应给 " + count + " 个请求";
}
}
2 更高级的服务端实现
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
@Service
public class LongPollingService {
// 使用CompletableFuture实现异步
private final ConcurrentHashMap<String, CompletableFuture<String>> requests = new ConcurrentHashMap<>();
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
@PostConstruct
public void init() {
// 启动后台线程模拟数据推送
simulateDataPush();
}
@PreDestroy
public void destroy() {
scheduler.shutdown();
}
/**
* 等待消息更新
*/
public CompletableFuture<String> waitForUpdate(String requestId, long timeout) {
CompletableFuture<String> future = new CompletableFuture<>();
requests.put(requestId, future);
// 设置超时
scheduler.schedule(() -> {
future.complete("timeout");
requests.remove(requestId);
}, timeout, TimeUnit.MILLISECONDS);
return future;
}
/**
* 推送消息给指定请求
*/
public void pushMessage(String requestId, String message) {
CompletableFuture<String> future = requests.remove(requestId);
if (future != null) {
future.complete(message);
}
}
/**
* 模拟数据推送
*/
private void simulateDataPush() {
scheduler.scheduleAtFixedRate(() -> {
System.out.println("模拟数据推送: 当前等待请求数 " + requests.size());
// 可以在这里添加业务逻辑
}, 5, 5, TimeUnit.SECONDS);
}
}
3 Spring WebFlux实现（响应式）
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@RestController
public class WebFluxLongPollingController {
// 使用Sinks实现多播
private final Map<String, Sinks.Many<String>> sinks = new ConcurrentHashMap<>();
@GetMapping("/api/reactive-long-polling")
public Flux<ServerSentEvent<String>> reactiveLongPolling(String requestId) {
Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();
sinks.put(requestId, sink);
// 设置超时
return sink.asFlux()
.take(1)
.map(data -> ServerSentEvent.builder(data).build())
.timeout(Duration.ofSeconds(30))
.doFinally(signalType -> sinks.remove(requestId));
}
@GetMapping("/api/reactive-trigger")
public String reactiveTrigger(String requestId, String message) {
Sinks.Many<String> sink = sinks.get(requestId);
if (sink != null) {
sink.tryEmitNext(message);
return "sent";
}
return "not found";
}
}
客户端实现
1 Java原生客户端
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class LongPollingClient {
private static final String SERVER_URL = "http://localhost:8080/api/long-polling";
private static final ExecutorService executor = Executors.newCachedThreadPool();
public static void main(String[] args) {
// 启动多个客户端
for (int i = 0; i < 3; i++) {
final String clientId = "client-" + i;
executor.submit(() -> startLongPolling(clientId));
}
// 防止主线程退出
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static void startLongPolling(String clientId) {
while (true) {
try {
String response = sendLongPollingRequest(clientId);
System.out.println(clientId + " 收到响应: " + response);
// 收到响应后立即发起新的请求
Thread.sleep(100); // 短暂延迟
} catch (Exception e) {
System.err.println(clientId + " 请求异常: " + e.getMessage());
try {
Thread.sleep(1000); // 错误延迟
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
break;
}
}
}
}
private static String sendLongPollingRequest(String clientId) throws Exception {
String urlStr = SERVER_URL + "?requestId=" + clientId + "&timeout=30000";
URL url = new URL(urlStr);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod("GET");
conn.setConnectTimeout(2000);
conn.setReadTimeout(35000); // 稍大于服务端超时
int responseCode = conn.getResponseCode();
if (responseCode == HttpURLConnection.HTTP_OK) {
BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream()));
String response = in.readLine();
in.close();
return response;
} else {
return "HTTP Error: " + responseCode;
}
}
}
2 Spring WebClient客户端
import org.springframework.web.reactive.function.client.WebClient;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
public class WebClientLongPolling {
private final WebClient webClient;
private final AtomicInteger requestCount = new AtomicInteger(0);
public WebClientLongPolling() {
this.webClient = WebClient.builder()
.baseUrl("http://localhost:8080")
.build();
}
public void startPolling(String clientId) {
webClient.get()
.uri(uriBuilder -> uriBuilder
.path("/api/long-polling")
.queryParam("requestId", clientId)
.queryParam("timeout", 30000)
.build())
.retrieve()
.bodyToMono(String.class)
.timeout(Duration.ofSeconds(35))
.doOnSuccess(response -> {
System.out.println("收到响应: " + response);
int count = requestCount.incrementAndGet();
System.out.println("请求次数: " + count);
// 递归调用
startPolling(clientId);
})
.doOnError(error -> {
System.err.println("请求错误: " + error.getMessage());
// 延迟重试
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
startPolling(clientId);
})
.subscribe();
}
public static void main(String[] args) {
WebClientLongPolling client = new WebClientLongPolling();
client.startPolling("webclient-1");
// 防止主线程退出
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
3 HTML/JavaScript客户端
<!DOCTYPE html>
<html>
<head>Long Polling Demo</title>
<script>
class LongPollingClient {
constructor(url) {
this.url = url;
this.isPolling = false;
this.requestCount = 0;
}
async start() {
if (this.isPolling) return;
this.isPolling = true;
document.getElementById('status').textContent = '开始轮询...';
this.poll();
}
async poll() {
if (!this.isPolling) return;
try {
document.getElementById('status').textContent = '等待数据...';
const response = await fetch(this.url);
const data = await response.text();
this.requestCount++;
document.getElementById('requests').textContent = this.requestCount;
document.getElementById('messages').innerHTML +=
`收到消息: ${data}<br>`;
console.log('收到响应:', data);
// 再次发起请求
this.poll();
} catch (error) {
console.error('请求失败:', error);
document.getElementById('status').textContent = '请求失败，重试中...';
// 错误延迟重试
setTimeout(() => this.poll(), 1000);
}
}
stop() {
this.isPolling = false;
document.getElementById('status').textContent = '已停止';
}
}
// 初始化
const client = new LongPollingClient('/api/long-polling?requestId=client1&timeout=30000');
</script>
<style>
body { font-family: Arial, sans-serif; padding: 20px; }
.container { max-width: 600px; margin: 0 auto; }
#messages {
border: 1px solid #ddd;
padding: 10px;
min-height: 200px;
margin: 20px 0;
}
button { margin: 5px; padding: 10px; }
</style>
</head>
<body>
<div class="container">
<h1>长轮询演示</h1>
<p>状态: <span id="status">未启动</span></p>
<p>请求次数: <span id="requests">0</span></p>
<button onclick="client.start()">开始轮询</button>
<button onclick="client.stop()">停止轮询</button>
<div id="messages">消息列表：<br></div>
</div>
</body>
</html>
完整配置示例
1 Spring Boot配置
# application.yml
spring:
application:
name: long-polling-demo
server:
port: 8080
tomcat:
max-threads: 200
min-spare-threads: 50
max-connections: 10000
accept-count: 100
logging:
level:
com.example: DEBUG
2 配置类
import org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory;
import org.springframework.boot.web.server.WebServerFactoryCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ServerConfig {
@Bean
public WebServerFactoryCustomizer<TomcatServletWebServerFactory> containerCustomizer() {
return factory -> {
// 设置连接超时
factory.addConnectorCustomizers(connector -> {
connector.setKeepAliveTimeout(60000);
connector.setMaxKeepAliveRequests(-1); // 无限
});
};
}
}
测试脚本
import java.util.concurrent.atomic.AtomicInteger;
public class LongPollingLoadTest {
private static final int THREAD_COUNT = 10;
private static final AtomicInteger successCount = new AtomicInteger(0);
private static final AtomicInteger timeoutCount = new AtomicInteger(0);
public static void main(String[] args) throws InterruptedException {
System.out.println("开始负载测试，线程数: " + THREAD_COUNT);
Thread[] threads = new Thread[THREAD_COUNT];
long startTime = System.currentTimeMillis();
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
threads[i] = new Thread(() -> {
try {
for (int j = 0; j < 10; j++) {
testRequest(threadId, j);
}
} catch (Exception e) {
System.err.println("线程 " + threadId + " 异常: " + e.getMessage());
}
});
threads[i].start();
}
// 等待所有线程完成
for (Thread thread : threads) {
thread.join();
}
long duration = System.currentTimeMillis() - startTime;
System.out.println("\n测试完成，耗时: " + duration + "ms");
System.out.println("成功请求: " + successCount.get());
System.out.println("超时请求: " + timeoutCount.get());
}
private static void testRequest(int threadId, int requestNum) throws Exception {
String requestId = "thread-" + threadId + "-req-" + requestNum;
long startTime = System.currentTimeMillis();
String response = LongPollingClientUtil.sendRequest(requestId);
long duration = System.currentTimeMillis() - startTime;
if ("timeout".equals(response)) {
timeoutCount.incrementAndGet();
System.out.println(requestId + " 超时，耗时: " + duration + "ms");
} else {
successCount.incrementAndGet();
System.out.println(requestId + " 成功: " + response + "，耗时: " + duration + "ms");
}
}
}
// 客户端工具类
class LongPollingClientUtil {
public static String sendRequest(String requestId) throws Exception {
// 简化实现，实际应使用HTTP客户端
java.net.URL url = new java.net.URL(
"http://localhost:8080/api/long-polling?requestId=" + requestId + "&timeout=1000"
);
java.net.HttpURLConnection conn = (java.net.HttpURLConnection) url.openConnection();
conn.setRequestMethod("GET");
conn.setConnectTimeout(2000);
conn.setReadTimeout(5000);
int code = conn.getResponseCode();
if (code == 200) {
java.io.BufferedReader reader = new java.io.BufferedReader(
new java.io.InputStreamReader(conn.getInputStream())
);
return reader.readLine();
}
return "error-" + code;
}
}
注意事项
- 连接管理：使用连接池避免频繁创建连接
- 超时处理：合理设置超时时间，避免资源浪费
- 并发控制：限制最大并发数，防止服务器过载
- 心跳机制：可考虑WebSocket或SSE实现心跳检测
- 失败重试：实现优雅的重试机制，避免雪崩
这个案例提供了完整的Java长轮询实现,包括服务端、客户端和测试代码，可以直接应用到实际项目中。
