IT评测·应用市场-qidao123.com技术社区

标题: SSE [打印本页]

---

作者: 何小豆儿在此    时间: 2025-4-1 14:33
标题: SSE
SSE

项目中遇到竞价功能，用户停留在竞价页面，服务端需要及时把竞价信息推送给每一个客户

什么是SSE

SSE是一种服务器推送技术，答应服务器及时向客户端发送数据流。它是一种轻量级的单向通讯机制，特别得当于及时性要求高的场景，如及时更新日志、及时新闻推送等。在Java中，SSE的实现依赖于Servlet 3.0及以上版本，通过使用Java的特定库和框架，可以轻松地向客户端推送及时数据。

业务上实现单向通讯有以下两种：

特性Spring WebFlux + SSESpring MVC + SseEmitter并发能力高（单线程支持数千毗连）低（受限于线程池巨细）线程占用无独占线程（EventLoop共享）每个推送占用线程池线程代码复杂度简单（内置广播机制）复杂（需手动管理毗连和线程）适用场景新项目/高并发需求旧项目/低并发需求Spring WebFlux + SSE 简单案例

1.接收器管理类

实际业务中，竞价不可能只是单一的一个物品竞价，可能同时存在多个物品竞价
ConcurrentMap dynamicSinks
一个物品竞价map就添加一条
业务结束（某一个竞价结束）调用closeSink

1. package com.dem.framework.config.sse;
3. import org.springframework.stereotype.Component;
4. import reactor.core.publisher.Flux;
5. import reactor.core.publisher.Sinks;
7. import java.util.concurrent.ConcurrentHashMap;
8. import java.util.concurrent.ConcurrentMap;
10. /**
11. * @program: dem-template-pc
12. * @description: 接收器管理类
13. * @author: lcb
14. * @create: 2025-04-01
15. **/
16. @Component
17. public class DynamicSinkManager {
18.     private final ConcurrentMap<String, Sinks.Many<String>> dynamicSinks = new ConcurrentHashMap<>();
20.     /**
21.      * 根据sinkKey获取Sink，如果不存在则创建一个
22.      * @param sinkKey Sink的唯一key
23.      * @author lcb
24.      * @date 2025/4/1
25.      * @return
26.      **/
27.     public Sinks.Many<String> getOrCreateSink(String sinkKey) {
28.         return dynamicSinks.computeIfAbsent(sinkKey,
29.                 k -> Sinks.many().multicast().directBestEffort()
30.         );
31.     }
33.     /**
34.      * 关闭指定Sink
35.      * @param sinkKey
36.      *
37.      * @author lcb
38.      * @date 2025/4/1
39.      * @return
40.      *
41.      **/
42.     public void closeSink(String sinkKey) {
43.         Sinks.Many<String> sink = dynamicSinks.remove(sinkKey);
44.         if (sink != null) {
45.             sink.tryEmitComplete();
46.         }
47.     }
49.     public Flux<String> getFlux(String sinkKey) {
50.         return getOrCreateSink(sinkKey).asFlux();
51.     }
53. }

复制代码

2.消息生产者

1. package com.dem.framework.config.sse;
3. import lombok.extern.slf4j.Slf4j;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.stereotype.Component;
6. import reactor.core.publisher.Flux;
7. import reactor.core.publisher.Sinks;
9. import java.util.List;
11. /**
12. * @program: dem-template-pc
13. * @description: sse 消息生产者
14. * @author: lcb
15. * @create: 2025-04-01
16. **/
17. @Component
18. @Slf4j
19. public class MessageProducer {
20.     @Autowired
21.     private DynamicSinkManager dynamicSinkManager;
23.     /**
24.      * 订阅 广播模式
25.      * @param sinkKey 订阅一个
26.      *
27.      * @author lcb
28.      * @date 2025/4/1 15:26
29.      * @return
30.      **/
31.     public Flux<String> asFluxShare(String sinkKey) {
32.         return dynamicSinkManager.getOrCreateSink(sinkKey).asFlux().share();
33.     }
35.     //
36.     public void sendMessage(String sinkKey,String message) {
37.         Sinks.EmitResult result = dynamicSinkManager.getOrCreateSink(sinkKey).tryEmitNext(message);
38.         // 处理发送结果
39.         if (result.isFailure()) {
40.             // 根据业务需求处理失败情况
41.             handleEmitFailure(result, message);
42.         }
43.     }
45.     // 批量发送
46.     public void sendBatch(String sinkKey,List<String> messages) {
47.         messages.forEach(msg->sendMessage(sinkKey, msg));
48.     }
50.     // 关闭消息流
51.     public void shutdown(String sinkKey) {
52.         dynamicSinkManager.getOrCreateSink(sinkKey).tryEmitComplete();
53.     }
55.     private void handleEmitFailure(Sinks.EmitResult result, String message) {
56.         // 实现你的失败处理逻辑
57.         log.error("消息发送失败，原因: {}", result);
58.     }
59. }

复制代码

3.后端测试入口

1. package com.dem.web.controller.business;
3. import com.dem.common.annotation.Anonymous;
4. import com.dem.framework.config.sse.MessageProducer;
5. import lombok.extern.slf4j.Slf4j;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.http.MediaType;
8. import org.springframework.web.bind.annotation.GetMapping;
9. import org.springframework.web.bind.annotation.PathVariable;
10. import org.springframework.web.bind.annotation.RequestMapping;
11. import org.springframework.web.bind.annotation.RestController;
12. import reactor.core.publisher.Flux;
15. /**
16. * @program: dem-template-pc
17. * @description: 测试
18. * @author: lcb
19. * @create: 2025-03-28 10:21
20. **/
22. @RestController
23. @RequestMapping("/sse")
24. @Slf4j
25. public class SSEController {
26.     @Autowired
27.     private MessageProducer messageProducer;
29.     /**
30.      * 与客户端建立链接
31.      *
32.      * @param sinkKey
33.      * @return
34.      * @author lcb
35.      * @date 2025/4/1
36.      **/
37.     @GetMapping(value = "/linkEvents/{sinkKey}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
38.     public Flux<String> linkEvents(@PathVariable String sinkKey) {
39.         return messageProducer.asFluxShare(sinkKey);
40.     }
42.     /**
43.      * 模拟添加数据
44.      *
45.      * @param sinkKey
46.      * @param str
47.      * @return
48.      * @author lcb
49.      * @date 2025/4/1 15:50
50.      **/
51.     @GetMapping(value = "/addData/{sinkKey}/{str}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
52.     public void addData(@PathVariable("sinkKey") String sinkKey, @PathVariable("str") String str) {
53.         /*
54.          * 处理业务。。。
55.          * */
56.         log.info("数据:{}", str);
57.         messageProducer.sendMessage(sinkKey, str);
58.     }
60.     /**
61.      * 模拟竞价结束
62.      *
63.      * @param sinkKey
64.      * @return
65.      * @author lcb
66.      * @date 2025/4/1 15:50
67.      **/
68.     @GetMapping(value = "/shutdown/{sinkKey}")
69.     public void guanb(@PathVariable String sinkKey) {
70.         /*
71.          * 竞价结束
72.          * */
73.         messageProducer.shutdown(sinkKey);
74.     }
75. }

复制代码

4.测试

• 前端代码
  1. const eventSource = new EventSource('http://localhost:8080/dem-vue-template-api/sse/linkEvents/dem123');
  2. eventSource.onmessage = function(event) {
  3.   console.log('Received:', event.data);
  4.   // 在这里处理接收到的数据，比如更新UI
  5. };

  复制代码
• 也可使用欣赏器直接哀求
  

其他参数

• Sinks.many():
  
  
  
  • 创建一个可以发射多个元素的 Sink (接收器)
    
  • 与 Sinks.one()(单元素)和 Sinks.empty()(无元素)相对
    
  
  
• .multicast():
  
  
  
  • 指定这是一个多播 Sink
    
  • 意味着多个订阅者可以订阅同一个源
    
  • 与 .unicast()(单播)相对
    
  
  
• .directBestEffort():
  

  由于这个业务是 竞价，后来订阅的用户只需要获取最新的数据即可
  如果后来订阅的用户需要获取之前的汗青数据可以参考
  .onBackpressureBuffer()
  .replay()
  

  
  
  
  
  • direct: 表现直接传递元素，不进行任何缓冲
    
  • bestEffort: 表现"尽力而为"的传递策略，如果卑鄙跟不上，可能会抛弃元素
    
  
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com技术社区 (https://dis.qidao123.com/)

Powered by Discuz! X3.4