devbox
木道寻08
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SSE 服务端推送实现和实际问题解决_sseemitter服务端关闭造成重连

作者：木道寻08 | 2024-07-19 06:46:16

赞

踩

sseemitter服务端关闭造成重连

实现框架springboot

很多文章都用它来和webscoket对比，就不细讲了，简单来说sse 只能server主动向client推送

并且特点是具备主动断开重连功能，这点是和webscoket我认为比较重要的区别点因为这点有实际bug,原因是sse客户端每次重连都是建立新通道，导致服务端找不到之前的通道，但是消息也会推送成功，只是报错就很烦了因为这是个性能问题（如果服务器是巨兽或低并发不考虑）

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.NoArgsConstructor;
  4.  
  5. /**
  6.  * 消息体
  7.  */
  8. @Data
  9. @AllArgsConstructor
  10. @NoArgsConstructor
  11. public class MessageVo {
  12.     /**
  13.      * 客户端id
  14.      */
  15.     private String clientId;
  16.     /**
  17.      * 传输数据体(json)
  18.      */
  19.     private String data;
  20. }

数据载体类

  1. import com.ruoyi.business.sse.main.MessageVo;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.java_websocket.client.WebSocketClient;
  4. import org.springframework.beans.factory.annotation.Value;
  5. import org.springframework.http.HttpStatus;
  6. import org.springframework.http.MediaType;
  7. import org.springframework.stereotype.Service;
  8. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
  9.  
  10. import java.io.IOException;
  11. import java.net.URI;
  12. import java.net.URISyntaxException;
  13. import java.util.HashMap;
  14. import java.util.Map;
  15. import java.util.concurrent.ConcurrentHashMap;
  16. import java.util.concurrent.atomic.AtomicBoolean;
  17. import java.util.function.Consumer;
  18.  
  19. /**
  20.  * 实现
  21.  */
  22. @Slf4j
  23. @Service
  24. public class SseEmitterServiceImpl implements SseEmitterService {
  25.  
  26.     @Value("${wss.url}")
  27.     private String wssUrl;
  28.  
  29.     /**
  30.      * 容器，保存连接，用于输出返回 ;可使用其他方法实现
  31.      */
  32.     private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
  33.  
  34.     /**
  35.      * 创建连接
  36.      *
  37.      * @param clientId 客户端ID
  38.      */
  39.     @Override
  40.     public SseEmitter createConnect(String clientId) throws URISyntaxException {
  41.  
  42.         if (sseCache.containsKey(clientId)) {
  43.             removeUser(clientId);
  44.         }
  45.         // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
  46.         SseEmitter sseEmitter = new SseEmitter(50 * 1000L);
  47.  
  48.         // 添加自己的业务
  49.  
  50.         // 注册回调
  51.         sseEmitter.onCompletion(completionCallBack(clientId,client));     // 长链接完成后回调接口(即关闭连接时调用)
  52.         sseEmitter.onTimeout(timeoutCallBack(clientId,client));        // 连接超时回调
  53.         sseEmitter.onError(errorCallBack(clientId,client));          // 推送消息异常时，回调方法
  54.         sseCache.put(clientId, sseEmitter);
  55.         System.out.println("创建新的sse连接，当前用户：{" + clientId + "}  累计用户:{" + sseCache.size() + "}");
  56.         try {
  57.             // 注册成功返回用户信息
  58.             sseEmitter.send(SseEmitter.event().id(String.valueOf(HttpStatus.OK)).data(clientId, MediaType.APPLICATION_JSON));
  59.         } catch (IOException e) {
  60.             sseEmitter.complete();
  61.             System.out.println("创建长链接异常，客户端ID:{}   异常信息:{" + clientId + "}    异常信息:{}{" + e.getMessage() + "}");
  62.         }
  63.         return sseEmitter;
  64.     }
  65.  
  66.     /**
  67.      * 关闭连接
  68.      *
  69.      * @param clientId 客户端ID
  70.      */
  71.     @Override
  72.     public void closeConnect(String clientId) {
  73.         SseEmitter sseEmitter = sseCache.get(clientId);
  74.         if (sseEmitter != null) {
  75.             sseEmitter.complete();
  76.             removeUser(clientId);
  77.         }
  78.     }
  79.  
  80.     /**
  81.      * 长链接完成后回调接口(即关闭连接时调用)
  82.      *
  83.      * @param clientId 客户端ID
  84.      **/
  85.     private Runnable completionCallBack(String clientId,SSEWebSocketClient client) {
  86.         return () -> {
  87.             System.out.println("连接关闭！！！");
  88.             closeConnect(clientId);
  89.             client.close();
  90.         };
  91.     }
  92.  
  93.     /**
  94.      * 连接超时时调用
  95.      *
  96.      * @param clientId 客户端ID
  97.      **/
  98.     private Runnable timeoutCallBack(String clientId,SSEWebSocketClient client) {
  99.         return () -> {
  100.             System.out.println("连接超时！！！");
  101.             closeConnect(clientId);
  102.             client.close();
  103.         };
  104.     }
  105.  
  106.     /**
  107.      * 推送消息异常时，回调方法
  108.      *
  109.      * @param clientId 客户端ID
  110.      **/
  111.     private Consumer<Throwable> errorCallBack(String clientId,SSEWebSocketClient client) {
  112.         return throwable -> {
  113.             System.out.println("消息异常！！！");
  114.             closeConnect(clientId);
  115.             client.close();
  116.         };
  117.     }
  118.  
  119.     /**
  120.      * 移除用户连接
  121.      *
  122.      * @param clientId 客户端ID
  123.      **/
  124.     private void removeUser(String clientId) {
  125.         sseCache.remove(clientId);
  126.         log.info("SseEmitterServiceImpl[removeUser]:移除用户：{}", clientId);
  127.     }
  128. }

业务实现类

  1. import com.alibaba.fastjson2.JSONObject;
  2. import com.ruoyi.business.sse.main.MessageVo;
  3. import lombok.SneakyThrows;
  4. import org.java_websocket.client.WebSocketClient;
  5. import org.java_websocket.handshake.ServerHandshake;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.http.HttpStatus;
  8. import org.springframework.http.MediaType;
  9. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
  10.  
  11. import javax.annotation.Resource;
  12. import java.io.IOException;
  13. import java.net.URI;
  14. import java.util.HashMap;
  15. import java.util.Map;
  16.  
  17. public class SSEWebSocketClient extends WebSocketClient {
  18.  
  19.     private String clientId;
  20.     private SseEmitter sseEmitter;
  21.  
  22.  
  23.     public String getClientId() {
  24.         return clientId;
  25.     }
  26.  
  27.     public void setClientId(String clientId) {
  28.         this.clientId = clientId;
  29.     }
  30.  
  31.     public SseEmitter getSseEmitter() {
  32.         return sseEmitter;
  33.     }
  34.  
  35.     public void setSseEmitter(SseEmitter sseEmitter) {
  36.         this.sseEmitter = sseEmitter;
  37.     }
  38.  
  39.     public SSEWebSocketClient(URI serverUri, Map<String, String> httpHeaders) {
  40.         super(serverUri, httpHeaders);
  41.     }
  42.  
  43.     @Override
  44.     public void onOpen(ServerHandshake handshakedata) {
  45.         System.out.println("新连接已打开");
  46.         super.send("{\"type\":\"connection_init\"}");
  47.         try {
  48.             Thread.sleep(700L);
  49.         } catch (InterruptedException e) {
  50.             e.printStackTrace();
  51.         }
  52.         String param = "private-search-" + clientId;
  53.         super.send("{\"id\": \"" + param + "\",\"payload\": {\"data\": \"{\\\"query\\\":\\\"subscription Subscribe($name: String!) {\\\\n  subscribe(name: $name) {\\\\n    name\\\\n    data\\\\n    __typename\\\\n  }\\\\n}\\\\n\\\",\\\"variables\\\":{\\\"name\\\":\\\"" + param + "\\\"}}\",\"extensions\": {\"authorization\": {\"host\": \"apndpknqrjhgpbhqwfwdmcx2du.appsync-api.cn-northwest-1.amazonaws.com.cn\",\"x-amz-date\": \"20240412T033202Z\",\"x-api-key\": \"da2-rduvi64eizehdotnp3a3jhokuq\",\"x-amz-user-agent\": \"aws-amplify/6.0.27 api/1 framework/2\"}}},\"type\": \"start\"}");
  54.     }
  55.  
  56.     @Override
  57.     public void onMessage(String message) {
  58.         System.out.println("接收到消息: " + message);
  59.         if (!message.contains("连接成功")) {
  60.             JSONObject result1 = JSONObject.parse(message);
  61.             if (result1.getString("type").equals("data")) {
  62.                 if (result1.getJSONObject("payload").getJSONObject("data").getJSONObject("subscribe").getJSONObject("data").getString("message").equals("streaming")) {
  63.                     MessageVo messageVo = new MessageVo(clientId, result1.getJSONObject("payload").getJSONObject("data").getJSONObject("subscribe").getString("data"));
  64.                     SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.OK))
  65.                             .data(messageVo, MediaType.APPLICATION_JSON);
  66.                     try {
  67.                         sseEmitter.send(sendData);
  68.                     } catch (IOException e) {
  69.                         e.printStackTrace();
  70.                         sseEmitter.complete();
  71.                     }
  72.                 }
  73.  
  74.                 if (result1.getJSONObject("payload").getJSONObject("data").getJSONObject("subscribe").getJSONObject("data").getString("message").equals("streaming_end")) {
  75.                     MessageVo messageVo = new MessageVo(clientId, result1.getJSONObject("payload").getJSONObject("data").getJSONObject("subscribe").getString("data"));
  76.                     SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.OK))
  77.                             .data(messageVo, MediaType.APPLICATION_JSON);
  78.                     try {
  79.                         sseEmitter.send(sendData);
  80.                         sseEmitter.complete();
  81.  
  82.                     } catch (IOException e) {
  83.                         e.printStackTrace();
  84.                         sseEmitter.complete();
  85.                     }
  86.                 }
  87.  
  88.             } else if (result1.getString("type").equals("start_ack")) {
  89.                 System.out.println("连接成功！！！");
  90.             } else {
  91.                 System.out.println(result1.toJSONString());
  92.             }
  93.         }
  94.     }
  95.  
  96.     @Override
  97.     public void onClose(int code, String reason, boolean remote) {
  98.         System.out.println("连接已关闭");
  99.     }
  100.  
  101.     @Override
  102.     public void onError(Exception ex) {
  103.         ex.printStackTrace();
  104.     }
  105.  
  106. }

接口类

  1. import com.ruoyi.business.sse.service.SseEmitterService;
  2. import com.ruoyi.common.core.domain.AjaxResult;
  3. import com.ruoyi.common.utils.StringUtils;
  4. import org.apache.http.client.methods.CloseableHttpResponse;
  5. import org.apache.http.client.methods.HttpPost;
  6. import org.apache.http.entity.StringEntity;
  7. import org.apache.http.impl.client.CloseableHttpClient;
  8. import org.apache.http.impl.client.HttpClients;
  9. import org.apache.http.util.EntityUtils;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import org.springframework.http.HttpEntity;
  12. import org.springframework.http.HttpHeaders;
  13. import org.springframework.http.MediaType;
  14. import org.springframework.http.ResponseEntity;
  15. import org.springframework.web.bind.annotation.*;
  16. import org.springframework.web.client.RestTemplate;
  17. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
  18.  
  19. import javax.annotation.Resource;
  20. import java.io.IOException;
  21. import java.net.URISyntaxException;
  22. import java.util.Map;
  23.  
  24. /**
  25.  * SSE长链接
  26.  */
  27. @RestController
  28. @RequestMapping("/sse")
  29. public class SseEmitterController {
  30.  
  31.     @Resource
  32.     private SseEmitterService sseEmitterService;
  33.  
  34.     @Value("${appsync.url}")
  35.     private String url;
  36.  
  37.     @CrossOrigin
  38.     @GetMapping("/createConnect/{clientId}")
  39.     public SseEmitter createConnect(@PathVariable("clientId") String clientId) throws URISyntaxException {
  40.         return sseEmitterService.createConnect(clientId);
  41.     }
  42.  
  43.     @CrossOrigin
  44.     @GetMapping("/closeConnect")
  45.     public void closeConnect(@RequestParam(required = true) String clientId) {
  46.         sseEmitterService.closeConnect(clientId);
  47.     }
  48.  
  49.     @CrossOrigin
  50.     @PostMapping("/sendAjaxMessage")
  51.     public AjaxResult sendMessage(@RequestBody Map map) {
  52.  
  53.         // map参数体校验
  54.         if (map == null) {
  55.             return AjaxResult.error(601, "请求接口需要传递正确的JSON数据");
  56.         }
  57.         // query 参数校验
  58.         String query = (String) map.get("query");
  59.         if (StringUtils.isEmpty((String) map.get("query"))) {
  60.             return AjaxResult.error(601, " query 不能为空");
  61.         }
  62.  
  63.         // query 参数校验
  64.         String messageId = (String) map.get("messageId");
  65.         if (StringUtils.isEmpty((String) map.get("messageId"))) {
  66.             return AjaxResult.error(601, " messageId 不能为空");
  67.         }
  68.  
  69.         // connectionId 参数校验
  70.         String connectionId = (String) map.get("connectionId");
  71.         if (StringUtils.isEmpty((String) map.get("connectionId"))) {
  72.             return AjaxResult.error(601, " connectionId 不能为空");
  73.         }
  74.  
  75.  
  76.         HttpHeaders headers = new HttpHeaders();
  77.         headers.setContentType(MediaType.APPLICATION_JSON);
  78.         headers.add("connectionId", connectionId);
  79.  
  80.         String indexName = "smart_search_qa_test";
  81.         String prompt = "";
  82.  
  83.  
  84.         String jsonBody = "{\n" +
  85.                 "    \"action\": \"search\",\n" +
  86.                 "    \"configs\": {\n" +
  87.                 "        \"name\": \"demo\",\n" +
  88.                 "        \"searchEngine\": \"opensearch\",\n" +
  89.                 "        \"llmData\": {\n" +
  90.                 "            \"strategyName\": \"unimportant\",\n" +
  91.                 "            \"type\": \"sagemaker_endpoint\",\n" +
  92.                 "            \"embeddingEndpoint\": \"huggingface-inference-eb\",\n" +
  93.                 "            \"modelType\": \"non_llama2\",\n" +
  94.                 "            \"recordId\": \"dji-inference-chatglm3-6b-62277\",\n" +
  95.                 "            \"sagemakerEndpoint\": \"dji-inference-chatglm3-6b\",\n" +
  96.                 "            \"streaming\": true\n" +
  97.                 "        },\n" +
  98.                 "        \"role\": \"\",\n" +
  99.                 "        \"language\": \"chinese\",\n" +
  100.                 "        \"taskDefinition\": \"\",\n" +
  101.                 "        \"outputFormat\": \"stream\",\n" +
  102.                 "        \"isCheckedGenerateReport\": false,\n" +
  103.                 "        \"isCheckedContext\": false,\n" +
  104.                 "        \"isCheckedKnowledgeBase\": true,\n" +
  105.                 "        \"indexName\": \"" + indexName + "\",\n" +
  106.                 "        \"topK\": 2,\n" +
  107.                 "        \"searchMethod\": \"vector\",\n" +
  108.                 "        \"txtDocsNum\": 0,\n" +
  109.                 "        \"vecDocsScoreThresholds\": 0,\n" +
  110.                 "        \"txtDocsScoreThresholds\": 0,\n" +
  111.                 "        \"isCheckedScoreQA\": true,\n" +
  112.                 "        \"isCheckedScoreQD\": true,\n" +
  113.                 "        \"isCheckedScoreAD\": true,\n" +
  114.                 "        \"contextRounds\": 2,\n" +
  115.                 "        \"isCheckedEditPrompt\": false,\n" +
  116.                 "        \"prompt\": \"" + prompt + "\",\n" +
  117.                 "        \"tokenContentCheck\": \"\",\n" +
  118.                 "        \"responseIfNoDocsFound\": \"Cannot find the answer\"\n" +
  119.                 "    },\n" +
  120.                 "    \"query\": \"" + query + "\"," +
  121.                 "    \"messageId\": \"" + messageId + "\"" +
  122.                 "}";
  123.         HttpEntity<String> entity = new HttpEntity<>(jsonBody, headers);
  124.  
  125.         try {
  126.             RestTemplate restTemplate = new RestTemplate();
  127.             ResponseEntity<String> response = restTemplate.postForEntity(url, entity, String.class);
  128.             return AjaxResult.success("操作成功");
  129.         } catch (Exception e) {
  130.             e.printStackTrace();
  131.             return AjaxResult.error("操作失败");
  132.         }
  133.     }
  134.  
  135.  
  136. }

控制类

MessageVo messageVo = new MessageVo(clientId,"你想推送给页面的消息这段代码写在自己的也业务");
                    SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.OK))
                            .data(messageVo, MediaType.APPLICATION_JSON);
                    try {
                        sseEmitter.send(sendData);
                        sseEmitter.complete();

                    } catch (IOException e) {
                        e.printStackTrace();
                        sseEmitter.complete();
                    }

有一点很抱歉接口封装的发送消息代码是个未完成的代码一定要用上者SseEmitter对象要注意甄别这个类似通道的对象

还有一点注意如果发送消息完成之后一定要complete() 不然会出现异常这个问题资料都避而不谈网上一直强调sse使用简单但是实现并且维护一个高压sse很复杂异常百出资料稀少；

还有一点注意sseEmitter.complete();要在你设置超时时间之内；

其实可以设置0L永不超时但是对服务器压力巨大，因为用户量也大，也根据实际情况(我这边是aws服务产品二开请求或连接不能超过1分钟，超过主动断开)

结语：虽然我实现了webscoket 和 sse 两个版本但是我还是最后采用了sse ,没有别的新技术还是要支持即使会有问题但是我也要多发现暴露出来别让后人踩坑，注意代码要结合自己的业务符合java语言整改，我也是找别人改的。

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号