<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>org.yelang</groupId>
	<artifactId>baidu-sse-client</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.7.18</version> <!-- 兼容JDK 8的Spring Boot版本 -->
		<relativePath /> <!-- lookup parent from repository -->
	</parent>
	<properties>
		<java.version>1.8</java.version> <!-- 使用JDK 8 -->
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
	</properties>
	<dependencies>
		<!-- Spring Boot WebFlux (包含WebClient) -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-webflux</artifactId>
		</dependency>
		<!-- Spring Boot Web (可選，如果也需要傳統(tǒng)Web MVC) -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<!-- Jackson JSON處理 -->
		<dependency>
			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
		</dependency>
		<!-- SpringBoot集成thymeleaf模板 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-thymeleaf</artifactId>
		</dependency>
		<!-- 測(cè)試依賴 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>io.projectreactor</groupId>
			<artifactId>reactor-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>
	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
			<!-- 確保使用JDK 8編譯 -->
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<configuration>
					<source>1.8</source>
					<target>1.8</target>
				</configuration>
			</plugin>
		</plugins>
	</build>
</project>

package org.yelang.service;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@Service
public class SseService {
	// 保存所有連接的 emitter
	private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
	private final AtomicInteger counter = new AtomicInteger(0);
	/**
	 * -創(chuàng)建新的 SSE 連接
	 */
	public SseEmitter createEmitter(String clientId) {
		// 設(shè)置超時(shí)時(shí)間（0表示永不超時(shí)）
		SseEmitter emitter = new SseEmitter(0L);
		emitters.put(clientId, emitter);
		// 設(shè)置完成和超時(shí)回調(diào)
		emitter.onCompletion(() -> {
			emitters.remove(clientId);
			System.out.println("SSE連接完成: " + clientId);
		});
		emitter.onTimeout(() -> {
			emitters.remove(clientId);
			System.out.println("SSE連接超時(shí): " + clientId);
		});
		emitter.onError((e) -> {
			emitters.remove(clientId);
			System.out.println("SSE連接錯(cuò)誤: " + clientId + ", 錯(cuò)誤: " + e.getMessage());
		});
		return emitter;
	}
	/**
	 * -發(fā)送消息給所有客戶端
	 */
	public void sendToAll(String message) {
		emitters.forEach((clientId, emitter) -> {
			try {
				SseEmitter.SseEventBuilder event = SseEmitter.event().data(message)
						.id(String.valueOf(counter.incrementAndGet())).name("message").reconnectTime(5000L);
				emitter.send(event);
			} catch (IOException e) {
				emitter.completeWithError(e);
				emitters.remove(clientId);
			}
		});
	}
	/**
	 * -發(fā)送消息給特定客戶端
	 */
	public void sendToClient(String clientId, String message) {
		SseEmitter emitter = emitters.get(clientId);
		if (emitter != null) {
			try {
				SseEmitter.SseEventBuilder event = SseEmitter.event().data(message)
						.id(String.valueOf(counter.incrementAndGet())).name("message");
				emitter.send(event);
			} catch (IOException e) {
				emitter.completeWithError(e);
				emitters.remove(clientId);
			}
		}
	}
	/**
	 * -獲取當(dāng)前連接數(shù)
	 */
	public int getConnectionCount() {
		return emitters.size();
	}
}

package org.yelang.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import org.yelang.service.SseService;
import java.util.UUID;
@Controller
@RequestMapping("/sseman")
public class SseController {
	@Autowired
	private SseService sseService;
	/**
	 * -首頁
	 */
	@GetMapping("/index")
	public String index(Model model) {
		model.addAttribute("connectionCount", sseService.getConnectionCount());
		return "sse/index";
	}
	/**
	 * -建立 SSE 連接
	 */
	@GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	public SseEmitter streamSse(@RequestParam(value = "clientId", required = false) String clientId) {
		if (clientId == null || clientId.trim().isEmpty()) {
			clientId = UUID.randomUUID().toString();
		}
		return sseService.createEmitter(clientId);
	}
	/**
	 * -發(fā)送消息給所有客戶端
	 */
	@PostMapping("/broadcast")
	@ResponseBody
	public String broadcastMessage(@RequestParam String message) {
		sseService.sendToAll(message);
		return "消息已廣播";
	}
	/**
	 * - 發(fā)送消息給特定客戶端
	 */
	@PostMapping("/send-to-client")
	@ResponseBody
	public String sendToClient(@RequestParam String clientId, @RequestParam String message) {
		sseService.sendToClient(clientId, message);
		return "消息已發(fā)送給客戶端: " + clientId;
	}
	/**
	 * -獲取連接數(shù)
	 */
	@GetMapping("/connection-count")
	@ResponseBody
	public int getConnectionCount() {
		return sseService.getConnectionCount();
	}
}

<style>
body {
	font-family: Arial, sans-serif;
	max-width: 90%;
	margin: 0 auto;
	padding: 5px;
}
.container {
	border: 1px solid #ddd;
	border-radius: 5px;
	padding: 10px;
	margin-bottom: 10px;
}
.message-area {
	height: 300px;
	border: 1px solid #ccc;
	padding: 10px;
	overflow-y: auto;
	background-color: #f9f9f9;
}
.input-group {
	margin-bottom: 10px;
}
input, button {
	padding: 8px;
	margin: 5px;
}
button {
	background-color: #007bff;
	color: white;
	border: none;
	border-radius: 3px;
	cursor: pointer;
}
button:hover {
	background-color: #0056b3;
}
.status {
	color: #28a745;
	font-weight: bold;
}
.error {
	color: #dc3545;
}
.message {
	margin: 5px 0;
	padding: 5px;
	border-left: 3px solid #007bff;
	background-color: white;
}
/* 新增樣式：雙列布局 */
.row-container {
	display: flex;
	gap: 20px;
}
.row-container .container {
	flex: 1;
	margin-bottom: 10px;
}
/* 響應(yīng)式設(shè)計(jì)：小屏幕時(shí)恢復(fù)單列布局 */
@media (max-width: 768px) {
	.row-container {
		flex-direction: column;
	}
}
</style>

<!-- 新增：將連接狀態(tài)和發(fā)送消息放在同一行 -->
	<div class="row-container">
		<div class="container">
			<h4>連接狀態(tài)</h4>
			<div>
				<span>當(dāng)前連接數(shù): </span> <span id="connectionCount"
					th:text="${connectionCount}" class="status">0</span>
			</div>
			<div>
				<span>我的客戶端ID: </span> <span id="clientId" class="status">未連接</span>
			</div>
			<div class="input-group">
				<button onclick="connectSSE()">連接 SSE</button>
				<button onclick="disconnectSSE()" style="background-color: #dc3545;">斷開連接</button>
				<button onclick="refreshConnectionCount()">刷新連接數(shù)</button>
			</div>
		</div>
		<div class="container">
			<h4>發(fā)送消息</h4>
			<div class="input-group">
				<input type="text" id="broadcastMessage" placeholder="輸入要廣播的消息">
				<button onclick="broadcastMessage()">廣播給所有客戶端</button>
			</div>
			<div class="input-group">
				<input type="text" id="targetClientId" placeholder="目標(biāo)客戶端ID">
				<input type="text" id="privateMessage" placeholder="輸入私密消息">
				<button onclick="sendPrivateMessage()">發(fā)送給特定客戶端</button>
			</div>
		</div>
	</div>
	<div class="container">
		<h4>接收消息</h4>
		<div id="messageArea" class="message-area">
			<!-- 消息將在這里顯示 -->
		</div>
	</div>

function generateClientId() {
    return 'client_' + Math.random().toString(36).substr(2, 9);
}

let eventSource = null;
let clientId = null;   
var ctx = "/bdsse/sseman";
function connectSSE() {
    if (eventSource) {
        addMessage('警告', '已經(jīng)連接到SSE服務(wù)器');
        return;
    }
    // 生成客戶端ID
    clientId = generateClientId();
    document.getElementById('clientId').textContent = clientId;
    // 創(chuàng)建 EventSource 連接
    eventSource = new EventSource(ctx + '/sse?clientId=' + clientId);
    // 處理消息事件
    eventSource.onmessage = function(event) {
        addMessage('服務(wù)器消息', event.data);
    };
    // 處理自定義事件
    eventSource.addEventListener('message', function(event) {
        addMessage('自定義消息', event.data);
    });
    // 處理連接打開
    eventSource.onopen = function(event) {
        addMessage('系統(tǒng)', 'SSE連接已建立');
        refreshConnectionCount();
    };
    // 處理錯(cuò)誤
    eventSource.onerror = function(event) {
        if (eventSource.readyState === EventSource.CLOSED) {
            addMessage('系統(tǒng)', 'SSE連接已關(guān)閉');
        } else {
            addMessage('錯(cuò)誤', 'SSE連接錯(cuò)誤: ' + event);
        }
    };
    addMessage('系統(tǒng)', '正在連接SSE服務(wù)器...');
 }

 function disconnectSSE() {
     console.log("斷開連接");
     if (eventSource) {
         eventSource.close();
         eventSource = null;
         addMessage('系統(tǒng)', 'SSE連接已斷開');
         refreshConnectionCount();
     } else {
         addMessage('警告', '沒有活動(dòng)的SSE連接');
     }
 }

function broadcastMessage() {
    const message = document.getElementById('broadcastMessage').value;
    if (!message) {
        alert('請(qǐng)輸入消息');
        return;
    }
    fetch(ctx + '/broadcast', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/x-www-form-urlencoded',
        },
        body: 'message=' + encodeURIComponent(message)
    })
    .then(response => response.text())
    .then(data => {
        addMessage('操作', data);
        document.getElementById('broadcastMessage').value = '';
    })
    .catch(error => {
        addMessage('錯(cuò)誤', '發(fā)送失敗: ' + error);
    });
}

 function sendPrivateMessage() {
            const targetClientId = document.getElementById('targetClientId').value;
            const message = document.getElementById('privateMessage').value;
            if (!targetClientId || !message) {
                alert('請(qǐng)輸入客戶端ID和消息');
                return;
            }
            fetch(ctx + '/send-to-client', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded',
                },
                body: 'clientId=' + encodeURIComponent(targetClientId) + 
                      '&message=' + encodeURIComponent(message)
            })
            .then(response => response.text())
            .then(data => {
                addMessage('操作', data);
                document.getElementById('privateMessage').value = '';
            })
            .catch(error => {
                addMessage('錯(cuò)誤', '發(fā)送失敗: ' + error);
            });
        }