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

Question

我有一个服务器客户端应用程序(Java EE和Android),通过websockets进行通信.通信工作以及协议本身用于以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秒后调用.该实现负责请求方的请求 - 响应匹配.

我试过和想过的:

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 上有一个项目支持它:)