springboot結(jié)合vue實(shí)現(xiàn)sse接口詳細(xì)流程

更新時(shí)間：2026年02月05日 09:19:39 作者：IT界Tony哥

這篇文章主要為大家詳細(xì)介紹了springboot結(jié)合vue實(shí)現(xiàn)sse接口詳細(xì)流程,文中的示例代碼講解詳細(xì),具有一定的借鑒價(jià)值,感興趣的小伙伴可以了解下

1. Spring Boot 后端實(shí)現(xiàn)

1.1 添加依賴

<!-- pom.xml -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

1.2 SSE控制器

// SseController.java
@RestController
@RequestMapping("/api/sse")
@CrossOrigin(origins = "*") // 允許跨域
public class SseController {
    
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    
    /**
     * 建立SSE連接
     */
    @GetMapping("/connect/{clientId}")
    public SseEmitter connect(@PathVariable String clientId) {
        // 設(shè)置超時(shí)時(shí)間（0表示永不超時(shí)）
        SseEmitter emitter = new SseEmitter(0L);
        
        // 注冊(cè)到Map中
        emitters.put(clientId, emitter);
        
        // 連接成功回調(diào)
        emitter.onCompletion(() -> {
            System.out.println("SSE連接完成: " + clientId);
            emitters.remove(clientId);
        });
        
        emitter.onTimeout(() -> {
            System.out.println("SSE連接超時(shí): " + clientId);
            emitters.remove(clientId);
        });
        
        emitter.onError((ex) -> {
            System.out.println("SSE連接錯(cuò)誤: " + clientId + ", 錯(cuò)誤: " + ex.getMessage());
            emitters.remove(clientId);
        });
        
        try {
            // 發(fā)送連接成功的消息
            emitter.send(SseEmitter.event()
                    .name("connect")
                    .data("連接成功")
                    .id(String.valueOf(System.currentTimeMillis())));
        } catch (IOException e) {
            emitter.completeWithError(e);
        }
        
        return emitter;
    }
    
    /**
     * 廣播消息給所有客戶端
     */
    @PostMapping("/broadcast")
    public ResponseEntity<String> broadcast(@RequestBody MessageDTO message) {
        sendToAllClients(message);
        return ResponseEntity.ok("廣播成功");
    }
    
    /**
     * 發(fā)送消息給指定客戶端
     */
    @PostMapping("/send/{clientId}")
    public ResponseEntity<String> sendMessage(
            @PathVariable String clientId, 
            @RequestBody MessageDTO message) {
        boolean success = sendToClient(clientId, message);
        if (success) {
            return ResponseEntity.ok("發(fā)送成功");
        } else {
            return ResponseEntity.status(HttpStatus.NOT_FOUND).body("客戶端不存在");
        }
    }
    
    /**
     * 獲取在線客戶端數(shù)量
     */
    @GetMapping("/online")
    public ResponseEntity<Map<String, Object>> getOnlineCount() {
        Map<String, Object> result = new HashMap<>();
        result.put("onlineCount", emitters.size());
        result.put("clientIds", new ArrayList<>(emitters.keySet()));
        return ResponseEntity.ok(result);
    }
    
    /**
     * 關(guān)閉指定客戶端的連接
     */
    @DeleteMapping("/close/{clientId}")
    public ResponseEntity<String> closeConnection(@PathVariable String clientId) {
        SseEmitter emitter = emitters.get(clientId);
        if (emitter != null) {
            emitter.complete();
            emitters.remove(clientId);
            return ResponseEntity.ok("連接已關(guān)閉");
        }
        return ResponseEntity.status(HttpStatus.NOT_FOUND).body("客戶端不存在");
    }
    
    /**
     * 向所有客戶端發(fā)送消息
     */
    private void sendToAllClients(MessageDTO message) {
        List<String> deadClients = new ArrayList<>();
        
        emitters.forEach((clientId, emitter) -> {
            try {
                emitter.send(SseEmitter.event()
                        .name(message.getType())
                        .data(message.getContent())
                        .id(String.valueOf(System.currentTimeMillis()))
                        .reconnectTime(3000)); // 重連時(shí)間
            } catch (IOException e) {
                deadClients.add(clientId);
                emitter.complete();
            }
        });
        
        // 清理無(wú)效連接
        deadClients.forEach(emitters::remove);
    }
    
    /**
     * 向指定客戶端發(fā)送消息
     */
    private boolean sendToClient(String clientId, MessageDTO message) {
        SseEmitter emitter = emitters.get(clientId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                        .name(message.getType())
                        .data(message.getContent())
                        .id(String.valueOf(System.currentTimeMillis())));
                return true;
            } catch (IOException e) {
                emitter.complete();
                emitters.remove(clientId);
            }
        }
        return false;
    }
}

1.3 消息DTO類

// MessageDTO.java
public class MessageDTO {
    private String type;      // 消息類型
    private String content;   // 消息內(nèi)容
    private Object data;      // 附加數(shù)據(jù)
    
    // 構(gòu)造方法、getter、setter
    public MessageDTO() {}
    
    public MessageDTO(String type, String content) {
        this.type = type;
        this.content = content;
    }
    
    public MessageDTO(String type, String content, Object data) {
        this.type = type;
        this.content = content;
        this.data = data;
    }
    
    // getter和setter方法
    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
    
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
    
    public Object getData() { return data; }
    public void setData(Object data) { this.data = data; }
}

1.4 配置類（可選）

// WebConfig.java
@Configuration
public class WebConfig implements WebMvcConfigurer {
    
    @Override
    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
        configurer.setDefaultTimeout(0); // 異步請(qǐng)求不超時(shí)
    }
}

2. Vue 前端實(shí)現(xiàn)

2.1 安裝依賴（可選）

npm install event-source-polyfill  # 如果需要兼容舊瀏覽器

2.2 SSE服務(wù)類

// src/services/sseService.js
class SSEService {
  constructor() {
    this.eventSource = null;
    this.clientId = null;
    this.listeners = new Map();
  }

  /**
   * 建立SSE連接
   * @param {string} clientId 客戶端ID
   * @param {string} baseURL 基礎(chǔ)URL
   */
  connect(clientId, baseURL = 'http://localhost:8080') {
    if (this.eventSource) {
      this.disconnect();
    }

    this.clientId = clientId;
    const url = `${baseURL}/api/sse/connect/${clientId}`;
    
    // 創(chuàng)建EventSource連接
    this.eventSource = new EventSource(url);

    // 監(jiān)聽連接打開
    this.eventSource.onopen = (event) => {
      console.log('SSE連接已建立');
      this.emit('connected', event);
    };

    // 監(jiān)聽消息
    this.eventSource.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        this.emit('message', data);
        
        // 根據(jù)事件名稱分發(fā)
        if (data.name) {
          this.emit(data.name, data.data || data);
        }
      } catch (error) {
        console.error('解析SSE消息失敗:', error);
        this.emit('raw-message', event.data);
      }
    };

    // 監(jiān)聽錯(cuò)誤
    this.eventSource.onerror = (event) => {
      console.error('SSE連接錯(cuò)誤:', event);
      this.emit('error', event);
      
      // 如果是網(wǎng)絡(luò)錯(cuò)誤，嘗試重連
      if (event.target.readyState === EventSource.CLOSED) {
        console.log('SSE連接已關(guān)閉，嘗試重連...');
        setTimeout(() => {
          this.connect(this.clientId, baseURL);
        }, 3000);
      }
    };

    return this;
  }

  /**
   * 添加事件監(jiān)聽器
   * @param {string} eventName 事件名稱
   * @param {function} callback 回調(diào)函數(shù)
   */
  on(eventName, callback) {
    if (!this.listeners.has(eventName)) {
      this.listeners.set(eventName, []);
    }
    this.listeners.get(eventName).push(callback);
    return this;
  }

  /**
   * 移除事件監(jiān)聽器
   * @param {string} eventName 事件名稱
   * @param {function} callback 回調(diào)函數(shù)
   */
  off(eventName, callback) {
    if (this.listeners.has(eventName)) {
      const callbacks = this.listeners.get(eventName);
      const index = callbacks.indexOf(callback);
      if (index > -1) {
        callbacks.splice(index, 1);
      }
    }
    return this;
  }

  /**
   * 觸發(fā)事件
   * @param {string} eventName 事件名稱
   * @param {any} data 數(shù)據(jù)
   */
  emit(eventName, data) {
    if (this.listeners.has(eventName)) {
      this.listeners.get(eventName).forEach(callback => {
        try {
          callback(data);
        } catch (error) {
          console.error(`執(zhí)行事件${eventName}的回調(diào)時(shí)出錯(cuò):`, error);
        }
      });
    }
  }

  /**
   * 發(fā)送消息到服務(wù)器
   * @param {object} message 消息對(duì)象
   * @param {string} endpoint 端點(diǎn)
   * @param {string} baseURL 基礎(chǔ)URL
   */
  async sendMessage(message, endpoint = 'broadcast', baseURL = 'http://localhost:8080') {
    try {
      const url = `${baseURL}/api/sse/${endpoint}`;
      const options = {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify(message)
      };

      const response = await fetch(url, options);
      return await response.json();
    } catch (error) {
      console.error('發(fā)送消息失敗:', error);
      throw error;
    }
  }

  /**
   * 發(fā)送到指定客戶端
   * @param {string} targetClientId 目標(biāo)客戶端ID
   * @param {object} message 消息對(duì)象
   * @param {string} baseURL 基礎(chǔ)URL
   */
  async sendToClient(targetClientId, message, baseURL = 'http://localhost:8080') {
    return this.sendMessage(message, `send/${targetClientId}`, baseURL);
  }

  /**
   * 斷開連接
   */
  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
      this.eventSource = null;
      console.log('SSE連接已斷開');
      this.emit('disconnected');
    }
  }

  /**
   * 檢查連接狀態(tài)
   */
  isConnected() {
    return this.eventSource && this.eventSource.readyState === EventSource.OPEN;
  }
}

// 創(chuàng)建單例實(shí)例
const sseService = new SSEService();

export default sseService;

2.3 Vue組件中使用

<!-- SSEDemo.vue -->
<template>
  <div class="sse-demo">
    <h2>SSE實(shí)時(shí)通信演示</h2>
    
    <!-- 連接控制 -->
    <div class="connection-panel">
      <input v-model="clientId" placeholder="輸入客戶端ID" />
      <button @click="connect" :disabled="isConnected">連接</button>
      <button @click="disconnect" :disabled="!isConnected">斷開</button>
      <span class="status" :class="{ connected: isConnected }">
        {{ isConnected ? '已連接' : '未連接' }}
      </span>
    </div>

    <!-- 消息發(fā)送 -->
    <div class="message-panel">
      <h3>發(fā)送消息</h3>
      <select v-model="messageType">
        <option value="chat">聊天</option>
        <option value="notification">通知</option>
        <option value="system">系統(tǒng)</option>
      </select>
      <input v-model="messageContent" placeholder="輸入消息內(nèi)容" />
      <button @click="sendBroadcast" :disabled="!isConnected">廣播</button>
      <button @click="sendToSelf" :disabled="!isConnected">發(fā)送給自己</button>
    </div>

    <!-- 在線信息 -->
    <div class="online-info">
      <button @click="getOnlineInfo" :disabled="!isConnected">刷新在線信息</button>
      <p>在線人數(shù): {{ onlineInfo.onlineCount || 0 }}</p>
    </div>

    <!-- 消息顯示 -->
    <div class="messages">
      <h3>收到的消息</h3>
      <div class="message-list">
        <div v-for="(msg, index) in messages" :key="index" 
             class="message-item" :class="msg.type">
          <span class="time">{{ formatTime(msg.timestamp) }}</span>
          <span class="type">[{{ msg.type }}]</span>
          <span class="content">{{ msg.content }}</span>
        </div>
      </div>
    </div>
  </div>
</template>

<script>
import sseService from '@/services/sseService';

export default {
  name: 'SSEDemo',
  data() {
    return {
      clientId: '',
      messageType: 'chat',
      messageContent: '',
      isConnected: false,
      messages: [],
      onlineInfo: {}
    };
  },
  mounted() {
    // 生成隨機(jī)客戶端ID
    this.clientId = 'client_' + Math.random().toString(36).substr(2, 9);
    
    // 注冊(cè)事件監(jiān)聽器
    this.registerEventListeners();
  },
  beforeUnmount() {
    // 組件銷毀前斷開連接
    sseService.disconnect();
  },
  methods: {
    registerEventListeners() {
      // 連接成功
      sseService.on('connected', () => {
        this.isConnected = true;
        this.addMessage('system', '連接成功');
      });

      // 連接斷開
      sseService.on('disconnected', () => {
        this.isConnected = false;
        this.addMessage('system', '連接已斷開');
      });

      // 接收消息
      sseService.on('message', (data) => {
        this.addMessage('message', JSON.stringify(data));
      });

      // 特定類型消息
      sseService.on('chat', (data) => {
        this.addMessage('chat', typeof data === 'string' ? data : JSON.stringify(data));
      });

      sseService.on('notification', (data) => {
        this.addMessage('notification', typeof data === 'string' ? data : JSON.stringify(data));
      });

      sseService.on('system', (data) => {
        this.addMessage('system', typeof data === 'string' ? data : JSON.stringify(data));
      });

      // 錯(cuò)誤
      sseService.on('error', (error) => {
        console.error('SSE錯(cuò)誤:', error);
        this.addMessage('system', '連接錯(cuò)誤');
      });
    },

    connect() {
      if (!this.clientId.trim()) {
        alert('請(qǐng)輸入客戶端ID');
        return;
      }
      sseService.connect(this.clientId);
    },

    disconnect() {
      sseService.disconnect();
    },

    async sendBroadcast() {
      if (!this.messageContent.trim()) {
        alert('請(qǐng)輸入消息內(nèi)容');
        return;
      }

      try {
        await sseService.sendMessage({
          type: this.messageType,
          content: this.messageContent
        });
        this.messageContent = '';
      } catch (error) {
        alert('發(fā)送失敗: ' + error.message);
      }
    },

    async sendToSelf() {
      if (!this.messageContent.trim()) {
        alert('請(qǐng)輸入消息內(nèi)容');
        return;
      }

      try {
        await sseService.sendToClient(this.clientId, {
          type: this.messageType,
          content: `[自發(fā)送] ${this.messageContent}`
        });
        this.messageContent = '';
      } catch (error) {
        alert('發(fā)送失敗: ' + error.message);
      }
    },

    async getOnlineInfo() {
      try {
        const response = await fetch('http://localhost:8080/api/sse/online');
        this.onlineInfo = await response.json();
      } catch (error) {
        console.error('獲取在線信息失敗:', error);
      }
    },

    addMessage(type, content) {
      this.messages.unshift({
        type,
        content,
        timestamp: new Date()
      });
      
      // 限制消息數(shù)量
      if (this.messages.length > 50) {
        this.messages.pop();
      }
    },

    formatTime(timestamp) {
      return new Date(timestamp).toLocaleTimeString();
    }
  }
};
</script>

<style scoped>
.sse-demo {
  max-width: 800px;
  margin: 0 auto;
  padding: 20px;
}

.connection-panel, .message-panel, .online-info {
  margin-bottom: 20px;
  padding: 15px;
  border: 1px solid #ddd;
  border-radius: 5px;
}

.connection-panel input, .message-panel input, .message-panel select {
  margin-right: 10px;
  padding: 5px 10px;
}

button {
  padding: 5px 15px;
  margin-right: 10px;
  cursor: pointer;
}

button:disabled {
  cursor: not-allowed;
  opacity: 0.5;
}

.status {
  padding: 2px 8px;
  border-radius: 3px;
}

.status.connected {
  background-color: #4CAF50;
  color: white;
}

.messages {
  border: 1px solid #ddd;
  border-radius: 5px;
  padding: 15px;
}

.message-list {
  height: 300px;
  overflow-y: auto;
  border: 1px solid #eee;
  padding: 10px;
}

.message-item {
  margin-bottom: 10px;
  padding: 5px;
  border-radius: 3px;
}

.message-item.chat {
  background-color: #e3f2fd;
}

.message-item.notification {
  background-color: #fff3e0;
}

.message-item.system {
  background-color: #f3e5f5;
}

.time {
  font-size: 12px;
  color: #666;
  margin-right: 10px;
}

.type {
  font-weight: bold;
  margin-right: 10px;
}
</style>

3. 高級(jí)特性示例

3.1 帶認(rèn)證的SSE連接

// 在Controller中添加認(rèn)證
@GetMapping("/connect/{clientId}")
public SseEmitter connect(@PathVariable String clientId, 
                         @RequestHeader(value = "Authorization", required = false) String token) {
    // 驗(yàn)證token邏輯
    if (!validateToken(token)) {
        throw new SecurityException("未授權(quán)訪問");
    }
    
    // ... 其余代碼相同
}

3.2 心跳檢測(cè)

// 定期發(fā)送心跳
@Component
public class HeartbeatScheduler {
    
    @Autowired
    private SseController sseController;
    
    @Scheduled(fixedRate = 30000) // 每30秒發(fā)送一次心跳
    public void sendHeartbeat() {
        MessageDTO heartbeat = new MessageDTO("heartbeat", "ping");
        sseController.sendToAllClients(heartbeat);
    }
}

這個(gè)完整的實(shí)現(xiàn)提供了：

后端功能：連接管理、消息廣播、定向發(fā)送、連接監(jiān)控
前端功能：連接管理、消息收發(fā)、事件監(jiān)聽、狀態(tài)顯示
錯(cuò)誤處理：自動(dòng)重連、連接狀態(tài)監(jiān)控
擴(kuò)展性：易于添加新的消息類型和業(yè)務(wù)邏輯

你可以根據(jù)具體需求調(diào)整消息格式、認(rèn)證方式和業(yè)務(wù)邏輯。

到此這篇關(guān)于springboot結(jié)合vue實(shí)現(xiàn)sse接口詳細(xì)流程的文章就介紹到這了,更多相關(guān)springboot vue實(shí)現(xiàn)sse接口內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: