graphql-java - 如何在 Spring Boot 中使用订阅? [英] graphql-java - How to use subscriptions with spring boot?

问题描述

在一个项目中，我使用 graphql-java 和spring boot 使用 postgreSQL 数据库.现在我想使用订阅功能 3.0.0 版发布.遗憾的是，订阅功能的应用信息还不是很成熟.

In a project I use graphql-java and spring boot with a postgreSQL Database. Now I would like to use the subscription feature published in version 3.0.0. Unfortunately, the information about the application of the subsciption function is not very mature.

如何通过订阅使用 graphql-java 实现实时功能?

How is the approach to achieve real-time functionality using graphql-java with subscriptions?

推荐答案

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

As of recent graphql-java versions, subscriptions are fully supported. The DataFetcher for a subscription must return a org.reactivestreams.Publisher, and graphql-java will take care of mapping the query function over the results.

该功能很好文档化，并且有完整示例 使用官方存储库中提供的网络套接字.

The feature is nicely documented and there's a complete example using web sockets available in the official repo.

如果您有一个响应式数据源(例如带有响应式驱动程序的 Mongo，或者任何R2DBC 支持)，一切就绪.只需使用 @Tailable 和 Spring Data 已经会给你一个 Flux(它实现了 Publisher)，你不需要做任何其他事情.

If you have a reactive data source in place (e.g. Mongo with a reactive driver, or probably anything that R2DBC supports), you're all set. Just use @Tailable and Spring Data will already give you a Flux (which implements Publisher) and there's nothing else you need to do.

至于更手动的 Spring 特定实现，我无法想象使用 Spring 自己的事件机制(一个不错的教程 此处 也是 Publisher 的基础.

As for a more manual Spring specific implementation, I can't imagine it being too hard to use Spring's own event mechanism (a nice tutorial here as well) to underlie the Publisher.

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

Every time there's an incoming subscription, create and register a new listener with the application context: context.addApplicationListener(listener) that will publish to the correct Publisher. E.g. in the 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;

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

When the web socket disconnects or you somehow know the event stream is finished, you must make sure to remove the listener.

我还没有真正尝试过这些，请注意，我只是在大声思考.

I haven't actually tried any of this, mind you, I'm just thinking aloud.

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

Another option is to use Reactor directly (with or without Spring WebFlux). There's a sample using Reactor and WebSocket (through 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);
}

然后从其他地方推送新值(可能是相关的突变或反应式存储)，如下所示:

And then push new values from elsewhere (probably a related mutation or a reactive storage) like this:

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 兼容的订阅实施所需的全部内容.

With SPQR Spring Starter, this is all that's needed to get you an Apollo-compatible subscription implementation.

这篇关于graphql-java - 如何在 Spring Boot 中使用订阅?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！