使用 Spring Boot 实现 SSE(送外卖示例)

你以前听说过 SSE（服务器发送事件）吗？它是一种通过 HTTP 协议工作的单向消息传递技术，可以从服务器向客户端发送消息。因此，每当有可用数据时，客户端都会实时接收并更新。

SSE 工作原理

SSE 具有单向通信功能，只有服务器会向客户端发送事件，这些事件由 JavaScript 中的 EventSource API 捕捉，浏览器使用该 API 监听事件（我们将在本文稍后部分详细讨论）。使用 SSE 无法发送二进制数据，只能发送消息。我们可以在天气、体育、股票交易等应用中找到一些 SSE 用例。

SSE 是唯一的选择吗？

替代 SSE 的选择不止一种，但两者没有优劣之分，这取决于您的需求。除了 SSE，你还可以找到一些替代方案，比如 WebSocket 和 Event Data Pooling，但如果你只需要从服务器获取一些接收到的数据，也许 SSE 才是最佳选择。让我来解释一下两者的区别：

• WebSocket

WebSocket 与 SSE 稍有不同，它是一种在客户端和服务器之间进行双向通信的协议。除了信息，还可以在有效载荷中发送二进制文件。一些 WebSocket 用例包括聊天应用和多人游戏，这两种应用都需要全双工通信。

• Event Data Pooling

与 SSE 和 WebSocket 不同的是，客户端需要向服务器请求数据，这通常会占用更多的硬件资源，因为在服务器做出响应之前，客户端会发出大量请求。

如何使用 Spring Boot 实现 SSE？

考虑到本文的重点是 SSE，我们将通过一个实践示例来说明如何在 Java 应用程序中使用 Spring Boot 框架。

我们将使用 Servlet Stack (MVC)，因为它简单易懂，而且大多数开发人员都习惯使用命令式代码进行开发。不过，你也可以使用反应堆栈（WebFlux）来实现它。

让我们通过一个食品配送系统(送外卖)来了解它是如何工作的。我们要做的是向用户显示订单进度，可用的状态有：订单已下、在厨房、在路上和已送达。

考虑到这只是一个用来说明 SSE 用例的原型，系统会自动处理所有的订单步骤。

创建 SSE 入口点

首先，需要在控制器中声明一个返回 SseEmitter 接口的端点。

private final Collection<SseEmitter> emitters = new CopyOnWriteArrayList<>();

@GetMapping(path = "/order-status", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
SseEmitter orderStatus() {

   SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

   emitter.onCompletion(() -> emitters.remove(emitter));
   emitter.onTimeout(() -> emitters.remove(emitter));
   emitter.onError(throwable -> {
      emitters.remove(emitter);
      emitter.completeWithError(throwable);
   });

   emitters.add(emitter);

   return emitter;
}

在接口签名中，我们将媒体类型设为 TEXT_EVENT_STREAM_VALUE，以便向客户端发送实时事件。我们的方法主体由一个新的 SseEmitter 组成，该 SseEmitter 添加到并发列表中，从而使广播发送成为可能。每当发生完成、错误或超时时，我们还会声明一些事件回调。

声明谁需要知道订单信息

在端点中连接用户后，我们如何确定订单已到达并需要准备？考虑到我们要在订单状态发生变化时向客户端发送事件，为什么不采用可观察模式并在每次变化后通知所有监听器呢？

让我们创建一个监听器，并注册代表每个订单状态的所有可观察对象。

@Component
public class OrderFoodListener {

    private final EventService eventService;

    private final Collection<Observer> observers;

    public OrderFoodListener(EventService eventService) {
        this.eventService = eventService;
        this.observers = List.of(
                new OrderObservable(eventService),
                new KitchenObservable(eventService),
                new OnTheWayObservable(eventService),
                new DeliveredObservable(eventService)
        );
    }

   public void notifyAll(OrderFood orderFood) {
        observers.forEach(foodObserver -> foodObserver.update(orderFood));
   }
}

让我来解释一下这里发生了什么：

• Event Service 接口：这是我们发送事件方式的依赖反转接口（具体类应在基础架构层中声明）
• Observers 集合： 我们希望在订单到来时通知的所有观察者。
• 通知方法： 将调用 notifyAll 方法通知所有观察者新订单的信息。

通知客户订单

正如我们在前面的代码块中看到的，有四个观测变量，每个观测变量都有自己的业务规则。

让我向您展示一下我们将如何通知客户：

• 订单即将到来

if (orderFood.getStatus() == ORDER_PLACED) {

   sendEvent(orderFood, "order");
}

订单观测器会检查初始状态是否为已下订单，如果是，它就会发送一个事件，事件名称为 “订单”。

• 它将进入厨房

if (orderFood.getStatus() == FoodStatus.ORDER_PLACED) {

   orderFood.setStatus(FoodStatus.IN_THE_KITCHEN);
   sendEvent(orderFood, "kitchen");
   waitForProcess();
}

下单后，订单将被送往厨房，因此我们需要检查之前的状态是否仍为下单状态，然后进行更改，并发送包含更新状态的 “kitchen “事件。

OBS. waitForProcess()方法是一个线程睡眠声明，用于模拟厨房的持续进程。

• 现在就看送餐员的了

if (orderFood.getStatus() == IN_THE_KITCHEN) {

   orderFood.setStatus(ON_THE_WAY);
   sendEvent(orderFood, "on-the-way");
   waitForProcess();
}

现在是通知用户已到达的时候了。

在客户端接收事件

在客户端，用户将等待订单更新。因此，让我们来实现它。

var eventSource = new EventSource("/order-status");

eventSource.addEventListener("order", (event) => renderEvent(event));

eventSource.addEventListener("kitchen", (event) => renderEvent(event));

eventSource.addEventListener("on-the-way", (event) => renderEvent(event));

eventSource.addEventListener("delivered", (event) => renderEvent(event));

创建了一个简单的静态 HTML 文件，并放置了一些 JavaScript 来监听来自服务器的所有事件并将它们打印在屏幕上。

结论

希望通过本文，你能理解 SSE 的主要概念，并了解类似技术之间的区别。此外，我们还看到了一个如何使用 Spring Boot 框架实现 SSE 的实践示例。

GitHub 看看本文使用的全部代码：https://github.com/GermanoSchneider/food-delivery-sse-app。