WebSocket vs SSE: 实时数据推送到前端的选择与实现(详细)

简介: WebSocket vs SSE: 实时数据推送到前端的选择与实现(详细)


在现代 Web 应用程序中,实时数据推送给前端变得越来越重要。无论是实时聊天、实时通知还是仪表板上的实时更新,都需要一种有效的方式来将数据推送给前端。本文将介绍两种常用的实现方法:WebSocket 和 Server-Sent Events(SSE),并提供详细的实现步骤。

二者对比

WebSocket 和 Server-Sent Events (SSE) 都是用于实现实时数据推送的技术,但它们在设计、用途和实现上有一些重要的区别。让我们详细比较这两种技术。

WebSocket:

  1. 双向通信
  • WebSocket 允许双向通信,客户端和服务器都可以在任何时候向对方发送数据。
  • 这使得 WebSocket 非常适用于需要双向交互的应用,如在线聊天、多人协作工具等。
  1. 持久连接
  • WebSocket 建立持久连接,客户端和服务器之间的连接保持打开状态。
  • 这减少了与建立和关闭连接相关的开销,适用于频繁的数据交换。
  1. 低延迟
  • 由于持久连接,WebSocket 可以实现低延迟的实时数据传输,适用于需要快速响应的应用。
  1. 复杂性
  • 实现 WebSocket 可能相对复杂,需要更多的服务器资源和额外的协议处理。
  1. 跨域通信
  • WebSocket 通常需要配置服务器以允许跨域通信,因为它们使用自定义协议。
  1. 浏览器支持
  • WebSocket 在现代浏览器中得到广泛支持。

Server-Sent Events (SSE):

  1. 单向通信
  • SSE 是一种单向通信,只允许服务器向客户端发送数据。客户端无法向服务器发送数据。
  1. HTTP 协议
  • SSE 建立在 HTTP 协议之上,使用标准 HTTP 请求和响应。
  • 这使得 SSE 更容易部署,因为它与现有的 HTTP 基础设施兼容。
  1. 简单性
  • SSE 的实现相对简单,服务器和客户端都不需要太多复杂的逻辑。
  1. 无需专用库
  • SSE 不需要额外的库或协议处理,客户端可以使用浏览器的原生 EventSource API 来接收数据。
  1. 跨域通信
  • SSE 支持跨域通信,可以通过 CORS(跨域资源共享)机制进行配置。
  1. 浏览器支持
  • SSE 在现代浏览器中也得到广泛支持,但与 WebSocket 相比,它的历史要长一些。

选择 WebSocket 还是 SSE:

  • WebSocket 适用于需要双向通信和低延迟的场景,例如在线游戏、实时聊天应用等。
  • SSE 适用于单向服务器到客户端的实时数据推送,例如新闻更新、实时股票报价、天气预报等,特别是当你希望使用现有的 HTTP 基础设施时。
  • 在某些情况下,你甚至可以同时使用 WebSocket 和 SSE,根据不同的需求选择合适的技术。

无论选择哪种技术,都需要考虑你的应用程序的具体需求和复杂性。WebSocket 提供了更多的灵活性和功能,而 SSE 更加简单和易于部署。最终的选择取决于你的项目目标和资源。

Websocket 实现

使用原生 WebSocket API:

  1. 简单性
  • Spring Boot 提供了对原生 WebSocket API 的支持,使得创建 WebSocket 应用相对简单。
  • 开发人员可以直接使用 Java 标准库中的 WebSocket 相关类来处理 WebSocket 通信。
  1. 依赖
  • 原生 WebSocket 不需要额外的依赖,因为 WebSocket API 已经包含在 Java 标准库中。
  1. 性能
  • 原生 WebSocket API 在性能方面表现良好,适用于大多数中小型应用。
  1. 生态系统
  • 使用原生 WebSocket 可以更容易地集成到现有的 Spring Boot 生态系统中,例如 Spring Security 等。
  1. 简单应用
  • 当你需要创建相对简单的 WebSocket 应用时,原生 WebSocket 是一个不错的选择。

使用 Netty 创建 WebSocket:

  1. 灵活性
  • Netty 是一个高度可定制的异步事件驱动框架,它可以用于创建各种网络应用,包括 WebSocket。
  • Netty 提供了更多的灵活性和自定义选项,适用于复杂的 WebSocket 应用。
  1. 性能
  • Netty 以其高性能和低延迟而闻名,适用于需要处理大量并发连接的应用。
  1. 协议支持
  • Netty 支持多种协议,不仅限于 WebSocket。这意味着你可以在同一个应用程序中处理多种网络通信需求。
  1. 集成
  • 尽管 Netty 可以集成到 Spring Boot 中,但其集成可能需要更多的配置和代码。
  1. 复杂应用
  • 当你需要处理复杂的 WebSocket 场景,如高并发、自定义协议、复杂的消息处理等时,使用 Netty 是更好的选择。

总结和选择:

选择原生 WebSocket 还是使用 Netty 创建 WebSocket 应取决于你的项目需求和复杂性:

  • 如果你的应用相对简单,对性能要求不是很高,可以考虑使用原生 WebSocket API,它更容易上手并且不需要额外的依赖。
  • 如果你的应用需要处理高并发、复杂的协议、自定义消息处理或需要最大程度的性能和灵活性,那么使用 Netty 创建 WebSocket 可能更合适。Netty 为你提供了更多的控制权和自定义选项。

无论你选择哪种方法,Spring Boot 都提供了良好的支持,使得在应用中集成 WebSocket 变得相对容易。因此,你可以根据具体的项目需求来选择适合你的方法。

Netty 实现 Websocket

  • 添加 maven 坐标
<!-- netty -->  
  <dependency>  
      <groupId>io.netty</groupId>  
      <artifactId>netty-common</artifactId>  
      <version>4.1.79.Final</version>  
  </dependency>
  • 创建 NettyWebsocketServer
package com.todoitbo.baseSpringbootDasmart.netty.server;  
  import com.todoitbo.baseSpringbootDasmart.netty.handler.HeartbeatHandler;  
  import com.todoitbo.baseSpringbootDasmart.netty.handler.WebSocketHandler;  
  import io.netty.bootstrap.ServerBootstrap;  
  import io.netty.channel.*;  
  import io.netty.channel.nio.NioEventLoopGroup;  
  import io.netty.channel.socket.SocketChannel;  
  import io.netty.channel.socket.nio.NioServerSocketChannel;  
  import io.netty.handler.codec.http.HttpObjectAggregator;  
  import io.netty.handler.codec.http.HttpServerCodec;  
  import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;  
  import io.netty.handler.stream.ChunkedWriteHandler;  
  import io.netty.handler.timeout.IdleStateHandler;  
  import io.netty.handler.traffic.ChannelTrafficShapingHandler;  
  
  /**  
   * @author xiaobo  
   * @date 2023/9/5  
   */
  public class NettyWebsocketServer {  
     private final int port;  
  
     public NettyWebsocketServer(int port) {  
       this.port = port;  
     }  
  
     public void run() throws Exception {  
       EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 创建用于接受客户端连接的 boss 线程池  
       EventLoopGroup workerGroup = new NioEventLoopGroup(); // 创建用于处理客户端请求的 worker 线程池  
  
       try {  
         ServerBootstrap b = new ServerBootstrap();  
         b.group(bossGroup, workerGroup)  
           .channel(NioServerSocketChannel.class)  
           .childHandler(new ChannelInitializer<SocketChannel>() {  
             @Override  
             public void initChannel(SocketChannel ch) throws Exception {  
               ChannelTrafficShapingHandler trafficShapingHandler = new ChannelTrafficShapingHandler(  
                 1, // 读取速率限制(字节/秒)  
                 1, // 写入速率限制(字节/秒)  
                 1, // 流量检查时间间隔(毫秒)  
                 1 // 最大允许的时间窗口(毫秒)  
               );  
               ChannelPipeline pipeline = ch.pipeline();  
               // 添加心跳检测处理器,3秒内没有读写事件将触发 IdleStateEvent,下面的顺序错了也会出现问题的  
               pipeline.addLast(new IdleStateHandler(30, 0, 0));  
               pipeline.addLast(new HeartbeatHandler());  
               pipeline.addLast(new HttpServerCodec()); // 处理 HTTP 请求  
               pipeline.addLast(new ChunkedWriteHandler()); // 写大数据流的处理器  
               pipeline.addLast(new HttpObjectAggregator(8192)); // 将 HTTP 消息聚合为 FullHttpRequest 或 FullHttpResponse  
               // pipeline.addLast(new WebSocketFrameAggregator(8192)); // 将 HTTP 消息聚合为 FullHttpRequest 或 FullHttpResponse  
               // pipeline.addLast(new WebSocketServerCompressionHandler()); // 消息压缩  
               pipeline.addLast(new WebSocketHandler()); // 自定义 WebSocket 处理器  
               pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10)); // 处理 WebSocket 升级握手和数据帧处理  
             }  
           })  
           .option(ChannelOption.SO_BACKLOG, 128)          // 设置服务器接受队列大小  
           .childOption(ChannelOption.SO_KEEPALIVE, true); // 开启 TCP 连接的 Keep-Alive 功能  
  
         // Bind and start to accept incoming connections.  
         System.out.println("TCP server started successfully");  
         ChannelFuture f = b.bind(port).sync(); // 绑定端口并等待绑定完成  
  
         // Wait until the server socket is closed.  
         // In this example, this does not happen, but you can do that to gracefully            // shut down your server.            f.channel().closeFuture().sync(); // 阻塞直到服务器关闭  
  
       } finally {  
         // 优雅地关闭线程池  
         workerGroup.shutdownGracefully();  
         bossGroup.shutdownGracefully();  
       }  
     }  
 }

这里需要注意一下,pipeline.addLast的顺序不一致可能会导致程序报错,运行时

  • 创建心跳 handle
package com.todoitbo.baseSpringbootDasmart.netty.handler;  
 
import io.netty.channel.ChannelHandlerContext;  
import io.netty.channel.ChannelInboundHandlerAdapter;  
import io.netty.handler.timeout.IdleState;  
import io.netty.handler.timeout.IdleStateEvent;  
public class HeartbeatHandler extends ChannelInboundHandlerAdapter {  
   int readTimeOut = 0;  
   @Override  
   public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {  
     IdleStateEvent event = (IdleStateEvent) evt;  
     if(event.state() == IdleState.READER_IDLE){  
       readTimeOut++;  
     }  
     if(readTimeOut >= 3){  
       System.out.println("超时超过3次,断开连接");  
       ctx.close();  
     }  
   }  
}
  • 创建WebSocketHandler
package com.todoitbo.baseSpringbootDasmart.netty.handler;  
 
 import cn.hutool.core.collection.CollectionUtil;  
 import com.todoitbo.baseSpringbootDasmart.netty.NamedChannelGroup;  
 import io.netty.buffer.ByteBuf;  
 import io.netty.buffer.Unpooled;  
 import io.netty.channel.*;  
 import io.netty.handler.codec.http.*;  
 import io.netty.handler.codec.http.websocketx.*;  
 import io.netty.util.AttributeKey;  
 import io.netty.util.CharsetUtil;  
 import lombok.extern.slf4j.Slf4j;  
 
 import java.util.HashMap;  
 import java.util.List;  
 import java.util.Map;  
 
 /**  
  * @author xiaobo  
  */
 @Slf4j  
 public class WebSocketHandler extends SimpleChannelInboundHandler<Object> {  
 
   private WebSocketServerHandshaker handshaker;  
 
   public static final AttributeKey<String> USER_ID_KEY = AttributeKey.valueOf("userId");  
   public static final AttributeKey<String> GROUP_ID_KEY = AttributeKey.valueOf("groupId");  
 
   private static final Map<Channel, String> WORK_CHANNEL_MAP = new HashMap<Channel,String>();  
 
   @Override  
   public void channelActive(ChannelHandlerContext ctx) throws Exception {  
     log.info("与客户端建立连接,通道开启!");  
     // 添加到channelGroup通道组(广播)  
     // 之后可以根据ip来进行分组  
     NamedChannelGroup.getChannelGroup("default").add(ctx.channel());  
   }  
   @Override  
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {  
     log.info("与客户端断开连接,通道关闭!");  
     // 从channelGroup通道组(广播)中删除  
     // 之后可以根据ip来进行分组  
     Channel channel = ctx.channel();  
     NamedChannelGroup.getChannelGroup("default").remove(channel);  
     WORK_CHANNEL_MAP.remove(channel);  
   }  
 
   public boolean userAuthentication(ChannelHandlerContext ctx,FullHttpRequest req) {  
     // 提取URI参数  
     QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());  
     Map<String, List<String>> parameters = queryStringDecoder.parameters();  
     // 根据参数进行处理  
     List<String> userId = parameters.get("userId");  
     List<String> groupId = parameters.get("groupId");  
     if (CollectionUtil.isNotEmpty(userId) && CollectionUtil.isNotEmpty(groupId)) {  
       ctx.channel().attr(USER_ID_KEY).set(userId.get(0));  
       ctx.channel().attr(GROUP_ID_KEY).set(groupId.get(0));  
       return true;  
     }else {  
       return false;  
     }  
 
   }  
 
 
   private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {  
     // 检查是否升级到WebSocket  
     if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {  
       // 如果不是WebSocket协议的握手请求,返回400 Bad Request响应  
       sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));  
       return;  
     }  
 
     // 构建握手响应  
     WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(  
       getWebSocketLocation(req), null, true);  
     handshaker = wsFactory.newHandshaker(req);  
 
     if (handshaker == null) {  
       // 如果不支持WebSocket版本,返回HTTP 426 Upgrade Required响应  
       WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());  
     } else {  
       handshaker.handshake(ctx.channel(), req);  
       // 进行WebSocket握手  
       // 在认证成功后,设置用户ID到Channel属性中  
       boolean authentication = userAuthentication(ctx,req);// 这里需要实现用户认证逻辑  
       if (!authentication) {  
         // 用户认证失败,可能需要关闭连接或发送认证失败消息  
         // 1. 关闭连接:  
         ctx.close();  
         // 2. 发送认证失败消息给客户端:  
         String failureMessage = "认证失败,请提供有效的身份验证信息。";  
         ctx.writeAndFlush(failureMessage);  
         return;  
       }  
       // 其他逻辑...  
       WORK_CHANNEL_MAP.put(ctx.channel(), ctx.channel().attr(GROUP_ID_KEY).get());  
     }  
   }  
 
   private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {  
     // 发送HTTP响应  
     if (res.status().code() != 200) {  
       ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);  
       res.content().writeBytes(buf);  
       buf.release();  
       HttpUtil.setContentLength(res, res.content().readableBytes());  
     }  
 
     ChannelFuture future = ctx.channel().writeAndFlush(res);  
     if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {  
       future.addListener(ChannelFutureListener.CLOSE);  
     }  
   }  
 
   private String getWebSocketLocation(FullHttpRequest req) {  
     return "ws://" + req.headers().get(HttpHeaderNames.HOST) + req.uri();  
   }  
 
   private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {  
     // 处理WebSocket消息,可以根据实际需求进行处理  
     if (frame instanceof TextWebSocketFrame) {  
       // 处理文本消息  
       String text = ((TextWebSocketFrame) frame).text();  
       System.out.println("Received message: " + text);  
 
       // 可以在这里处理WebSocket消息并发送响应  
       // ...  
     } else if (frame instanceof BinaryWebSocketFrame) {  
       // 处理二进制WebSocket消息  
       // ...  
       System.out.println("123");  
     } else if (frame instanceof CloseWebSocketFrame) {  
       // 处理WebSocket关闭请求  
       handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());  
     } else if (frame instanceof PingWebSocketFrame) {  
       // 处理WebSocket Ping消息  
       System.out.println("cs");  
       ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));  
     }  
   }  
 
   @Override  
   protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {  
     if (msg instanceof FullHttpRequest) {  
       // 处理HTTP握手请求  
       handleHttpRequest(ctx, (FullHttpRequest) msg);  
     } else if (msg instanceof WebSocketFrame) {  
       // 处理WebSocket消息  
       handleWebSocketFrame(ctx, (WebSocketFrame) msg);  
     }  
   }  
 
   @Override  
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {  
     // 发生异常时的处理  
     log.error(cause.getMessage());  
     ctx.close();  
   }  
 
}
  • 创建NamedChannelGroup
package com.todoitbo.baseSpringbootDasmart.netty;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class NamedChannelGroup{
    private String groupName;
    public static Map<String,ChannelGroup> channelGroupMap = new ConcurrentHashMap<>();
    static {
        channelGroupMap.put("default", new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
    }
    public static void setGroupName(String groupName){
        channelGroupMap.put(groupName, new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
    }
    public static ChannelGroup getChannelGroup(String groupName){
        return channelGroupMap.get(groupName);
    }
}

Server-Sent Events (SSE)实现

创建DataManager

package com.todoitbo.baseSpringbootDasmart.sse;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
 * 数据管理器用于管理Server-Sent Events (SSE) 的订阅和数据推送。
 * @author xiaobo
 */
@Component
public class DataManager {
    private final Map<String, List<SseEmitter>> dataEmitters = new HashMap<>();
    /**
     * 订阅特定数据类型的SSE连接。
     *
     * @param dataType 要订阅的数据类型
     * @param emitter  SSE连接
     */
    public void subscribe(String dataType, SseEmitter emitter) {
        dataEmitters.computeIfAbsent(dataType, k -> new ArrayList<>()).add(emitter);
        emitter.onCompletion(() -> removeEmitter(dataType, emitter));
        emitter.onTimeout(() -> removeEmitter(dataType, emitter));
    }
    /**
     * 推送特定数据类型的数据给所有已订阅的连接。
     *
     * @param dataType 要推送的数据类型
     * @param data     要推送的数据
     */
    public void pushData(String dataType, String data) {
        List<SseEmitter> emitters = dataEmitters.getOrDefault(dataType, new ArrayList<>());
        emitters.forEach(emitter -> {
            try {
                emitter.send(SseEmitter.event().data(data, MediaType.TEXT_PLAIN));
            } catch (IOException e) {
                removeEmitter(dataType, emitter);
            }
        });
    }
    private void removeEmitter(String dataType, SseEmitter emitter) {
        List<SseEmitter> emitters = dataEmitters.get(dataType);
        if (emitters != null) {
            emitters.remove(emitter);
        }
    }
}

接口实现

package com.todoitbo.baseSpringbootDasmart.controller;  
  
import com.todoitbo.baseSpringbootDasmart.sse.DataManager;  
import org.springframework.http.MediaType;  
import org.springframework.http.ResponseEntity;  
import org.springframework.web.bind.annotation.GetMapping;  
import org.springframework.web.bind.annotation.PathVariable;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RestController;  
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;  
  
import javax.annotation.Resource;  
  
/**  
 * @author xiaobo  
 */
@RestController  
@RequestMapping("/environment")  
public class EnvironmentController {  
  
    @Resource    private DataManager dataManager;  
  
    @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)  
    public SseEmitter subscribe() {  
        SseEmitter emitter = new SseEmitter();  
        dataManager.subscribe("environment", emitter);  
        return emitter;  
    }  
  
    // 示例:推送环境监测数据给前端  
    @GetMapping("/push/{testText}")  
    public ResponseEntity<String> pushEnvironmentData(@PathVariable String testText) {  
        dataManager.pushData("environment", testText);  
        return ResponseEntity.ok("Data pushed successfully.");  
    }  
}

实现说明

每个不同类型的数据推送都需要一个对应的SSE订阅端点(subscribe)。每个数据类型都有一个对应的订阅端点,用于前端建立SSE连接,并在后端接收和处理特定类型的数据推送。

在你的后端应用中,对于每种数据类型,你需要创建一个对应的Controller或处理器来处理该数据类型的SSE订阅。每个Controller会有自己的SSE订阅端点,前端可以连接到不同的端点以接收相应类型的数据。

这种方式允许你将不同类型的数据推送逻辑分离,使代码更具可维护性和可扩展性。当有新的数据可用时,只需调用相应类型的数据推送方法,而不必修改通用的SSE管理逻辑。

前端实现

<!DOCTYPE html>
<html>
<head>
    <title>SSE Data Receiver</title>
</head>
<body>
    <h1>Real-time Data Display</h1>
    <div id="data-container"></div>
    <script>
        const dataContainer = document.getElementById('data-container');
        // 创建一个 EventSource 对象,指定 SSE 服务器端点的 URL
        const eventSource = new EventSource('http://127.0.0.1:13024/environment/subscribe'); // 根据你的控制器端点来设置URL
        // 添加事件处理程序,监听服务器端发送的事件
        eventSource.onmessage = (event) => {
            const data = event.data;
            // 在这里处理从服务器接收到的数据
            // 可以将数据显示在页面上或进行其他操作
            const newDataElement = document.createElement('p');
            newDataElement.textContent = data;
            dataContainer.appendChild(newDataElement);
        };
        eventSource.onerror = (error) => {
            // 处理连接错误
            console.error('Error occurred:', error);
        };
    </script>
</body>
</html>

弊端以及解决方案

如果你没什么处理的话,在它首次调用subscribe时候可能会出现连接超时的问题,因为这个是一个长连接,出现这种问题是因为,此时并没有数据产生,至此,除非你刷新页面,否则即使有数据产生前端也不会受到了

你希望前端在第一次订阅SSE连接后,即使后端没有数据产生,之后也能接收到数据。这可以通过以下方式来实现:

  1. 保持持久连接: 确保前端建立的SSE连接是持久性连接,不会在第一次连接成功后关闭。这可以让连接一直保持打开状态,即使后端没有即时数据产生。你可以在前端代码中使用以下方式来确保连接持久:
const eventSource = new EventSource('/environment/subscribe');
  1. 默认情况下,EventSource对象会自动重连,以保持连接的持久性。
  2. 定期发送心跳数据: 在后端定期发送一些心跳数据,以确保连接保持活跃。这可以防止连接超时关闭。你可以在后端定期发送一个包含无用信息的SSE事件,例如:
@Scheduled(fixedRate = 30000) // 每30秒发送一次心跳数据
public void sendHeartbeat() {
    dataManager.pushData("heartbeat", "Heartbeat data");
}
  1. 前端可以忽略这些心跳事件,但它们会保持连接处于活跃状态。
  2. 前端自动重连: 在前端代码中添加自动重连逻辑,以处理连接断开的情况。这样,如果连接由于某种原因断开,前端会自动尝试重新建立连接。示例:
const eventSource = new EventSource('/environment/subscribe');
eventSource.onerror = (error) => {
    // 处理连接错误
    console.error('Error occurred:', error);
    // 重新建立连接
    eventSource.close();
    setTimeout(() => {
        // 重新建立连接
        eventSource = new EventSource('/environment/subscribe');
    }, 1000); // 1秒后重试
};

通过结合上述方法,你可以确保前端能够建立并保持持久SSE连接,即使后端没有即时数据产生。这样,一旦后端有数据产生,前端也可以接收到数据而无需重新订阅。

相关文章
|
1月前
|
前端开发 JavaScript Python
Python Web应用中的WebSocket实战:前后端分离时代的实时数据交换
在前后端分离的Web应用开发模式中,如何实现前后端之间的实时数据交换成为了一个重要议题。传统的轮询或长轮询方式在实时性、资源消耗和服务器压力方面存在明显不足,而WebSocket技术的出现则为这一问题提供了优雅的解决方案。本文将通过实战案例,详细介绍如何在Python Web应用中运用WebSocket技术,实现前后端之间的实时数据交换。
79 0
|
27天前
|
JavaScript 前端开发 测试技术
前端全栈之路Deno篇(五):如何快速创建 WebSocket 服务端应用 + 客户端应用 - 可能是2025最佳的Websocket全栈实时应用框架
本文介绍了如何使用Deno 2.0快速构建WebSocket全栈应用,包括服务端和客户端的创建。通过一个简单的代码示例,展示了Deno在WebSocket实现中的便捷与强大,无需额外依赖,即可轻松搭建具备基本功能的WebSocket应用。Deno 2.0被认为是最佳的WebSocket全栈应用JS运行时,适合全栈开发者学习和使用。
|
1月前
|
JSON 前端开发 Java
震惊!图文并茂——Java后端如何响应不同格式的数据给前端(带源码)
文章介绍了Java后端如何使用Spring Boot框架响应不同格式的数据给前端,包括返回静态页面、数据、HTML代码片段、JSON对象、设置状态码和响应的Header。
136 1
震惊!图文并茂——Java后端如何响应不同格式的数据给前端(带源码)
|
28天前
|
监控 JavaScript 前端开发
前端的混合之路Meteor篇(六):发布订阅示例代码及如何将Meteor的响应数据映射到vue3的reactive系统
本文介绍了 Meteor 3.0 中的发布-订阅模型,详细讲解了如何在服务器端通过 `Meteor.publish` 发布数据,包括简单发布和自定义发布。客户端则通过 `Meteor.subscribe` 订阅数据,并使用 MiniMongo 实现实时数据同步。此外,还展示了如何在 Vue 3 中将 MiniMongo 的 `cursor` 转化为响应式数组,实现数据的自动更新。
|
28天前
|
JSON 分布式计算 前端开发
前端的全栈之路Meteor篇(七):轻量的NoSql分布式数据协议同步协议DDP深度剖析
本文深入探讨了DDP(Distributed Data Protocol)协议,这是一种在Meteor框架中广泛使用的发布/订阅协议,支持实时数据同步。文章详细介绍了DDP的主要特点、消息类型、协议流程及其在Meteor中的应用,包括实时数据同步、用户界面响应、分布式计算、多客户端协作和离线支持等。通过学习DDP,开发者可以构建响应迅速、适应性强的现代Web应用。
|
1月前
|
前端开发 算法 测试技术
前端大模型应用笔记(五):大模型基础能力大比拼-计数篇-通义千文 vs 文心一言 vs 智谱 vs 讯飞vsGPT
本文对比测试了通义千文、文心一言、智谱和讯飞等多个国产大模型在处理基础计数问题上的表现,特别是通过链式推理(COT)提示的效果。结果显示,GPTo1-mini、文心一言3.5和讯飞4.0Ultra在首轮测试中表现优秀,而其他模型在COT提示后也能显著提升正确率,唯有讯飞4.0-Lite表现不佳。测试强调了COT在提升模型逻辑推理能力中的重要性,并指出免费版本中智谱GLM较为可靠。
前端大模型应用笔记(五):大模型基础能力大比拼-计数篇-通义千文 vs 文心一言 vs 智谱 vs 讯飞vsGPT
|
1月前
|
JavaScript 前端开发 Python
django接收前端vue传输的formData图片数据
django接收前端vue传输的formData图片数据
43 4
|
28天前
|
NoSQL 前端开发 MongoDB
前端的全栈之路Meteor篇(三):运行在浏览器端的NoSQL数据库副本-MiniMongo介绍及其前后端数据实时同步示例
MiniMongo 是 Meteor 框架中的客户端数据库组件,模拟了 MongoDB 的核心功能,允许前端开发者使用类似 MongoDB 的 API 进行数据操作。通过 Meteor 的数据同步机制,MiniMongo 与服务器端的 MongoDB 实现实时数据同步,确保数据一致性,支持发布/订阅模型和响应式数据源,适用于实时聊天、项目管理和协作工具等应用场景。
|
1月前
|
网络协议 API 网络安全
Web实时通信的学习之旅:轮询、WebSocket、SSE的区别以及优缺点
Web实时通信的学习之旅:轮询、WebSocket、SSE的区别以及优缺点
150 0
|
1月前
|
存储 前端开发 API
前端开发中,Web Storage的存储数据的方法localstorage和sessionStorage的使用及区别
前端开发中,Web Storage的存储数据的方法localstorage和sessionStorage的使用及区别
93 0
下一篇
无影云桌面