java - nio上下文中的请求-响应同步/匹配

Question

我有一个服务器客户端应用程序(Java EE 和 Android)，通过 websockets 进行通信。通信正常，协议(protocol)本身也可以将对象作为 json 发送，这些对象将被正确包装、序列化、发送、反序列化、解包和重建。这两个应用程序都在使用另一个库项目，其中包含所有可能的请求和响应类。

现在解决我的问题:图书馆还应该实现非阻塞通信策略，但透明的请求-响应实现。可能我不是第一个遇到这个问题的人，所以我认为那里可能有一些不错的实现:)。

我想要的:

// server should sleep 5000ms and then return 3*3
Future<Integer> f1 = server.put(
  new SleepAndReturnSquareRequest(5000, 3),
  new FutureCallback<Integer>{
    public void onSuccess(Integer square) {
      runOnUiThread(new Runnable{
        // Android Toast show square
      });
    }

    // impl onFailure
  }
);

Future<Date> f2 = server.put(
  new TimeRequest(),
  new FutureCallback<Date>{
    public void onSuccess(Date date) {
      // called before other onSuccess
    }

    // impl onFailure
  }
);

// e.g. when the activity in android changes I'll cancel all futures. That means no more callbacks and (later) if possible client sends cancellations to the server for long requests.

代码应该发送一个 SleepAndReturnRequest 然后一个 TimeRequest 当然是非阻塞的。第一个请求需要 5 秒，第二个请求几乎为零毫秒。我希望实现在收到响应后立即调用第二个回调，而在大约 5 秒后调用第一个回调。实现负责请求方的请求-响应匹配。

我的尝试和思考:

Google 的 guava listenable future 我认为对于“响应方”来说是一个很好的方法，因为它只是一个在任何线程上运行的任务，最后将结果发回。那应该更容易。

对于“请求方”，我需要一些实现，它向消息添加唯一标识符以便能够匹配响应。希望您能告诉我一些可以完成这项工作的软件包。

感谢您的帮助。

//编辑:

我认为我的问题被误解了或者不够精确。想想如何通过 websocket 实现 GET 或 POST。每个 GET/POST 请求都有一个响应，然后关闭连接。客户端连接到特定端口，服务器从线程池中获取线程，处理请求并响应。我认为请求与响应的匹配是在传输层 #4 中完成的。

因为我想使用 websockets，所以我必须在软件层 7 中实现匹配。

以下是我正在实现的一些步骤。 K是唯一键类型，V是消息内容的通用类型。这可能是一个字符串、字节流，等等。

public class Synchronizer<K, V> implements UniqueMessageListener<K, V> {

    private final ConcurrentMap<K, FutureCallback<V>> callbackMap = new ConcurrentHashMap<>();

    private final ListeningExecutorService executor;

    private final UniqueMessageFactory<K, V> factory;
    private final UniqueMessageSender<K, V> sender;
    private UniqueMessageReceiver<K, V> receiver = null;

    public Synchronizer(
            ListeningExecutorService executor,
            UniqueMessageFactory<K, V> factory,
            UniqueMessageSender<K, V> sender
    ) {
        this.executor = executor;
        this.factory = factory;
        this.sender = sender;
    }

    public void register(UniqueMessageReceiver<K, V> receiver) {

        unregister();

        this.receiver = receiver;
        receiver.addListener(this);
    }

    public void unregister() {
        if(receiver != null) {
            receiver.removeListener(this);
            receiver = null;
        }
    }

    public Future<V> put(Message<V> message, final FutureCallback<V> callback) {
        final UniqueMessage<K, V> uniqueMessage = factory.create(message);

        final Future<Boolean> sendFuture = sender.send(uniqueMessage);

        final ListenableFuture<Boolean> listenableSendFuture =
                JdkFutureAdapters.listenInPoolThread(sendFuture, executor);


        listenableSendFuture.addListener(
                new Runnable() {
                    @Override
                    public void run() {
                        try {
                            if(listenableSendFuture.get() == true) {
                                callbackMap.put(
                                        uniqueMessage.getId(),
                                        callback
                                );
                            } else {
                                // maybe try it later again?
                            }
                        } catch(Exception e) {
                            // ...
                        }
                    }
                },
                executor
        );

        // implement cancel
        return new SynchronizeFuture<>(
                listenableSendFuture,
                callback
        );
    }

    @Override
    public void onReceive(UniqueMessage<K, V> message) {
        K id = message.getId();
        FutureCallback<V> callback;

        callback = callbackMap.remove(id);

        if(callback != null) {
            callback.onSuccess(message.getContent());
        }
    }
}

对我来说有很多要测试的，但我认为它会起作用。

Answer 1

你可以试试 ProtoBuf-RPC-Pro - 但它不适用于 WebSockets，尽管我认为 GitHub 上有一个项目支持它 :)