MINA：执行同步写入请求/读取响应

Question

我正在尝试使用 MINA 2.0 RC1 在基于 demux 的客户端应用程序中执行同步写入/读取，但它似乎被卡住了。这是我的代码：

public boolean login(final String username, final String password) {
    // block inbound messages
    session.getConfig().setUseReadOperation(true);

    // send the login request
    final LoginRequest loginRequest = new LoginRequest(username, password);
    final WriteFuture writeFuture = session.write(loginRequest);
    writeFuture.awaitUninterruptibly();

    if (writeFuture.getException() != null) {
        session.getConfig().setUseReadOperation(false);
        return false;
    }

    // retrieve the login response
    final ReadFuture readFuture = session.read();
    readFuture.awaitUninterruptibly();

    if (readFuture.getException() != null) {
        session.getConfig().setUseReadOperation(false);
        return false;
    }

    // stop blocking inbound messages
    session.getConfig().setUseReadOperation(false);

    // determine if the login info provided was valid
    final LoginResponse loginResponse = (LoginResponse)readFuture.getMessage();
    return loginResponse.getSuccess();
}

我可以在服务器端看到 LoginRequest 对象被检索，并且 LoginResponse 消息被发送。在客户端，DemushingProtocolCodecFactory 接收响应，但在添加一些日志记录后，我可以看到客户端卡在对 readFuture.awaitUninterruptically() 的调用上。

我一生都无法根据我自己的代码弄清楚为什么它被困在这里。我在会话配置中正确地将读取操作设置为 true，这意味着应该阻止消息。但是，当我尝试同步读取响应消息时，该消息似乎不再存在。

有什么线索可以解释为什么这对我不起作用吗？

Answer 1

这对我不起作用的原因是因为我的代码中其他地方存在问题，我愚蠢地忽略了实现消息响应编码器/解码器。啊。不管怎样，我问题中的代码在我解决这个问题后就立即生效了。

Answer 2

我更喜欢这个（克里斯蒂安·穆勒：http://apache-mina.10907.n7.nabble.com/Mina-Client-which-sends-receives-messages-synchronous-td35672.html)

public class UCPClient { 

private Map<Integer, BlockingQueue<UCPMessageResponse>> concurrentMap = new ConcurrentHashMap<Integer, BlockingQueue<UCPMessageResponse>>(); 

// some other code 

public UCPMessageResponse send(UCPMessageRequest request) throws Throwable { 
    BlockingQueue<UCPMessageResponse> queue = new LinkedBlockingQueue<UCPMessageResponse>(1); 
    UCPMessageResponse res = null; 

    try { 
        if (sendSync) { 
            concurrentMap.put(Integer.valueOf(request.getTransactionReference()), queue); 
        } 

        WriteFuture writeFuture = session.write(request); 

        if (sendSync) { 
            boolean isSent = writeFuture.await(transactionTimeout, TimeUnit.MILLISECONDS); 

            if (!isSent) { 
                throw new TimeoutException("Could not sent the request in " + transactionTimeout + " milliseconds."); 

            } 

            if (writeFuture.getException() != null) { 
                throw writeFuture.getException(); 
            } 

            res = queue.poll(transactionTimeout, TimeUnit.MILLISECONDS); 

            if (res == null) { 
                throw new TimeoutException("Could not receive the response in " + transactionTimeout + " milliseconds."); 
            } 
        } 
    } finally { 
        if (sendSync) { 
            concurrentMap.remove(Integer.valueOf(request.getTransactionReference())); 
        } 
    } 

    return res; 
} 

}

和 IoHandler:

public class InnerHandler implements IoHandler { 

// some other code 

public void messageReceived(IoSession session, Object message) throws Exception { 
    if (sendSync) { 
        UCPMessageResponse res = (UCPMessageResponse) message; 
        BlockingQueue<UCPMessageResponse> queue = concurrentMap.get(res.getTransactionReference()); 
        queue.offer(res); 
    } 
} 

}

Answer 3

我遇到了这个问题。事实证明，这是因为我在 IoHandler.sessionCreated() 实现中进行读/写操作。我将处理转移到建立连接的线程上，而不是仅仅等待不久的将来。

Answer 4

您不得在 IoHandler 线程中使用 login() 函数：

如果您在 的覆盖事件函数中调用 IoFuture.awaitUninterruptically() >IoHandler，

IoHandler 不工作并卡住。

您可以在其他线程中调用login()，它将正常工作。