Mri*_*jay 10 java spring spring-websocket project-reactor spring-webflux
我们正在尝试使用spring webflux WebSocket实现设计WebSocket服务器.服务器具有通常的HTTP服务器操作,例如create/fetch/update/fetchall.使用WebSockets,我们试图公开一个端点,以便客户端可以利用单个连接进行所有类型的操作,因为WebSockets就是为此目的而设计的.它是一个使用webflux和WebSockets的正确设计吗？
我们正在开始一个将使用反应式网络套接字的项目spring-webflux.我们需要构建一个反应式客户端库,消费者可以使用它来连接到服务器.
在服务器上,我们收到请求,读取消息,保存并返回静态响应:
public Mono<Void> handle(WebSocketSession webSocketSession) {
Flux<WebSocketMessage> response = webSocketSession.receive()
.map(WebSocketMessage::retain)
.concatMap(webSocketMessage -> Mono.just(webSocketMessage)
.map(parseBinaryToEvent) //logic to get domain object
.flatMap(e -> service.save(e))
.thenReturn(webSocketSession.textMessage(SAVE_SUCCESSFUL))
);
return webSocketSession.send(response);
}
在客户端,当有人调用save方法并从中返回响应时,我们想要打电话server.
public Mono<String> save(Event message) {
new ReactorNettyWebSocketClient().execute(uri, session -> {
session
.send(Mono.just(session.binaryMessage(formatEventToMessage)))
.then(session.receive()
.map(WebSocketMessage::getPayloadAsText)
.doOnNext(System.out::println).then()); //how to return this to client
});
return null;
}
我们不确定,如何设计这个.理想情况下,我们认为应该有
1)client.execute应该只调用一次并以某种方式持有session.应使用相同的会话在后续调用中发送数据.
2)如何从我们进入的服务器返回响应session.receive？
3)fetch如果响应很大(不仅仅是静态字符串而是事件列表),该session.receive怎么办？
我们正在做一些研究,但我们无法在网上找到适合webflux-websocket-client文档/实现的资源.关于如何前进的任何指示.
Ole*_*uka 10
这是绝对正确的设计,并且值得节省资源并且仅为每个客户端使用所有可能的操作的连接.
但是,不要实现轮子并使用协议,它为您提供所有这些类型的通信.
其中一个选项是使用RSocket协议的RSocket-Java实现.RSocket-Java建立在Project Reactor之上,因此它自然适合Spring WebFlux生态系统.
不幸的是,没有与Spring生态系统的特色集成.幸运的是,我花了几个小时来提供一个简单的RSocket Spring Boot Starter,它将Spring WebFlux与RSocket集成在一起,并将WebSocket RSocket服务器与WebFlux Http服务器一起公开.
基本上,RSocket隐藏了自己实现相同方法的复杂性.使用RSocket,我们不必关心交互模型定义作为自定义协议和Java中的实现.RSocket为我们提供了将数据传送到特定逻辑信道的功能.它提供了一个内置客户端,可以将消息发送到同一个WS连接,因此我们不必为此创建自定义实现.
由于RSocket只是一个协议,它不提供任何消息格式,因此这个挑战是针对业务逻辑的.但是,有一个RSocket-RPC项目,它提供一个协议缓冲区作为消息格式,并重用与GRPC相同的代码生成技术.因此,使用RSocket-RPC,我们可以轻松地为客户端和服务器构建API,并且根本不需要传输和协议抽象.
相同的RSocket Spring Boot集成也提供了RSocket-RPC使用的示例.
所以,为此目的,你必须自己实现这个地狱.我之前已经做过一次,但我不能指出那个项目,因为它是一个企业项目.不过,我可以分享几个代码示例,可以帮助您构建适当的客户端和服务器.
必须考虑的第一点是,一个物理连接中的所有逻辑流都应存储在某处:
class MyWebSocketRouter implements WebSocketHandler {
final Map<String, EnumMap<ActionMessage.Type, ChannelHandler>> channelsMapping;
@Override
public Mono<Void> handle(WebSocketSession session) {
final Map<String, Disposable> channelsIdsToDisposableMap = new HashMap<>();
...
}
}
上面的示例中有两个地图.第一个是您的路由映射,它允许您根据传入的消息参数识别路由,或者左右.第二个是为请求流用例创建的(在我的例子中它是活动订阅的映射),因此您可以发送创建订阅的消息帧,或者订阅特定操作并在取消订阅后保留该订阅执行操作如果存在订阅,您将被取消订阅.
为了从所有逻辑流发回消息,您必须将消息多路复用到一个流.例如,使用Reactor,您可以使用UnicastProcessor以下方法执行此操作:
@Override
public Mono<Void> handle(WebSocketSession session) {
final UnicastProcessor<ResponseMessage<?>> funIn = UnicastProcessor.create(Queues.<ResponseMessage<?>>unboundedMultiproducer().get());
...
return Mono
.subscriberContext()
.flatMap(context -> Flux.merge(
session
.receive()
...
.cast(ActionMessage.class)
.publishOn(Schedulers.parallel())
.doOnNext(am -> {
switch (am.type) {
case CREATE:
case UPDATE:
case CANCEL: {
...
}
case SUBSCRIBE: {
Flux<ResponseMessage<?>> flux = Flux
.from(
channelsMapping.get(am.getChannelId())
.get(ActionMessage.Type.SUBSCRIBE)
.handle(am) // returns Publisher<>
);
if (flux != null) {
channelsIdsToDisposableMap.compute(
am.getChannelId() + am.getSymbol(), // you can generate a uniq uuid on the client side if needed
(cid, disposable) -> {
...
return flux
.subscriberContext(context)
.subscribe(
funIn::onNext, // send message to a Processor manually
e -> {
funIn.onNext(
new ResponseMessage<>( // send errors as a messages to Processor here
0,
e.getMessage(),
...
ResponseMessage.Type.ERROR
)
);
}
);
}
);
}
return;
}
case UNSABSCRIBE: {
Disposable disposable = channelsIdsToDisposableMap.get(am.getChannelId() + am.getSymbol());
if (disposable != null) {
disposable.dispose();
}
}
}
})
.then(Mono.empty()),
funIn
...
.map(p -> new WebSocketMessage(WebSocketMessage.Type.TEXT, p))
.as(session::send)
).then()
);
}
正如我们从上面的示例中看到的,那里有很多东西:
Flux消息或只是一个Mono(如果是单声道,它可以在服务器端实现更简单,所以你不必保持唯一的流ID).客户端也不那么简单:
为了处理连接,我们必须分配两个处理器,以便我们可以使用它们来复用和解复用消息:
UnicastProcessor<> outgoing = ...
UnicastPorcessor<> incoming = ...
(session) -> {
return Flux.merge(
session.receive()
.subscribeWith(incoming)
.then(Mono.empty()),
session.send(outgoing)
).then();
}
所有创建的流无论是Mono或Flux应该存储在某个地方,所以我们将能够区分哪个流消息涉及:
Map<String, MonoSink> monoSinksMap = ...;
Map<String, FluxSink> fluxSinksMap = ...;
自MonoSink以来我们必须保留两张地图,而FluxSink没有相同的父接口.
在上面的示例中,我们只考虑了客户端的初始部分.现在我们必须构建一个消息路由机制:
...
.subscribeWith(incoming)
.doOnNext(message -> {
if (monoSinkMap.containsKey(message.getStreamId())) {
MonoSink sink = monoSinkMap.get(message.getStreamId());
monoSinkMap.remove(message.getStreamId());
if (message.getType() == SUCCESS) {
sink.success(message.getData());
}
else {
sink.error(message.getCause());
}
} else if (fluxSinkMap.containsKey(message.getStreamId())) {
FluxSink sink = fluxSinkMap.get(message.getStreamId());
if (message.getType() == NEXT) {
sink.next(message.getData());
}
else if (message.getType() == COMPLETE) {
fluxSinkMap.remove(message.getStreamId());
sink.next(message.getData());
sink.complete();
}
else {
fluxSinkMap.remove(message.getStreamId());
sink.error(message.getCause());
}
}
})
上面的代码示例显示了我们如何路由传入的消息.
最后一部分是消息多路复用.为此,我们将涵盖可能的发送者类impl:
class Sender {
UnicastProcessor<> outgoing = ...
UnicastPorcessor<> incoming = ...
Map<String, MonoSink> monoSinksMap = ...;
Map<String, FluxSink> fluxSinksMap = ...;
public Sender () {
//在这里创建websocket连接并放置前面提到的代码}
Mono<R> sendForMono(T data) {
//generate message with unique
return Mono.<R>create(sink -> {
monoSinksMap.put(streamId, sink);
outgoing.onNext(message); // send message to server only when subscribed to Mono
});
}
Flux<R> sendForFlux(T data) {
return Flux.<R>create(sink -> {
fluxSinksMap.put(streamId, sink);
outgoing.onNext(message); // send message to server only when subscribed to Flux
});
}
}
| 归档时间： |
|
| 查看次数： |
1543 次 |
| 最近记录： |