SpringBoot+SSE 构建 AI 实时流式对话系统：原理剖析与代码实战

SpringBoot+SSE 构建 AI 实时流式对话系统：原理剖析与代码实战 | 极客日志

 data: 这是第一条推送消息\n\n data: 这是第二条推送消息\n\n

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@RestController
public class SseController {
    private static final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

    @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter sse() {
        SseEmitter emitter = new SseEmitter();
        executor.scheduleAtFixedRate(() -> {
            try {
                String data = "这是实时推送的数据：" + System.currentTimeMillis();
                emitter.send(ServerSentEvent.builder(data).build());
            } catch (IOException e) {
                emitter.completeWithError(e);
            }
        }, 0, 5, TimeUnit.SECONDS);
        emitter.onTimeout(() -> emitter.complete());
        emitter.onError((e) -> emitter.completeWithError(e));
        return emitter;
    }
}

特性	SSE	WebSocket
通信方向	单向（服务器→客户端）	全双工双向（服务器⇔客户端）
协议	基于 HTTP 协议	独立的 WebSocket 协议
实现复杂度	较低，基于 HTTP，无需复杂的协议升级	较高，需要专门的握手过程来升级协议
兼容性	现代浏览器基本都支持，兼容性较好	广泛支持，但在一些旧版本浏览器中可能存在兼容性问题
重连机制	浏览器内置自动重连机制	需要手动实现心跳检测和重连逻辑
适用场景	适用于服务器单向推送数据的场景，如新闻推送、实时监控、AI 对话等	适用于需要双向实时通信的场景，如在线聊天、实时协作、游戏等

const eventSource = new EventSource('/ai/chat/stream');
eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    // 将接收到的数据显示在聊天界面
    document.getElementById('chat-content').innerHTML += data.message + '<br>';
};
eventSource.onerror = (err) => {
    console.error('SSE 错误:', err);
};

spring.mvc.async.request-timeout=60000

import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class SSEServer {
    // 存放所有用户的 SseEmitter 连接
    private static final Map<String, SseEmitter> sseClients = new ConcurrentHashMap<>();

    // 建立连接
    public static SseEmitter connect(String userId) {
        // 设置超时时间为 0，即不超时，默认是 30 秒，超时未完成任务则会抛出异常
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册连接完成、超时、异常时的回调函数
        sseEmitter.onTimeout(timeoutCallback(userId));
        sseEmitter.onCompletion(completionCallback(userId));
        sseEmitter.onError(errorCallback(userId));
        sseClients.put(userId, sseEmitter);
        System.out.println("SSE connect, userId: " + userId);
        return sseEmitter;
    }

    // 发送消息
    public static void sendMsg(String userId, String message, SSEMsgType msgType) {
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }
        if (sseClients.containsKey(userId)) {
            SseEmitter sseEmitter = sseClients.get(userId);
            sendEmitterMessage(sseEmitter, userId, message, msgType);
        }
    }

    private static void sendEmitterMessage(SseEmitter sseEmitter, String userId, String message, SSEMsgType msgType) {
        // 指定事件名称 (name)，前端根据这个名称监听
        SseEmitter.SseEventBuilder msgEvent = SseEmitter.event().id(userId).data(message).name(msgType.type);
        try {
            sseEmitter.send(msgEvent);
        } catch (IOException e) {
            System.out.println("SSE send message error, userId: " + userId + ", error: " + e.getMessage());
            close(userId); // 发送异常时，移除该连接
        }
    }

    // 关闭连接
    public static void close(String userId) {
        SseEmitter emitter = sseClients.get(userId);
        if (emitter != null) {
            emitter.complete(); // 这会触发 onCompletion 回调，回调中已经包含了 remove 操作
        }
    }

    // 连接超时的回调函数
    private static Runnable timeoutCallback(String userId) {
        return () -> {
            System.out.println("SSE connection timeout, userId: " + userId);
            close(userId);
        };
    }

    // 连接完成的回调函数
    private static Runnable completionCallback(String userId) {
        return () -> {
            System.out.println("SSE connection completed, userId: " + userId);
            sseClients.remove(userId);
        };
    }

    // 连接错误的回调函数
    private static java.util.function.Consumer<Throwable> errorCallback(String userId) {
        return throwable -> {
            System.out.println("SSE connection error, userId: " + userId + ", error: " + throwable.getMessage());
            close(userId);
        };
    }
}

public enum SSEMsgType {
    MESSAGE("message", "单次发送的普通信息"),
    ADD("add", "消息追加，适用于流式 stream 推送"),
    FINISH("finish", "消息发送完成");

    public final String type;
    public final String value;

    SSEMsgType(String type, String value) {
        this.type = type;
        this.value = value;
    }
}

import com.example.demo.enums.SSEMsgType;
import com.example.demo.utils.SSEServer;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
@RequestMapping("/sse")
public class SSEController {
    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect(@RequestParam String userId) {
        return SSEServer.connect(userId);
    }
}

import org.springframework.http.MediaType;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.web.client.ResponseExtractor;
import org.springframework.web.client.RestTemplate;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class DeepSeekClient {
    private static final String API_URL = "https://api.deepseek.com/stream";
    private final RestTemplate restTemplate;

    public DeepSeekClient(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    public void streamChat(String prompt, String userId) {
        restTemplate.execute(API_URL, HttpMethod.POST, request -> {
            request.getHeaders().setContentType(MediaType.APPLICATION_JSON);
            // 构建请求参数
            String requestBody = "{\"prompt\":\"" + prompt + "\",\"stream\":true}";
            request.getBody().write(requestBody.getBytes());
            return request;
        }, new ResponseExtractor<Void>() {
            @Override
            public Void extractData(ClientHttpResponse response) throws IOException {
                BufferedReader reader = new BufferedReader(new InputStreamReader(response.getBody()));
                String line;
                while ((line = reader.readLine()) != null) {
                    // 处理每一行响应数据，这里假设每行是一个完整的文本片段
                    SSEServer.sendMsg(userId, line, SSEMsgType.ADD);
                }
                return null;
            }
        });
    }
}

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;
import java.lang.reflect.Method;
import java.util.concurrent.Executor;

@Service
@EnableAsync
public class AIService {
    @Autowired
    private DeepSeekClient deepSeekClient;

    @Async
    public void callAI(String prompt, String userId) {
        try {
            deepSeekClient.streamChat(prompt, userId);
            // AI 生成完成后，推送消息发送完成信号
            SSEServer.sendMsg(userId, "AI 生成完成", SSEMsgType.FINISH);
        } catch (IOException e) {
            SSEServer.sendMsg(userId, "AI 调用发生错误：" + e.getMessage(), SSEMsgType.MESSAGE);
            SSEServer.close(userId);
        } catch (TimeoutException e) {
            SSEServer.sendMsg(userId, "AI 调用超时，请稍后重试", SSEMsgType.MESSAGE);
            SSEServer.close(userId);
        }
    }
}

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>AI 实时对话</title>
</head>
<body>
    <div id="chat-container"><!-- 聊天内容将显示在这里 --></div>
    <form id="question-form">
        <input type="text" id="question-input" placeholder="请输入问题">
        <button type="submit">发送</button>
    </form>
    <script>
        const userId = "user123"; // 假设用户 ID
        const eventSource = new EventSource(`/sse/connect?userId=${userId}`);
        eventSource.onopen = function() {
            console.log('SSE 连接已建立');
        };
        eventSource.onerror = function(error) {
            console.error('SSE 连接错误:', error);
            // 处理连接异常，例如自动重连
            setTimeout(() => {
                eventSource.close();
                eventSource = new EventSource(`/sse/connect?userId=${userId}`);
            }, 5000);
        };
    </script>
</body>
</html>

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    const chatContainer = document.getElementById('chat-container');
    const messageElement = document.createElement('p');
    if (data.type === 'ADD') {
        // 模拟打字机效果，逐字显示
        const text = data.message;
        let index = 0;
        const interval = setInterval(() => {
            messageElement.textContent += text[index];
            index++;
            if (index >= text.length) {
                clearInterval(interval);
            }
        }, 50);
        chatContainer.appendChild(messageElement);
    } else if (data.type === 'FINISH') {
        // 停止加载动画（假设之前有显示加载动画）
        const loadingElement = document.getElementById('loading');
        if (loadingElement) {
            loadingElement.style.display = 'none';
        }
        // 可以在这里更新对话状态，比如显示对话结束提示
        const endMessage = document.createElement('p');
        endMessage.textContent = 'AI 回复已完成';
        chatContainer.appendChild(endMessage);
    }
};

#chat-container p {
    margin: 5px;
    padding: 5px 10px;
    border-radius: 10px;
    max-width: 80%;
}
#chat-container p.user-question {
    background-color: #e1f5fe;
    align-self: flex-start;
}
#chat-container p.ai-answer {
    background-color: #dcedc8;
    align-self: flex-end;
}

// 添加用户提问
const questionForm = document.getElementById('question-form');
questionForm.addEventListener('submit', function(event) {
    event.preventDefault();
    const questionInput = document.getElementById('question-input');
    const question = questionInput.value;
    const userQuestionElement = document.createElement('p');
    userQuestionElement.textContent = `用户：${question}`;
    userQuestionElement.classList.add('user-question');
    const chatContainer = document.getElementById('chat-container');
    chatContainer.appendChild(userQuestionElement);
    // 发送问题到后端
    // 这里假设通过其他方式（如 fetch）将问题发送到后端
    questionInput.value = '';
});

// 添加 AI 回复
eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    const chatContainer = document.getElementById('chat-container');
    const messageElement = document.createElement('p');
    if (data.type === 'ADD') {
        // 模拟打字机效果，逐字显示
        const text = data.message;
        let index = 0;
        const interval = setInterval(() => {
            messageElement.textContent += text[index];
            index++;
            if (index >= text.length) {
                clearInterval(interval);
            }
        }, 50);
        messageElement.textContent = `AI: `;
        messageElement.classList.add('ai-answer');
        chatContainer.appendChild(messageElement);
    } else if (data.type === 'FINISH') {
        // 停止加载动画（假设之前有显示加载动画）
        const loadingElement = document.getElementById('loading');
        if (loadingElement) {
            loadingElement.style.display = 'none';
        }
        // 可以在这里更新对话状态，比如显示对话结束提示
        const endMessage = document.createElement('p');
        endMessage.textContent = 'AI 回复已完成';
        chatContainer.appendChild(endMessage);
    }
};

npm install @microsoft/fetch-event-source

import { fetchEventSource } from '@microsoft/fetch-event-source';
const userId = "user123"; // 假设用户 ID
const controller = new AbortController();
const signal = controller.signal;

fetchEventSource(`/sse/connect?userId=${userId}`, {
    method: 'GET',
    signal: signal,
    headers: {
        'Accept': 'text/event-stream'
    },
    onopen: async (response) => {
        if (response.ok) {
            console.log('SSE 连接已建立');
        } else {
            console.error('连接失败:', response.status);
        }
    },
    onmessage: (event) => {
        const data = JSON.parse(event.data);
        // 处理消息，同 EventSource 的 onmessage 处理
    },
    onerror: (err) => {
        console.error('SSE 连接错误:', err);
        // 处理连接异常，例如自动重连
        setTimeout(() => {
            controller.abort();
            fetchEventSource(`/sse/connect?userId=${userId}`, {
                method: 'GET',
                signal: new AbortController().signal,
                headers: {
                    'Accept': 'text/event-stream'
                },
                onopen: async (response) => {
                    if (response.ok) {
                        console.log('SSE 连接已重新建立');
                    } else {
                        console.error('重新连接失败:', response.status);
                    }
                },
                onmessage: (event) => {
                    const data = JSON.parse(event.data);
                    // 处理消息，同 EventSource 的 onmessage 处理
                },
                onerror: (err) => {
                    console.error('重新连接时错误:', err);
                }
            });
        }, 5000);
    }
});

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>AI 实时对话</title>
</head>
<body>
    <div id="connection-status">连接中...</div>
    <div id="chat-container"><!-- 聊天内容将显示在这里 --></div>
    <form id="question-form">
        <input type="text" id="question-input" placeholder="请输入问题">
        <button type="submit">发送</button>
    </form>
    <script>
        // 引入 fetch-event-source 库的代码...
        const connectionStatusElement = document.getElementById('connection-status');
        fetchEventSource(`/sse/connect?userId=${userId}`, {
            // 其他配置...
            onopen: async (response) => {
                if (response.ok) {
                    connectionStatusElement.textContent = '已连接';
                    console.log('SSE 连接已建立');
                } else {
                    connectionStatusElement.textContent = '连接失败';
                    console.error('连接失败:', response.status);
                }
            },
            onerror: (err) => {
                connectionStatusElement.textContent = '连接已断开，正在重试...';
                console.error('SSE 连接错误:', err);
                // 自动重连代码...
            }
        });
    </script>
</body>
</html>

spring.mvc.async.request-timeout=60000

private static void sendEmitterMessage(SseEmitter sseEmitter, String userId, String message, SSEMsgType msgType) {
    SseEmitter.SseEventBuilder msgEvent = SseEmitter.event().id(userId).data(message).name(msgType.type);
    try {
        sseEmitter.send(msgEvent);
    } catch (IOException e) {
        System.out.println("SSE send message error, userId: " + userId + ", error: " + e.getMessage());
        close(userId); // 发送异常时，移除该连接
    }
}

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    if (data.type === 'ADD') {
        // 处理消息
    } else if (data.type === 'FINISH') {
        // 处理消息
    }
};

<dependency>
    <groupId>com.zaxxer</groupId>
    <artifactId>HikariCP</artifactId>
    <version>4.0.3</version>
</dependency>

# 最大连接数
hikari.maximum-pool-size=100
# 最小空闲连接数
hikari.minimum-idle=10
# 连接超时时间
hikari.connection-timeout=30000

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class SSEServer {
    private static final HikariDataSource dataSource;
    private static final Map<String, SseEmitter> sseClients = new ConcurrentHashMap<>();

    static {
        HikariConfig config = new HikariConfig();
        config.setMaximumPoolSize(100);
        config.setMinimumIdle(10);
        config.setConnectionTimeout(30000);
        dataSource = new HikariDataSource(config);
    }

    public static SseEmitter connect(String userId) {
        SseEmitter sseEmitter = new SseEmitter(dataSource.getConnection());
        sseEmitter.onTimeout(timeoutCallback(userId));
        sseEmitter.onCompletion(completionCallback(userId));
        sseEmitter.onError(errorCallback(userId));
        sseClients.put(userId, sseEmitter);
        return sseEmitter;
    }

    // 发送消息和关闭连接等方法不变
}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

spring.redis.host=127.0.0.1
spring.redis.port=6379

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class RedisPublisher {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void publish(String channel, Object message) {
        redisTemplate.convertAndSend(channel, message);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class SSEServer {
    // 省略其他代码
    @Autowired
    private RedisPublisher redisPublisher;

    public static void sendMsg(String userId, String message, SSEMsgType msgType) {
        // 构建消息对象
        SSEMessage sseMessage = new SSEMessage(userId, message, msgType);
        redisPublisher.publish("sse-channel", sseMessage);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class RedisSubscriber implements MessageListener {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        SSEMessage sseMessage = (SSEMessage) redisTemplate.getValueSerializer().deserialize(message.getBody());
        String userId = sseMessage.getUserId();
        String messageContent = sseMessage.getMessage();
        SSEMsgType msgType = sseMessage.getMsgType();
        SSEServer.sendMsg(userId, messageContent, msgType);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
public class RedisConfig {
    @Autowired
    private RedisSubscriber redisSubscriber;

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(redisSubscriber);
        container.addMessageListener(messageListenerAdapter, new PatternTopic("sse-channel"));
        return container;
    }
}

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>

import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class SSEServer {
    // 省略其他代码
    @Retryable(value = IOException.class, maxAttempts = 3, backoff = @Backoff(delay = 1000))
    public static void sendMsg(String userId, String message, SSEMsgType msgType) {
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }
        if (sseClients.containsKey(userId)) {
            SseEmitter sseEmitter = sseClients.get(userId);
            sendEmitterMessage(sseEmitter, userId, message, msgType);
        }
    }

    private static void sendEmitterMessage(SseEmitter sseEmitter, String userId, String message, SSEMsgType msgType) throws IOException {
        SseEmitter.SseEventBuilder msgEvent = SseEmitter.event().id(userId).data(message).name(msgType.type);
        sseEmitter.send(msgEvent);
    }
}

import org.springframework.util.CollectionUtils;
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.UUID;

public class SSEServer {
    // 省略其他代码
    public static void sendMsg(String userId, String message, SSEMsgType msgType) {
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }
        if (sseClients.containsKey(userId)) {
            SseEmitter sseEmitter = sseClients.get(userId);
            String eventId = UUID.randomUUID().toString();
            sendEmitterMessage(sseEmitter, userId, message, msgType, eventId);
        }
    }

    private static void sendEmitterMessage(SseEmitter sseEmitter, String userId, String message, SSEMsgType msgType, String eventId) throws IOException {
        SseEmitter.SseEventBuilder msgEvent = SseEmitter.event().id(eventId).data(message).name(msgType.type);
        sseEmitter.send(msgEvent);
    }
}

const eventSource = new EventSource('/sse/connect?userId=${userId}', {
    withCredentials: true,
    headers: {
        'Last-Event-ID': lastEventId // lastEventId 为上次接收到的最后一个事件 ID
    }
});

import com.example.demo.enums.SSEMsgType;
import com.example.demo.utils.SSEServer;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
@RequestMapping("/sse")
public class SSEController {
    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect(@RequestParam String userId, @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {
        SseEmitter sseEmitter = SSEServer.connect(userId);
        if (lastEventId != null) {
            // 根据 lastEventId 判断是否需要重新推送消息
            // 这里可以实现具体的逻辑，例如查询已推送消息记录等
        }
        return sseEmitter;
    }
}

public interface AIModel {
    void streamChat(String prompt, String userId, SseEmitter emitter) throws IOException;
}

import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

public class OpenAIModel implements AIModel {
    private static final String API_URL = "https://api.openai.com/v1/chat/completions";
    private static final String API_KEY = "your-openai-api-key";

    @Override
    public void streamChat(String prompt, String userId, SseEmitter emitter) throws IOException {
        RestTemplate restTemplate = new RestTemplate();
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.setBearerAuth(API_KEY);
        // 构建请求体
        String requestBody = "{\"model\":\"gpt-3.5-turbo\",\"messages\":[{\"role\":\"user\",\"content\":\"" + prompt + "\"}],\"stream\":true}";
        HttpEntity<String> request = new HttpEntity<>(requestBody, headers);
        // 发送请求并处理响应
        ResponseEntity<String> response = restTemplate.postForEntity(API_URL, request, String.class);
        // 处理流式响应，将内容通过 SseEmitter 发送给客户端
        // 这里省略具体的处理逻辑，可参考之前的 DeepSeekClient 实现
    }
}

import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.web.client.RestTemplate;

public class DeepSeekModel implements AIModel {
    private static final String API_URL = "https://api.deepseek.com/stream";
    private static final String API_KEY = "your-deepseek-api-key";

    @Override
    public void streamChat(String prompt, String userId, SseEmitter emitter) throws IOException {
        RestTemplate restTemplate = new RestTemplate();
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.setBearerAuth(API_KEY);
        // 构建请求体
        String requestBody = "{\"prompt\":\"" + prompt + "\",\"stream\":true}";
        HttpEntity<String> request = new HttpEntity<>(requestBody, headers);
        // 发送请求并处理响应
        ResponseEntity<String> response = restTemplate.postForEntity(API_URL, request, String.class);
        // 处理流式响应，将内容通过 SseEmitter 发送给客户端
        // 这里省略具体的处理逻辑，可参考之前的 DeepSeekClient 实现
    }
}

import java.util.HashMap;
import java.util.Map;

public class AIModelFactory {
    private static final Map<String, AIModel> modelMap = new HashMap<>();

    static {
        modelMap.put("openai", new OpenAIModel());
        modelMap.put("deepseek", new DeepSeekModel());
        // 可以继续添加其他模型
    }

    public static AIModel getModel(String modelName) {
        return modelMap.get(modelName);
    }
}

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;
import java.lang.reflect.Method;
import java.util.concurrent.Executor;

@Service
@EnableAsync
public class AIService {
    @Value("${ai.model.name}")
    private String modelName;

    @Async
    public void callAI(String prompt, String userId, SseEmitter emitter) {
        AIModel model = AIModelFactory.getModel(modelName);
        try {
            model.streamChat(prompt, userId, emitter);
            // AI 生成完成后，推送消息发送完成信号
            SSEServer.sendMsg(userId, "AI 生成完成", SSEMsgType.FINISH);
        } catch (IOException e) {
            SSEServer.sendMsg(userId, "AI 调用发生错误：" + e.getMessage(), SSEMsgType.MESSAGE);
            SSEServer.close(userId);
        } catch (Exception e) {
            SSEServer.sendMsg(userId, "AI 调用发生未知错误", SSEMsgType.MESSAGE);
            SSEServer.close(userId);
        }
    }
}

ai.model.name=deepseek

SpringBoot+SSE 构建 AI 实时流式对话系统：原理剖析与代码实战

一、引言：告别等待！AI 实时对话的流式解决方案

1.1 传统 AI 对话的痛点与技术瓶颈

1.2 本文核心内容与技术栈说明

二、原理篇：SSE 为何是 AI 流式对话的最优解

2.1 SSE 核心原理：服务器单向推送的长连接技术

2.2 SSE vs WebSocket：技术选型的关键考量

2.3 AI 流式对话与 SSE 的契合点

三、实战篇：SpringBoot 整合 SSE 构建 AI 对话系统

3.1 环境准备：SpringBoot 项目搭建与依赖配置

3.2 核心组件 1：SSE 连接管理器（SSEServer）

3.2.1 连接管理的核心需求

3.2.2 关键方法实现（connect/sendMsg/close）

3.3 核心组件 2：消息类型枚举（SSEMsgType）

3.4 核心组件 3：SSE 控制器（SSEController）

3.5 业务层整合：AI 大模型流式 API 调用

3.5.1 AI 流式接口对接（以 DeepSeek 为例）

3.5.2 异步处理与异常兜底

四、前端实现：基于 EventSource 的实时消息渲染

4.1 EventSource 初始化与连接建立

4.2 流式文本渲染与交互优化

4.3 前端兼容性处理

五、测试与验证：一键运行实时对话系统

5.1 功能测试步骤

5.2 常见问题排查与解决方案

六、进阶优化：提升系统稳定性与扩展性

6.1 连接池优化与分布式适配

6.2 消息重试与幂等性保障

6.3 多模型集成与动态切换

七、总结与展望

7.1 本文核心内容总结

7.2 技术拓展方向

八、附录：完整代码仓库与资源链接

8.1 参考资料与学习链接

更多推荐文章

相关免费在线工具

SpringBoot+SSE 构建 AI 实时流式对话系统：原理剖析与代码实战

一、引言：告别等待！AI 实时对话的流式解决方案

1.1 传统 AI 对话的痛点与技术瓶颈

1.2 本文核心内容与技术栈说明

二、原理篇：SSE 为何是 AI 流式对话的最优解

2.1 SSE 核心原理：服务器单向推送的长连接技术

2.2 SSE vs WebSocket：技术选型的关键考量

2.3 AI 流式对话与 SSE 的契合点

三、实战篇：SpringBoot 整合 SSE 构建 AI 对话系统

3.1 环境准备：SpringBoot 项目搭建与依赖配置

3.2 核心组件 1：SSE 连接管理器（SSEServer）

3.2.1 连接管理的核心需求

3.2.2 关键方法实现（connect/sendMsg/close）

3.3 核心组件 2：消息类型枚举（SSEMsgType）

3.4 核心组件 3：SSE 控制器（SSEController）

3.5 业务层整合：AI 大模型流式 API 调用

3.5.1 AI 流式接口对接（以 DeepSeek 为例）

3.5.2 异步处理与异常兜底

四、前端实现：基于 EventSource 的实时消息渲染

4.1 EventSource 初始化与连接建立

4.2 流式文本渲染与交互优化

4.3 前端兼容性处理

五、测试与验证：一键运行实时对话系统

5.1 功能测试步骤

5.2 常见问题排查与解决方案

六、进阶优化：提升系统稳定性与扩展性

6.1 连接池优化与分布式适配

6.2 消息重试与幂等性保障

6.3 多模型集成与动态切换

七、总结与展望

7.1 本文核心内容总结

7.2 技术拓展方向

八、附录：完整代码仓库与资源链接

8.1 参考资料与学习链接

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具