java - graphql-java - 如何在 spring boot 中使用订阅？

Question

在一个项目中我使用 graphql-java 和带有 postgreSQL 数据库的 spring boot。现在我想使用 subscription feature 发布于 3.0.0 版本。不幸的是，关于订阅功能应用的资料还不是很成熟。

使用带有订阅的 graphql-java 实现实时功能的方法是怎样的？

Answer 1

从最近的 graphql-java 版本开始，完全支持订阅。订阅的 DataFetcher 必须返回一个 org.reactivestreams.Publisher，而 graphql-java 将负责将查询函数映射到结果上。

功能很好documented还有 a complete example使用官方仓库中可用的网络套接字。

如果您有一个响应式(Reactive)数据源(例如带有反应式驱动程序的 Mongo，或者可能是 R2DBC 支持的任何东西)，您就已经准备好了。只需使用 @Tailable并且 Spring Data 已经为您提供了一个 Flux(它实现了 Publisher)，您无需执行任何其他操作。

至于更手动的 Spring 特定实现，我无法想象它太难使用了 Spring's own event mechanism (也是一个很好的教程 here)作为 Publisher 的基础。

每次有传入订阅时，创建一个新的监听器并将其注册到应用程序上下文:context.addApplicationListener(listener)，它将发布到正确的Publisher。例如。在 DataFetcher 中:

// Somehow create a publisher, probably using Spring's Reactor project. Or RxJava.
Publisher<ResultObject> publisher = ...; 
//The listener reacts on application events and pushes new values through the publisher
ApplicationListener listener = createListener(publisher);
context.addApplicationListener(listener);
return publisher;

当网络套接字断开连接或者您以某种方式知道事件流已完成时，您必须确保移除监听器。

请注意，我实际上还没有尝试过任何这些，我只是在大声思考。

另一种选择是直接使用 Reactor(有或没有 Spring WebFlux)。有一个使用 Reactor 和 WebSocket 的示例(通过 GraphQL SPQR Spring Boot Starter )here .

你像这样创建一个Publisher:

//This is really just a thread-safe wrapper around Map<String, Set<FluxSink<Task>>>
private final ConcurrentMultiRegistry<String, FluxSink<Task>> subscribers = new ConcurrentMultiRegistry<>();

@GraphQLSubscription
public Publisher<Task> taskStatusChanged(String taskId) {
    return Flux.create(subscriber -> subscribers.add(taskId, subscriber.onDispose(() -> subscribers.remove(taskId, subscriber))), FluxSink.OverflowStrategy.LATEST);
}

然后像这样从其他地方(可能是相关的突变或 react 性存储)推送新值:

subscribers.get(taskId).forEach(subscriber -> subscriber.next(task));

例如

@GraphQLMutation
public Task updateTask(@GraphQLNonNull String taskId, @GraphQLNonNull Status status) {
    Task task = repo.byId(taskId); //find the task
    task.setStatus(status); //update the task
    repo.save(task); //persist the task
    //Notify all the subscribers following this task
    subscribers.get(taskId).forEach(subscriber -> subscriber.next(task));
    return task;
}

使用 SPQR Spring Starter，这就是让您获得与 Apollo 兼容的订阅实现所需的全部内容。