java - 有状态 Rsocket 应用程序

Question

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

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

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

想象一个示例服务和几个客户端。每个客户端都有 Rsocket channel ，服务已启动并正在运行。有没有办法使用 Rsocket channel 在服务器端识别这些客户端？如果您能展示此类行为的编程示例，那就太棒了。谢谢!

编辑(更详细地描述案例)

这是我的例子。

我们目前有三个 CORBA 对象，如图所示:

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

所以在这里我们使用 session 对象在我们的服务上执行交易。

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

Answer 1

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

不幸的是，您需要在本地构建我们的开发分支来试用它 (https://github.com/netifi/netifi-java) - 如果您不想在本地构建它，应该会在本周末发布包含代码的版本。

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

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

谢谢，罗伯特