5

我正在尝试使用 CompletionHandler 而不是 Futures 在 vert.x 工作线程中的 AsynchronousSocketChannel 上实现单个请求/响应。来自 vert.x 文档：

“Worker Verticle 永远不会被多个线程同时执行。”

所以这是我的代码（不确定我是否 100% 正确处理了套接字 - 请发表评论）：

    // ommitted: asynchronousSocketChannel.open, connect ...

    eventBus.registerHandler(address, new Handler<Message<JsonObject>>() {
        @Override
        public void handle(final Message<JsonObject> event) {
            final ByteBuffer receivingBuffer = ByteBuffer.allocateDirect(2048);
            final ByteBuffer sendingBuffer = ByteBuffer.wrap("Foo".getBytes());

            asynchronousSocketChannel.write(sendingBuffer, 0L, new CompletionHandler<Integer, Long>() {
                public void completed(final Integer result, final Long attachment) {
                    if (sendingBuffer.hasRemaining()) {
                        long newFilePosition = attachment + result;
                        asynchronousSocketChannel.write(sendingBuffer, newFilePosition, this);
                    }

                    asynchronousSocketChannel.read(receivingBuffer, 0L, new CompletionHandler<Integer, Long>() {
                        CharBuffer charBuffer = null;
                        final Charset charset = Charset.defaultCharset();
                        final CharsetDecoder decoder = charset.newDecoder();

                        public void completed(final Integer result, final Long attachment) {
                            if (result > 0) {
                                long p = attachment + result;
                                asynchronousSocketChannel.read(receivingBuffer, p, this);
                            }

                            receivingBuffer.flip();

                            try {
                                charBuffer = decoder.decode(receivingBuffer);
                                event.reply(charBuffer.toString()); // pseudo code
                            } catch (CharacterCodingException e) { }


                        }

                        public void failed(final Throwable exc, final Long attachment) { }
                    });
                }

                public void failed(final Throwable exc, final Long attachment) { }
            });
        }
    });

我在负载测试期间遇到了很多 ReadPendingException 和 WritePendingException，如果在句柄方法中一次真的只有一个线程，这似乎有点奇怪。如果一次只有 1 个线程与 AsynchronousSocketChannel 一起工作，怎么可能读取或写入尚未完全完成？

4

1 回答 1

1

AsynchronousSocketChannel 的处理程序在它们自己的 AsynchronousChannelGroup 上执行，AsynchronousChannelGroup 是 ExecutorService 的派生。除非您做出特别努力，否则处理程序将与启动 I/O 操作的代码并行执行。

要在 Verticle 中执行 I/O 完成处理程序，您必须从该 Verticle 创建并注册一个处理程序，该处理程序执行 AsynchronousSocketChannel 的处理程序现在所做的事情。

AsynchronousSocketChannel 的处理程序应该只将其参数（结果和附件）打包到消息中并将该消息发送到事件总线。

于 2012-09-14T13:23:01.090 回答