Java调用SSE流式接口并流式返回给前端实现打字输出效果_Java

1.sse概述

1.1 什么是是sse

server-sent events (sse)sse是一种简单的事件推送技术，它允许服务器异步地向客户端发送更新，而无需客户端显式请求这些更新。这对于实时应用程序非常有用，例如股票价格更新、消息通知等。sse基于http协议，使用一个持久的http连接来维持客户端和服务端之间的通信。

2.2 与长链接(long polling)的区别

server-sent events (sse) 和长链接(long polling)都是实现服务器向客户端推送数据的技术，但它们之间存在一些关键区别。下面我将详细解释这两种技术的不同之处：

长链接(long polling)

长链接是一种实现服务器推送数据到客户端的技术，它基于http请求/响应模型。在这种模式下，客户端发起一个http请求，服务器在没有数据可发送的情况下会保持连接打开，直到有数据可发送或者超时。一旦服务器有数据要发送，它就会响应客户端的请求，并关闭连接。客户端接收到数据后立即重新发起一个新的请求，从而保持与服务器的“长链接”。

特点：

客户端主动发起请求：客户端需要不断地向服务器发起请求以获取数据。
服务器被动响应：服务器只在客户端请求时才发送数据。
连接短暂：虽然每个连接可能会持续一段时间，但每次请求结束后连接会被关闭。
实现简单：易于用现有http技术实现。
兼容性好：几乎所有浏览器都支持http请求/响应模型。

server-sent events (sse)

server-sent events 是一种更为现代的技术，用于实现服务器向客户端的单向数据推送。sse基于http协议，但使用了一个持久的http连接来维持客户端和服务端之间的通信。服务器可以主动向客户端发送数据，而不需要等待客户端的请求。

特点：

服务器主动推送：服务器可以主动向客户端发送数据，而不需要客户端发起请求。
持久连接：客户端和服务端之间建立了一个持久的连接，直到客户端或服务器关闭该连接。
格式特定：sse使用特定的格式来发送数据，包括data:字段和空行作为分隔符。
资源效率高：由于连接是持久的，因此减少了建立连接的开销。
实现复杂度适中：虽然比长链接稍微复杂，但现代浏览器和服务器框架提供了良好的支持。

比较

实时性：sse提供更好的实时性，因为它不需要客户端不断发起请求。
性能：sse在性能上通常优于长链接，因为它避免了重复建立连接的开销。
实现复杂度：sse需要客户端和服务端双方的支持，而长链接可以更容易地在现有的http基础设施上实现。
兼容性：sse在现代浏览器中得到了广泛支持，但对于一些旧版浏览器可能不适用；长链接则具有更好的向后兼容性。

总结

选择哪种技术取决于你的具体需求。如果你的应用需要较低延迟的数据推送，并且可以依赖现代浏览器和服务器环境，那么sse是一个不错的选择。如果你需要更广泛的浏览器兼容性，并且对实时性要求不是特别高，那么长链接可能更适合你。

2.通过okhttp调用sse流式接口并流式返回给前端

环境要求

spring framework 5.0
jdk1.8

使用okhttp相关依赖

<dependency>
    <groupid>com.squareup.okhttp3</groupid>
    <artifactid>okhttp</artifactid>
    <version>4.2.0</version>
</dependency>
<dependency>
    <groupid>com.squareup.okhttp3</groupid>
    <artifactid>okhttp-sse</artifactid>
    <version>4.2.0</version>
</dependency>

示例

@getmapping(value = "/test1", produces = mediatype.text_event_stream_value)
    public sseemitter ssetest1() {
        sseemitter sseemitter = new sseemitter();
        string prompt = "";
        string url = "";
        formbody formbody = new formbody.builder().add("prompt", prompt).build();
        request request = new request.builder().url(url).post(formbody).build();
        // 使用eventsourcelistener处理来自服务器的sse事件
        eventsourcelistener listener = new eventsourcelistener() {
            @override
            public void onopen(@notnull eventsource eventsource, @notnull response response) {
                log.info("connection opened.");
            }
            @override
            public void onclosed(@notnull eventsource eventsource) {
                log.info("connection closed.");
                sseemitter.complete();
            }
            @override
            public void onevent(@notnull eventsource eventsource, @nullable string id, @nullable string type, @notnull string data) {
                try {
                    jsonobject jsonobject = jsonutil.parseobj(data);
                    string event = jsonobject.getstr("event");
                    if ("message".equals(event)) {
                        sseemitter.send(jsonobject.getstr("answer"));
                    }
                } catch (exception e) {
                    log.error("推送数据失败", e);
                }
            }
            @override
            public void onfailure(@notnull eventsource eventsource, @nullable throwable t, @nullable response response) {
                log.error("connection failed.", t);
                sseemitter.completewitherror(t);
            }
        };
        okhttpclient client = new okhttpclient.builder().connecttimeout(10, timeunit.seconds).writetimeout(50, timeunit.seconds).readtimeout(10, timeunit.minutes).build();
        eventsource.factory factory = eventsources.createfactory(client);
        factory.neweventsource(request, listener);
        return sseemitter;
    }

注意

该接口需为get请求，contenttype为 text/event-stream
sseemitter 是spring framework 5.0引入的一个新特性，用于简化server-sent events (sse) 的实现。它提供了一种简单的方式来发送事件数据到客户端，特别适用于构建实时数据推送的应用程序。

3. 如果spring framework 低于5.0，可使用servlet 3.0进行流式返回

使用asynccontext：servlet 3.0 引入了异步支持，允许servlet在不同的线程中处理请求。你可以使用asynccontext来启动一个异步线程，在该线程中发送sse事件。

配置async-supported使用asynccontext前需配置async-supported

async-supported元素用于指定servlet是否支持异步处理。这个配置通常是在部署描述符 web.xml 文件中进行设置的。

配置示例

<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
         xsi:schemalocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">

    <servlet>
        <servlet-name>myservlet</servlet-name>
        <servlet-class>com.example.myservlet</servlet-class>
        <async-supported>true</async-supported>
    </servlet>

    <servlet-mapping>
        <servlet-name>myservlet</servlet-name>
        <url-pattern>/myservlet</url-pattern>
    </servlet-mapping>

</web-app>

后端代码示例

@getmapping("/test3")
    public void ssetest3(httpservletrequest req, httpservletresponse resp) {
        resp.setcontenttype("text/event-stream");
        resp.setcharacterencoding("utf-8");
        resp.setheader("cache-control", "no-cache");
        resp.setheader("connection", "keep-alive");

        try {
            //
            asynccontext asynccontext = req.startasync(req, resp);
            asynccontext.settimeout(10 * 60 * 1000);
            printwriter writer = asynccontext.getresponse().getwriter();

            string prompt = "";
            string url = "";
            formbody formbody = new formbody.builder().add("prompt", prompt).build();
            request request = new request.builder().url(url).post(formbody).build();
            // 使用eventsourcelistener处理来自服务器的sse事件
            eventsourcelistener listener = new eventsourcelistener() {
                @override
                public void onopen(@notnull eventsource eventsource, @notnull response response) {
                    log.info("connection opened.");
                }

                @override
                public void onclosed(@notnull eventsource eventsource) {
                    log.info("connection closed.");
                    writer.write("data: __stop__\n\n");
                    writer.flush();
                    asynccontext.complete();
                }

                @override
                public void onevent(@notnull eventsource eventsource, @nullable string id, @nullable string type, @notnull string data) {
                    try {
                        jsonobject jsonobject = jsonutil.parseobj(data);
                        string event = jsonobject.getstr("event");
                        if ("message".equals(event)) {
                            string answer = jsonobject.getstr("answer");
                            log.info("message: {}", answer);
                            writer.write("data: " + answer + "\n\n");
                            writer.flush();
                        }
                    } catch (exception e) {
                        log.error("推送数据失败", e);
                    }
                }

                @override
                public void onfailure(@notnull eventsource eventsource, @nullable throwable t, @nullable response response) {
                    log.error("connection failed.", t);
                    asynccontext.complete();
                }
            };
            okhttpclient client = new okhttpclient.builder().connecttimeout(10, timeunit.seconds).writetimeout(50, timeunit.seconds).readtimeout(10, timeunit.minutes).build();
            eventsource.factory factory = eventsources.createfactory(client);
            factory.neweventsource(request, listener);
        } catch (ioexception e) {
            e.printstacktrace();
        } finally {

        }
    }

注意返回数据格式:

// 以data: 开头  /n/n结束
"data: xxxxx /n/n"

4. 前端调用sse接口

方式1 使用javascript的 eventsource api

<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <title>sse example</title>
</head>
<body>
    <div id="events"></div>
    <script>
        const source = new eventsource('/sse');

        source.onmessage = function(event) {
            const data = json.parse(event.data);
            // 约定一个结束标识
            if(data == '__stop__') {
            	source.close()
            	return
            }
            document.getelementbyid('events').innerhtml += `<p>${data.message}</p>`;
        };

        source.onerror = function(error) {
            console.error('error occurred:', error);
            source.close();
        };
    </script>
</body>
</html>

注意

后端返回需返回完整消息的对象（包括换行符），例：{“data”: “哈哈哈/n/n”}，如果后端将data取出，则会导致换行符丢失！
eventsource 只支持get请求，如果请求参数过长会导致调用失败！

方式2 使用 fetcheventsource 插件

安装插件

npm install --save @microsoft/fetch-event-source

简单示例

// 导入依赖
import { fetcheventsource } from '@microsoft/fetch-event-source';
 
send() {
	const vm = this;
	const ctrlabout = new abortcontroller();
	const { signal } = ctrlabout;
	fetcheventsource(url, {
	  method: 'post',
	  headers: {
	    "content-type": 'application/json',
	    "accept": 'text/event-stream'
	  },
	  body: json.stringify(data),
	  signal: ctrl.signal, // abortsignal
	  onmessage(event) {
	     console.info(event.data);
	     // 在这里操作流式数据
	     const message = json.parse(event.data)
	     vm.content += message.data
	  },
	  onclose(e) {
	     // 关闭流
	     // 中断流式返回
	     ctrl.abort()
	  }
	  onerror(error) {
	    // 返回流报错
		console.info(error);
		// 中断流式返回
		ctrl.abort()
		throw err // 直接抛出错误，避免反复调用
	  }
	})
}

注意

传参时需注意参数类型为json字符串

5. 使用原生的http调用sse流式接口

示例

@getmapping("/test2")
    public void ssetest2() {
        string urladdr = "";
        bufferedreader reader = null;
        try {
            url url = new url(urladdr);
            // 建立链接
            httpurlconnection connection = (httpurlconnection) url.openconnection();
            connection.setrequestmethod("post");
            connection.setrequestproperty("accept", "text/event-stream");
            connection.setrequestproperty("content-type", "application/json; charset=utf-8");
            connection.setrequestproperty("cache-control", "no-cache");
            connection.setrequestproperty("connection", "keep-alive");
            // 允许输入和输出
            connection.setdoinput(true);
            connection.setdooutput(true);
            // 设置超时为0，表示无限制
            connection.setconnecttimeout(0);
            connection.setreadtimeout(0);
            // 传参
            string params = "prompt=哈哈哈哈";
            // 写入post数据
            dataoutputstream out = new dataoutputstream(connection.getoutputstream());
            out.write(params.getbytes(standardcharsets.utf_8));
            out.flush();
            out.close();

            // 读取sse事件
            reader = new bufferedreader(new inputstreamreader(connection.getinputstream(), standardcharsets.utf_8));
            stringbuilder eventbuilder = new stringbuilder();
            string line;
            while ((line = reader.readline()) != null) {
                system.out.println(line);
            }
            reader.close();
            // 断开链接
            connection.disconnect();
        } catch (exception e) {
            e.printstacktrace();
        } finally {
            ioutil.close(reader);
        }
    }