3

这是有状态 Rsocket 应用程序线程的延续。这是我的例子。

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

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

在此处输入图像描述

所以这里我们使用会话对象在我们的服务上执行事务。为了用 rsocket 替换它，我们编写了一个简单的 protobuf：

message LoginRequest {
    string username = 1;
    string password = 2;
}

message LoginResponse {
    bool success = 1;
}

message Command {
    string command = 1;
}

message TransactionResult {
    string result = 1;
}

service SimpleService {
    rpc login (LoginRequest) returns (LoginResponse) {}
    rpc transaction (stream Command) returns (TransactionResult) {}
}

这个想法是，一旦用户登录，他将开始将命令流式传输到服务器。最后，客户端将提交或删除更改。

所以这里是开始向服务器发出值的客户端代码：

public class SimpleClientWrapper {
    SimpleServiceClient client;
    ...
    public void runClient() {
        SimpleServiceProto.LoginRequest login_request= SimpleServiceProto.LoginRequest.newBuilder().setUsername(this.username).setPassword(this.pass).build();
        System.out.println("Trying to log in...");
        client.login(login_request).doOnNext(response->{
            if(response.getSuccess()) {
                runCommands();
            }else {
                disconnect();
            }
        }).block();
    }

    public void runCommands() {
        System.out.println("Login successfull. About to run some commands.");
        Flux<Command> requests =
                Flux.range(1, 11)
                    .map(i -> "sending -> " + i)
                    .map(s -> Command.newBuilder().setCommand(s).build());


        TransactionResult response = client.transaction(requests).block();

        System.out.println("Result was: " + response.getResult());
    }

    ...
}

如您所见，客户端将在 runClient 方法中登录，如果登录成功，客户端将执行 runCommands 方法，该方法只会发出一些值。

在我的服务器上，根据 protobuf，我创建了事务方法：

    @Override
    public Mono<TransactionResult> transaction(Publisher<Command> messages, ByteBuf metadata) {
        return Flux.from(messages)
                 .windowTimeout(10, Duration.ofSeconds(500))
                 .take(1)
                 .flatMap(Function.identity())
                 .reduce(
                   new ConcurrentHashMap<Character, AtomicInteger>(),
                   (map, s) -> {
                     char[] chars = s.getCommand().toCharArray();
                     for (char c : chars) {
                       map.computeIfAbsent(c, _c -> new AtomicInteger()).incrementAndGet();
                     }

                     return map;
                   })
                 .map(
                   map -> {
                     StringBuilder builder = new StringBuilder();

                     map.forEach(
                       (character, atomicInteger) -> {
                         builder
                           .append("character -> ")
                           .append(character)
                           .append(", count -> ")
                           .append(atomicInteger.get())
                           .append("\n");
                       });

                     String s = builder.toString();

                     return TransactionResult.newBuilder().setResult(s).build();
                   });
    }

现在，当我运行它时，由于客户端上的阻塞错误，它无法工作：

Exception in thread "main" java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-tcp-nio-4

但是我的问题不是它不起作用，而是我希望每个客户端都与服务器进行会话，并且对于每个客户端，服务器应该保持事务状态。这是完整代码的 git 链接：https ://github.com/oe19fyfa/rsocket-clientstream-example

所以在写完所有这些之后，我的问题是：建立事务会话的正确方法是什么？我知道我的代码非常业余，所以我愿意接受任何类型的建议？

4

0 回答 0