Posted on 2010-08-18 18:08
dennis 阅读(4877)
评论(0) 编辑 收藏 所属分类:
java
这个问题的由来是有一个朋友报告xmemcached在高并发下会频繁断开重连，导致cache不可用，具体请看这个
issue。
我一开始以为是他使用方式上有问题，沟通了半天还是搞不明白。后来听闻他说他的代码会中断运行时间过长的任务，这些任务内部调用了xmemcached跟memcached交互。我才开始怀疑是不是因为中断引起了连接的关闭。
我们都知道,nio的socket channel都是实现了
java.nio.channels.InterruptibleChannel接口，看看这个接口描述：
A channel that can be asynchronously closed and interrupted.
A channel that implements this interface is asynchronously closeable: If a thread is blocked in an I/O operation on an interruptible channel then another thread may invoke the channel's close method. This will cause the blocked thread to receive an AsynchronousCloseException.
A channel that implements this interface is also
interruptible: If a thread is blocked in an I/O operation on an interruptible channel then another thread may invoke the blocked thread's interrupt method.
This will cause the channel to be closed, the blocked thread to receive a
ClosedByInterruptException, and the blocked thread's interrupt status to be set.
If a thread's interrupt status is already set and it invokes a blocking I/O operation upon a channel then the channel will be closed and the thread will immediately receive a ClosedByInterruptException; its interrupt status will remain set.
意思是说实现了这个接口的channel，首先可以被异步关闭，阻塞的线程抛出AsynchronousCloseException，其次阻塞在该 channel上的线程如果被中断，会引起channel关闭并抛出ClosedByInterruptException的异常。如果在调用 channel的IO方法之前，线程已经设置了中断状态，同样会引起channel关闭和抛出ClosedByInterruptException。
回到xmemcached的问题，为什么中断会引起xmemcached关闭连接？难道xmemcached会在用户线程调用channel的IO operations。答案是肯定的，xmemcached的网络层实现了一个小优化，当连接里的缓冲队列为空的情况下，会直接调用 channel.write尝试发送数据；如果队列不为空，则放入缓冲队列，等待Reactor去执行实际的发送工作，这个优化是为了最大化地提高发送效率。这会导致在用户线程中调用channel.write（缓冲的消息队列为空的时候），如果这时候用户线程中断，就会导致连接断开，这就是那位朋友反馈的问题的根源。
Netty3的早期版本也有同样的优化，但是在之后的版本，这个优化被另一个方案替代，写入消息的时候无论如何都会放入缓冲队列，但是Netty会判断当前写入的线程是不是NioWorker， 如果是的话，就直接flush整个发送队列做IO写入，如果不是，则加入发送缓冲区等待NioWorker线程去发送。这个是在NioWorker的 writeFromUserCode方法里实现的：
void writeFromUserCode(final NioSocketChannel channel) {
if (!channel.isConnected()) {
cleanUpWriteBuffer(channel);
return;
}
if (scheduleWriteIfNecessary(channel)) {
return;
}
// 这里我们可以确认 Thread.currentThread() == workerThread.
if (channel.writeSuspended) {
return;
}
if (channel.inWriteNowLoop) {
return;
}
write0(channel);
}
我估计netty的作者后来也意识到了在用户线程调用channel的IO操作的危险性。xmemcached这个问题的解决思路也应该跟Netty差不多。但是从我的角度，我希望交给用户去选择，如果你确认你的用户线程没有调用中断，那么允许在用户线程去write可以达到更高的发送效率，更短的响应时间；如果你的用户线程有调用中断，那么最好有个选项去设置，发送的消息都将加入缓冲队列让Reactor去写入，有更好的吞吐量，同时避免用户线程涉及到 IO操作。