Java长轮询案例

wen java案例 2026-08-03 199

本文目录导读：

Java长轮询案例

基础概念
服务端实现
客户端实现
完整配置示例
测试脚本
注意事项

我将为您提供一个完整的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;
    }
}