5

在我的项目中，我想让多个客户端连接到服务。我正在使用 java Rsocket 实现。

服务应该为每个客户端维护一个状态。现在，我可以通过一些标识符来管理客户端。这个选项我已经实现了。但我不想使用字符串手动管理会话。

所以另一个想法是通过 Rsocket 连接来识别客户端。有没有办法使用 Rsocket 通道来识别特定的客户端？

想象一个示例服务和几个客户端。每个客户端都有 Rsocket 通道，服务启动并运行。有没有办法使用 Rsocket 通道在服务器端识别这些客户端？如果您可以展示这种行为的编程示例，那将是惊人的。谢谢！

编辑（更详细地描述案例）

这是我的例子。

我们目前使用了三个 CORBA 对象，如图所示：

  • LoginObject（通过 NamingService 检索到的引用）。客户端可以调用 login() 方法来获取会话
  • Session 对象有多种方法用于查询当前服务上下文的详细信息，最重要的是获取 Transaction 对象
  • Transaction 对象可用于通过将 commandName 和键值对列表作为参数的通用方法执行各种命令。在客户端执行 n 个命令后，他可以提交或回滚事务（也可以通过 Transaction 对象上的方法）。

在此处输入图像描述

所以这里我们使用会话对象在我们的服务上执行事务。

现在我们决定从 CORBA 迁移到 Rsocket。因此我们需要 Rsocket 微服务能够存储会话的状态，否则我们无法知道将要提交或回滚什么。这可以通过每个客户的单个发布者来完成吗？

4

3 回答 3

4

这是我前几天制作的一个示例，它将使用 Netifi 的代理创建一个有状态的 RSocket： https ://github.com/netifi/netifi-stateful-socket

不幸的是，您需要在本地构建我们的开发分支来试用它（https://github.com/netifi/netifi-java） - 如果您不这样做，应该在本周末发布带有代码的版本想在本地构建它。

我也在研究一个纯 RSocket 示例，但如果您想了解它如何查看示例中的 StatefulSocket。它应该为您提供如何使用纯 RSocket 处理会话的线索。

关于您关于事务管理器的其他问题 - 您需要将您的事务与正在发出的 Reactive Streams 信号联系起来 - 如果您收到取消，您将回滚 onError，如果收到 onComplete 您将提交事务. Flux/Mono 有一些副作用方法应该可以轻松处理。根据您在做什么，您还可以使用 BaseSubscriber，因为它具有处理不同 Reactive Streams 信号的钩子。

谢谢，罗伯特

于 2019-05-15T05:47:33.107 回答
2

恢复连接的示例，即维护服务器上的状态，已登陆 rsocket-java 存储库

https://github.com/rsocket/rsocket-java/commit/d47629147dd1a4d41c7c8d5af3d80838e01d3ba5

恢复整个连接，包括与每个单独通道相关的任何状态等。

有一个rsocket-cli项目可以让你尝试一下。启动和停止socat进程，观察客户端和服务器的进度。

$ socat -d TCP-LISTEN:5001,fork,reuseaddr TCP:localhost:5000

$ ./rsocket-cli --debug --resume --server -i cli:time tcp://localhost:5000

$ ./rsocket-cli -i client --stream --resume tcp://localhost:5001
于 2019-05-16T06:17:04.793 回答
0

根据您的描述，它看起来channel效果最好，我以前没有使用过频道，所以我不能保证（对不起）。但我建议你尝试这样的事情：

一个事务控制器：

public class TransactionController implements Publisher<Payload> {

    List<Transaction> transcations = new ArrayList<>();

    @Override
    public void subscribe(Subscriber<? super Payload> subscriber) {

    }

    public void processPayload(Payload payload) {
        // handle transcations...
    }
}

并在您的RSocket实现中覆盖requestChannel：

@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
    // Create new controller for each channel
    TranscationController cntrl = new TranscationController();
    Flux.from(payloads)
      .subscribe(cntrl::processPayload);
    return Flux.from(cntrl);
}
于 2019-05-09T13:48:30.913 回答