PROSAGA码农传奇-运筹学-Webflux websocketclient，如何在同一会话中发送多个请求[设计客户端库]

<div class =“post-text”itemprop =“text”>
  <H1>
    请！使用
    <a href="http://rsocket.io/" rel="nofollow noreferrer">
      方法RSocket
    </A>
    ！
  </H1>
  <P>
    这是绝对正确的设计，并且值得节省资源并且仅为每个客户端使用所有可能的操作的连接。
  </p>
  <P>
    但是，不要实现轮子并使用协议，它为您提供所有这些类型的通信。
  </p>
  <UL>
    <LI>
      RSocket有一个
      <EM>
        请求 - 响应
      </EM>
       模型，它允许您今天做最常见的客户端 -  servert交互。
    </LI>
    <LI>
      RSocket有一个
      <EM>
        请求流
      </EM>
       通信模型，以便您可以满足所有需求并返回异步重用相同连接的事件流。 RSocket将逻辑流的所有内容都连接到物理连接并返回，因此您不会感受到自己这样做的痛苦。
    </LI>
    <LI>
      RSocket有更多的交互模型，比如

<EM>
        射后不理
      </EM>
       和
      <EM>
        流流
      </EM>
       这可能是有用的
以两种方式发送数据流。
    </LI>
  </UL>
  <H2>
    如何在Spring中使用RSocket
  </H2>
  <P>
    其中一个选项是使用RSocket协议的RSocket-Java实现。 RSocket-Java建立在Project Reactor之上，因此它自然适合Spring WebFlux生态系统。
  </p>
  <P>
    不幸的是，没有与Spring生态系统的特色集成。幸运的是，我花了几个小时来提供一个简单的
    <a href="https://github.com/CollaborationInEncapsulation/spring-boot-rsocket" rel="nofollow noreferrer">
      RSocket Spring Boot Starter
    </A>
     将Spring WebFlux与RSocket集成，并将WebSocket RSocket服务器与WebFlux Http服务器一起公开。
  </p>
  <H2>
    为什么RSocket是更好的方法？
  </H2>
  <P>
    基本上，RSocket隐藏了自己实现相同方法的复杂性。使用RSocket，我们不必关心交互模型定义作为自定义协议和Java中的实现。 RSocket为我们提供了将数据传送到特定逻辑信道的功能。它提供了一个内置客户端，可以将消息发送到同一个WS连接，因此我们不必为此创建自定义实现。
  </p>
  <H2>
    用它做得更好
    <a href="https://github.com/rsocket/rsocket-rpc-java/" rel="nofollow noreferrer">
      方法RSocket-RPC
    </A>
  </H2>
  <P>
    由于RSocket只是一个协议，它不提供任何消息格式，因此这个挑战是针对业务逻辑的。但是，有一个RSocket-RPC项目，它提供一个协议缓冲区作为消息格式，并重用与GRPC相同的代码生成技术。因此，使用RSocket-RPC，我们可以轻松地为客户端和服务器构建API，并且根本不需要传输和协议抽象。
  </p>
  <P>
    相同的RSocket Spring Boot集成提供了一个
    <a href="https://github.com/CollaborationInEncapsulation/spring-boot-rsocket/tree/master/samples" rel="nofollow noreferrer">
      例
    </A>
     RSocket-RPC的用法也是如此。
  </p>
  <H1>
    好吧，它还没有说服我，我想要一个自定义的WebSocket服务器
  </H1>
  <P>
    所以，为此目的，你必须自己实现这个地狱。我之前已经做过一次，但我不能指出那个项目，因为它是一个企业项目。
不过，我可以分享几个代码示例，可以帮助您构建适当的客户端和服务器。
  </p>
  <H2>
    服务器端
  </H2>
  <H3>
    处理程序和Open逻辑订阅服务器映射
  </H3>
  <P>
    必须考虑的第一点是，一个物理连接中的所有逻辑流都应存储在某处：
  </p>
   <pre>
    <code>
      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<>();
    ...
  }
}

</code>
  </pre>
  <P>
    上面的示例中有两个地图。第一个是您的路由映射，它允许您根据传入的消息参数识别路由，或者左右。第二个是为请求流用例创建的（在我的例子中是活动订阅的映射），因此您可以发送创建订阅的消息帧，或者订阅特定操作并保留该订阅以便取消订阅执行操作如果存在订阅，您将被取消订阅。
  </p>
  <H3>
    使用Processor进行消息多路复用
  </H3>
  <P>
    为了从所有逻辑流发回消息，您必须将消息多路复用到一个流。例如，使用Reactor，您可以使用
     <code>
      UnicastProcessor
    </code>
    ：
  </p>
   <pre>
    <code>
      @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()
    );
}

</code>
  </pre>
  <P>
    正如我们从上面的示例中看到的，那里有很多东西：
  </p>
  <OL>
    <LI>
      该消息应包括路由信息
    </ LI>
    <LI>
      该消息应包含与其相关的唯一流ID。
    </LI>
    <LI>
      用于消息多路复用的独立处理器，其中错误也应该是消息
    </LI>
    <LI>
      每个频道都应该存储在某个地方，在这种情况下我们都有一个简单的用例，每个消息都可以提供一个
       <code>
        Flux
      </code>
       消息或只是一个
       <code>
        Mono
      </code>
       （在单声道的情况下，它可以在服务器端更简单地实现，因此您不必保留唯一的流ID）。
    </LI>
    <LI>
      此示例不包含编码解码消息，因此您将面临这一挑战。
    </LI>
  </醇>
  <H2>
    客户端
  </H2>
  <P>
    客户端也不那么简单：
  </p>
  <H3>
    处理会话
  </H3>
  <P>
    为了处理连接，我们必须分配两个处理器，以便我们可以使用它们来复用和解复用消息：
  </p>
   <pre>
    <code>
      UnicastProcessor<> outgoing = ...
UnicastPorcessor<> incoming = ...
(session) -> {
  return Flux.merge(
     session.receive()
            .subscribeWith(incoming)
            .then(Mono.empty()),
     session.send(outgoing)
  ).then();
}

</code>
  </pre>
  <H3>
    将所有逻辑流保存在某处
  </H3>
  <P>
    所有创建的流都是
     <code>
      Mono
    </code>
     要么
     <code>
      Flux
    </code>
     应存储在某处，以便我们能够区分哪个流消息涉及：
  </p>
   <pre>
    <code>
      Map<String, MonoSink> monoSinksMap = ...;
Map<String, FluxSink> fluxSinksMap = ...;

</code>
  </pre>
  <P>
    自MonoSink以来我们必须保留两张地图，而FluxSink没有相同的父接口。
  </p>
  <H3>
    消息路由
  </H3>
  <P>
    在上面的示例中，我们只考虑了客户端的初始部分。现在我们必须构建一个消息路由机制：
  </p>
   <pre>
    <code>
      ...
.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());
        }
    }
})

</code>
  </pre>
  <P>
    上面的代码示例显示了我们如何路由传入的消息。
  </p>
  <H3>
    多路请求
  </H3>
  <P>
    最后一部分是消息多路复用。为此，我们将涵盖可能的发送者类impl：
  </p>
   <pre>
    <code>
      class Sender {
    UnicastProcessor<> outgoing = ...
    UnicastPorcessor<> incoming = ...

Map<String, MonoSink> monoSinksMap = ...;
    Map<String, FluxSink> fluxSinksMap = ...;

public Sender () {

</code>
  </pre>
  <P>
    //在这里创建websocket连接并放置前面提到的代码
        }
  </p>
   <pre>
    <code>
          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
        });
     }
}

</code>
  </pre>
  <H2>
    自定义实现的总结
  </H2>
  <OL>
    <LI>
      性交
    </LI>
    <LI>
      没有实施背压支持，这可能是另一项挑战
    </LI>
    <LI>
      轻松拍摄自己的脚
    </LI>
  </醇>
  <H1>
    小贴士
  </H1>
  <OL>
    <LI>
      请使用RSocket，不要自己发明协议，这很难！
    </LI>
    <LI>
      要了解更多关于来自Pivotal球员的RSocket  - 
      <a href="https://www.youtube.com/watch?v=WVnAbv65uCU" rel="nofollow noreferrer">
        https://www.youtube.com/watch?v=WVnAbv65uCU
      </A>
    </LI>
    <LI>
      从我的一次演讲中了解更多关于RSocket的信息 - 
      <a href="https://www.youtube.com/watch?v=XKMyj6arY2A" rel="nofollow noreferrer">
        https://www.youtube.com/watch?v=XKMyj6arY2A
      </A>
    </LI>
    <LI>
      在RSocket之上构建了一个名为Proteus的特色框架 - 您可能对此感兴趣 - 
      <a href="https://www.netifi.com/" rel="nofollow noreferrer">
        https://www.netifi.com/
      </A>
    </LI>
    <LI>
      要从RSocket协议的核心开发人员那里了解有关Proteus的更多信息 - 
      <a href =“https://www.google.com/url?sa=t&source=web&rct=j&url=https://m.youtube.com/watch%3Fv%3D_rqQtkIeNIQ&ved=2ahUKEwjpyLTpsLzfAhXDDiwKHUUUA8gQt9IBMAR6BAgNEB8& ; usg = AOvVaw0B_VdOj42gjr0YrzLLUX1E“rel =”nofollow noreferrer“>
        https://www.google.com/url?sa=t&source=web&rct=j&url=https://m.youtube.com/watch%3Fv%3D_rqQtkIeNIQ&ved=2ahUKEwjpyLTpsLzfAhXDDiwKHUUUA8gQt9IBMAR6BAgNEB8&usg=AOvVaw0B_VdOj42gjr0YrzLLUX1E
      </A>
    </LI>
  </醇>
</DIV>