跳转到内容
FEAT

SSE 客户端详解

Server-Sent Events (SSE) 是一种服务器向客户端推送实时更新的技术。与 WebSocket 不同，SSE 是单向通信，只能服务器向客户端推送数据，但使用更简单，因为它是基于 HTTP 协议的。

快速上手

Section titled “快速上手”

以下是一个简单的示例，展示了如何使用 Feat SSEClient 连接 SSE 服务器并处理消息：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.SseEvent;


public class SSEClientDemo {
    public static void main(String[] args) {
        // 创建 HttpClient 实例
        HttpClient httpClient = Feat.httpClient("http://localhost:8080");


        // 注册 SSE 处理器并提交请求
        httpClient.get("/events")
                .onSSE(sseClient -> sseClient
                        // 注册默认事件处理器
                        .onData(event -> {
                            System.out.println("收到数据: " + event.getData());
                        })
                        // 注册连接成功回调
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }))
                // 提交请求
                .submit();
    }
}

API 参考：核心接口

Section titled “API 参考：核心接口”

在开始使用 SSEClient 之前，让我们先了解一下它的核心接口。

HttpRest.onSSE() 方法

Section titled “HttpRest.onSSE() 方法”

要使用 SSE 客户端，需要通过 HttpRest.onSSE() 方法注册 SSE 处理器：

HttpClient client = Feat.httpClient("http://localhost:8080");
client.get("/sse").onSSE(sseClient -> {
    // 配置 SSE 事件处理器
});

SseClient 类

Section titled “SseClient 类”

SseClient 是在 onSSE() 回调中提供的客户端对象：

方法说明
onData(Consumer<SseEvent>)处理默认数据事件
onEvent(String, Consumer<SseEvent>)处理特定类型事件
onOpen(Consumer<SseClient>)连接成功回调

SseEvent 类

Section titled “SseEvent 类”

SseEvent 封装了从服务器接收到的事件数据：

属性获取方法说明
idgetId()事件 ID
eventgetEvent()事件类型
datagetData()事件数据
retrygetRetry()重试时间

处理特定事件类型

Section titled “处理特定事件类型”

SSE 支持不同的事件类型，可以通过 onEvent 方法处理特定类型的事件：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.SseEvent;


public class SSEEventTypesDemo {
    public static void main(String[] args) {
        HttpClient httpClient = Feat.httpClient("http://localhost:8080");


        httpClient.get("/events")
                .onSSE(sseClient -> sseClient
                        // 处理默认消息事件
                        .onData(event -> {
                            System.out.println("默认消息: " + event.getData());
                        })
                        // 处理特定类型事件
                        .onEvent("notification", event -> {
                            System.out.println("通知事件: " + event.getData());
                        })
                        .onEvent("update", event -> {
                            System.out.println("更新事件: " + event.getData());
                        })
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }))
                .submit();
    }
}

高级配置

Section titled “高级配置”

如果需要自定义连接参数，可以通过 HttpClient 进行配置：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.SseEvent;


public class SSEAdvancedConfigDemo {
    public static void main(String[] args) {
        HttpClient httpClient = Feat.httpClient("http://localhost:8080", opt -> {
                    opt.connectTimeout(5000) // 设置连接超时时间为 5 秒
                       .readBufferSize(8192) // 设置读缓冲区大小为 8 KB
                       .debug(true); // 开启调试模式
                });


        httpClient.get("/events")
                // 添加自定义请求头
                .header(header -> {
                    header.add("Authorization", "Bearer your-token");
                    header.add("Last-Event-ID", "event-123");
                })
                .onSSE(sseClient -> sseClient
                        .onData(event -> {
                            System.out.println("收到数据: " + event.getData());
                            System.out.println("事件ID: " + event.getId());
                            System.out.println("事件类型: " + event.getEvent());
                        })
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }))
                .submit();
    }
}

手动关闭连接

Section titled “手动关闭连接”

当不再需要 HTTP 连接时，可以手动关闭它：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.SseEvent;


public class SSECloseDemo {
    public static void main(String[] args) throws InterruptedException {
        HttpClient httpClient = Feat.httpClient("http://localhost:8080");


        // 保存 HttpRest 引用以便后续关闭连接
        HttpRest httpRest = httpClient.get("/events")
                .onSSE(sseClient -> sseClient
                        .onData(event -> {
                            System.out.println("收到数据: " + event.getData());
                        })
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }));


        // 提交请求
        httpRest.submit();


        // 5秒后手动关闭连接
        Thread.sleep(5000);
        httpRest.close();
        System.out.println("连接已关闭");
    }
}

最佳实践

Section titled “最佳实践”

在实际项目中使用 SSEClient 时，有一些最佳实践可以帮助你写出更好的代码。

1. 错误处理和重连

Section titled “1. 错误处理和重连”

在生产环境中，网络问题可能导致连接断开，建议实现重连机制：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.SseEvent;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


public class SSEReconnectDemo {
    private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);


    public static void main(String[] args) {
        connectSSE();
    }


    private static void connectSSE() {
        HttpClient httpClient = Feat.httpClient("http://localhost:8080");


        httpClient.get("/events")
                .onSSE(sseClient -> sseClient
                        .onData(event -> {
                            System.out.println("收到数据: " + event.getData());
                        })
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }))
                .onFailure(throwable -> {
                    System.err.println("发生错误: " + throwable.getMessage());
                    // 安排重连
                    scheduleReconnect();
                })
                .onClose(() -> {
                    System.out.println("连接已关闭");
                    // 安排重连
                    scheduleReconnect();
                })
                .submit();
    }


    private static void scheduleReconnect() {
        System.out.println("将在 5 秒后尝试重连...");
        scheduler.schedule(SSEReconnectDemo::connectSSE, 5, TimeUnit.SECONDS);
    }
}

2. 资源管理

Section titled “2. 资源管理”

确保在应用程序关闭时正确释放资源：

import tech.smartboot.feat.Feat;
import tech.smartboot.feat.core.client.HttpClient;
import tech.smartboot.feat.core.client.HttpRest;
import tech.smartboot.feat.core.client.SseEvent;


public class SSEResourceManagementDemo {
    private static HttpRest httpRest;


    public static void main(String[] args) {
        connectSSE();


        // 添加关闭钩子以确保资源被释放
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (httpRest != null) {
                httpRest.close();
                System.out.println("HTTP 连接已关闭");
            }
        }));
    }


    private static void connectSSE() {
        HttpClient httpClient = Feat.httpClient("http://localhost:8080");


        httpRest = httpClient.get("/events")
                .onSSE(sseClient -> sseClient
                        .onData(event -> {
                            System.out.println("收到数据: " + event.getData());
                        })
                        .onOpen(client -> {
                            System.out.println("SSE 连接已建立");
                        }))
                .onFailure(throwable -> {
                    System.err.println("发生错误: " + throwable.getMessage());
                });


        httpRest.submit();
    }
}

性能与扩展性

Section titled “性能与扩展性”

高性能

Section titled “高性能”

Feat SSEClient 基于 Feat 框架的高性能网络通信能力，能够支持大规模的并发连接和高频率的消息传输。

线程安全

Section titled “线程安全”

Feat SSEClient 的设计充分考虑了线程安全问题，保证在高并发场景下的稳定性。

总结

Section titled “总结”

通过这篇文章，我们学习了 Feat SSEClient 的核心功能：

  1. 如何建立 SSE 连接
  2. 如何处理不同类型的事件
  3. 如何配置客户端选项
  4. 如何实现错误处理和重连机制

Feat SSEClient 是一款功能丰富、性能优越的 SSE 客户端实现，能够满足大多数实时数据推送场景的需求。通过简单的 API 和灵活的配置选项，开发者可以快速构建高效的实时数据接收应用。

如果你正在寻找一款高性能、易用的 SSE 客户端，Feat SSEClient 是一个值得考虑的选择。

有关完整示例，请参见 SSEClient 测试代码