下面列出了java.nio.channels.ClosedSelectorException#java.nio.channels.CancelledKeyException 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
public final void sendPacket(final SendablePacket<T> sp)
{
sp._client = _client;
if (_pendingClose)
return;
synchronized (getSendQueue())
{
_sendQueue.addLast(sp);
}
if (!_sendQueue.isEmpty())
{
try
{
_selectionKey.interestOps(_selectionKey.interestOps() | SelectionKey.OP_WRITE);
}
catch (CancelledKeyException e)
{
// ignore
}
}
}
private void processDebuggerActivity(SelectionKey key) {
Debugger dbg = (Debugger)key.attachment();
try {
if (key.isAcceptable()) {
try {
acceptNewDebugger(dbg, null);
} catch (IOException ioe) {
Log.w("ddms", "debugger accept() failed");
ioe.printStackTrace();
}
} else if (key.isReadable()) {
processDebuggerData(key);
} else {
Log.d("ddm-debugger", "key in unknown state");
}
} catch (CancelledKeyException cke) {
// key has been cancelled we can ignore that.
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
Call call = (Call)key.attachment();
if (call == null) {
return;
}
if (key.channel() != call.connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
synchronized(call.connection.responseQueue) {
if (processResponse(call.connection.responseQueue, false)) {
try {
key.interestOps(0);
} catch (CancelledKeyException e) {
/* The Listener/reader might have closed the socket.
* We don't explicitly cancel the key, so not sure if this will
* ever fire.
* This warning could be removed.
*/
LOG.warn("Exception while changing ops : " + e);
}
}
}
}
private boolean selectionInterestRequiresUpdate(NonBlockingConnection connection)
{
SelectionTask selectionTask = connection.getSelectionTask();
if(selectionTask != null)
{
final SelectionKey selectionKey = connection.getSocketChannel().keyFor(selectionTask.getSelector());
int expectedOps = (connection.wantsRead() ? SelectionKey.OP_READ : 0)
| (connection.wantsWrite() ? SelectionKey.OP_WRITE : 0);
try
{
return selectionKey == null || !selectionKey.isValid() || selectionKey.interestOps() != expectedOps;
}
catch (CancelledKeyException e)
{
return true;
}
}
else
{
return true;
}
}
private void disable()
{
if (m_enabled)
{
try
{
m_key.interestOps(0); // pass zero which means that we are not interested in being
// notified of anything for this channel.
}
catch (CancelledKeyException eat) // If we finished writing and didn't get an exception, then
{ // we probably don't need to throw this exception (if they try to write
// again, we will then throw an exception).
}
m_enabled = false;
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
Call call = (Call)key.attachment();
if (call == null) {
return;
}
if (key.channel() != call.connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
synchronized(call.connection.responseQueue) {
if (processResponse(call.connection.responseQueue, false)) {
try {
key.interestOps(0);
} catch (CancelledKeyException e) {
/* The Listener/reader might have closed the socket.
* We don't explicitly cancel the key, so not sure if this will
* ever fire.
* This warning could be removed.
*/
LOG.warn("Exception while changing ops : " + e);
}
}
}
}
private void disable()
{
if (m_enabled)
{
try
{
m_key.interestOps(0); // pass zero which means that we are not interested in being
// notified of anything for this channel.
}
catch (CancelledKeyException eat) // If we finished writing and didn't get an exception, then
{ // we probably don't need to throw this exception (if they try to write
// again, we will then throw an exception).
}
m_enabled = false;
}
}
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
private void doConnect(long now, Set<SelectionKey> keys) {
Iterator<SelectionKey> iter = keys.iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
try {
if (key.isConnectable()) {
MqttChannel channel = (MqttChannel) key.attachment();
if (!channel.finishConnect()) {
channelClosed(channel);
iter.remove();
}
}
} catch (CancelledKeyException e) {
iter.remove();
}
}
}
private void doRead(long now, Set<SelectionKey> keys) {
Iterator<SelectionKey> iter = keys.iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
try {
if (key.isReadable()) {
MqttChannel channel = (MqttChannel) key.attachment();
if (!channel.read(now)) {
channelClosed(channel);
iter.remove();
}
}
} catch (CancelledKeyException e) {
iter.remove();
}
}
}
private void doWrite(long now, Set<SelectionKey> keys) {
Iterator<SelectionKey> iter = keys.iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
try {
if (key.isWritable()) {
MqttChannel channel = (MqttChannel) key.attachment();
if (!channel.write(now)) {
channelClosed(channel);
iter.remove();
}
}
} catch (CancelledKeyException e) {
iter.remove();
}
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
SimpleServerRpcConnection connection = (SimpleServerRpcConnection) key.attachment();
if (connection == null) {
throw new IOException("doAsyncWrite: no connection");
}
if (key.channel() != connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
if (processAllResponses(connection)) {
try {
// We wrote everything, so we don't need to be told when the socket is ready for
// write anymore.
key.interestOps(0);
} catch (CancelledKeyException e) {
/*
* The Listener/reader might have closed the socket. We don't explicitly cancel the key, so
* not sure if this will ever fire. This warning could be removed.
*/
SimpleRpcServer.LOG.warn("Exception while changing ops : " + e);
}
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
Call call = (Call)key.attachment();
if (call == null) {
return;
}
if (key.channel() != call.connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
synchronized(call.connection.responseQueue) {
if (processResponse(call.connection.responseQueue, false)) {
try {
key.interestOps(0);
} catch (CancelledKeyException e) {
/* The Listener/reader might have closed the socket.
* We don't explicitly cancel the key, so not sure if this will
* ever fire.
* This warning could be removed.
*/
LOG.warn("Exception while changing ops : " + e);
}
}
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
Call call = (Call) key.attachment();
if (call == null) {
return;
}
if (key.channel() != call.connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
synchronized (call.connection.responseQueue) {
if (processResponse(call.connection.responseQueue, false)) {
try {
key.interestOps(0);
} catch (CancelledKeyException e) {
/*
* The Listener/reader might have closed the socket.
* We don't explicitly cancel the key, so not sure if this will
* ever fire.
* This warning could be removed.
*/
LOG.warn("Exception while changing ops : " + e);
}
}
}
}
private void doAsyncWrite(SelectionKey key) throws IOException {
Call call = (Call)key.attachment();
if (call == null) {
return;
}
if (key.channel() != call.connection.channel) {
throw new IOException("doAsyncWrite: bad channel");
}
synchronized(call.connection.responseQueue) {
if (processResponse(call.connection.responseQueue, false)) {
try {
key.interestOps(0);
} catch (CancelledKeyException e) {
/* The Listener/reader might have closed the socket.
* We don't explicitly cancel the key, so not sure if this will
* ever fire.
* This warning could be removed.
*/
LOG.warn("Exception while changing ops : " + e);
}
}
}
}
protected void registerForRead(final SelectionKey key, ObjectReader reader) {
if ( log.isTraceEnabled() )
log.trace("Adding key for read event:"+key);
reader.finish();
//register our OP_READ interest
Runnable r = new Runnable() {
@Override
public void run() {
try {
if (key.isValid()) {
// cycle the selector so this key is active again
key.selector().wakeup();
// resume interest in OP_READ, OP_WRITE
int resumeOps = key.interestOps() | SelectionKey.OP_READ;
key.interestOps(resumeOps);
if ( log.isTraceEnabled() )
log.trace("Registering key for read:"+key);
}
} catch (CancelledKeyException ckx ) {
NioReceiver.cancelledKey(key);
if ( log.isTraceEnabled() )
log.trace("CKX Cancelling key:"+key);
} catch (Exception x) {
log.error(sm.getString("nioReplicationTask.error.register.key", key),x);
}
}
};
receiver.addEvent(r);
}
@Override
public void run() {
SelectionKey sk = ch.keyFor(selector);
try {
if (sk == null) {
if (SelectionKey.OP_WRITE==(ops&SelectionKey.OP_WRITE)) countDown(key.getWriteLatch());
if (SelectionKey.OP_READ==(ops&SelectionKey.OP_READ))countDown(key.getReadLatch());
} else {
if (sk.isValid()) {
sk.interestOps(sk.interestOps() & (~ops));
if (SelectionKey.OP_WRITE==(ops&SelectionKey.OP_WRITE)) countDown(key.getWriteLatch());
if (SelectionKey.OP_READ==(ops&SelectionKey.OP_READ))countDown(key.getReadLatch());
if (sk.interestOps()==0) {
sk.cancel();
sk.attach(null);
}
}else {
sk.cancel();
sk.attach(null);
}
}
}catch (CancelledKeyException cx) {
if (sk!=null) {
sk.cancel();
sk.attach(null);
}
}
}
@Override
public final void onWriteDemand(WebSocket w) {
WebSocketImpl conn = (WebSocketImpl) w;
try {
conn.key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
} catch (CancelledKeyException e) {
// the thread which cancels key is responsible for possible cleanup
conn.outQueue.clear();
}
selector.wakeup();
}
public void putEventOps(SelectionKeyImpl sk, int ops) {
synchronized (closeLock) {
if (pollWrapper == null)
throw new ClosedSelectorException();
// make sure this sk has not been removed yet
int index = sk.getIndex();
if (index == -1)
throw new CancelledKeyException();
pollWrapper.putEventOps(index, ops);
}
}
@Override
protected void doRegister() throws Exception {
// 渠道监听状态
boolean selected = false;
for (;;) {
try {
// 通道注册到选择器上
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.//强制选择器现在选择，因为“已取消”的SelectionKey可能仍然是
//缓存，未删除，因为没有Select.select(..)操作被调用。
// 开始监听
eventLoop().selectNow();
// 修改选择器的监听状态
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?//我们之前在选择器上强制执行了select操作，但是SelectionKey仍然被缓存
//不管什么原因。JDK错误?
throw e;
}
}
}
}
public void putEventOps(SelectionKeyImpl sk, int ops) {
synchronized (closeLock) {
if (pollWrapper == null)
throw new ClosedSelectorException();
// make sure this sk has not been removed yet
int index = sk.getIndex();
if (index == -1)
throw new CancelledKeyException();
pollWrapper.putEventOps(index, ops);
}
}
public void putEventOps(SelectionKeyImpl sk, int ops) {
synchronized (closeLock) {
if (pollWrapper == null)
throw new ClosedSelectorException();
// make sure this sk has not been removed yet
int index = sk.getIndex();
if (index == -1)
throw new CancelledKeyException();
pollWrapper.putEventOps(index, ops);
}
}
private void closeConnection(SelectionKey currentKey, boolean allowReconnect, boolean shuttingDown)
{
Peer client = (TcpConnectorPeer) currentKey.attachment();
if (client != null)
{
connObserver.connectionClosed(client, allowReconnect, shuttingDown);
try
{
if (client.isConnected(false))
{
client.connectionClosing();
}
}
catch (CancelledKeyException ignored)
{
// connectionClosing() calls interestOps on the selection Key, which may fail
}
}
try
{
SelectableChannel channel = currentKey.channel();
if (channel != null)
{
channel.close();
}
}
catch (IOException closeIoExc)
{
// If close() fails with an I/O error, the reason may be interesting
// enough to file an error report
errorReporter.reportError(closeIoExc);
}
currentKey.cancel();
}
public void putEventOps(SelectionKeyImpl sk, int ops) {
synchronized (closeLock) {
if (pollWrapper == null)
throw new ClosedSelectorException();
// make sure this sk has not been removed yet
int index = sk.getIndex();
if (index == -1)
throw new CancelledKeyException();
pollWrapper.putEventOps(index, ops);
}
}
public final void close(final SendablePacket<T>[] closeList)
{
if (_pendingClose)
return;
synchronized (getSendQueue())
{
if (!_pendingClose)
{
_pendingClose = true;
_sendQueue.clear();
for (SendablePacket<T> sp : closeList)
_sendQueue.addLast(sp);
}
}
try
{
_selectionKey.interestOps(_selectionKey.interestOps() & ~SelectionKey.OP_WRITE);
}
catch (CancelledKeyException e)
{
// ignore
}
// _closePacket = sp;
_selectorThread.closeConnection(this);
}
protected void registerForRead(final SelectionKey key, ObjectReader reader) {
if ( log.isTraceEnabled() )
log.trace("Adding key for read event:"+key);
reader.finish();
//register our OP_READ interest
Runnable r = new Runnable() {
@Override
public void run() {
try {
if (key.isValid()) {
// cycle the selector so this key is active again
key.selector().wakeup();
// resume interest in OP_READ, OP_WRITE
int resumeOps = key.interestOps() | SelectionKey.OP_READ;
key.interestOps(resumeOps);
if ( log.isTraceEnabled() )
log.trace("Registering key for read:"+key);
}
} catch (CancelledKeyException ckx ) {
NioReceiver.cancelledKey(key);
if ( log.isTraceEnabled() )
log.trace("CKX Cancelling key:"+key);
} catch (Exception x) {
log.error("Error registering key for read:"+key,x);
}
}
};
receiver.addEvent(r);
}
protected void socketTimeouts() {
long now = System.currentTimeMillis();
if ( (now-lastCheck) < getSelectorTimeout() ) return;
//timeout
Selector tmpsel = this.selector.get();
Set<SelectionKey> keys = (isListening()&&tmpsel!=null)?tmpsel.keys():null;
if ( keys == null ) return;
for (Iterator<SelectionKey> iter = keys.iterator(); iter.hasNext();) {
SelectionKey key = iter.next();
try {
// if (key.interestOps() == SelectionKey.OP_READ) {
// //only timeout sockets that we are waiting for a read from
// ObjectReader ka = (ObjectReader) key.attachment();
// long delta = now - ka.getLastAccess();
// if (delta > (long) getTimeout()) {
// cancelledKey(key);
// }
// }
// else
if ( key.interestOps() == 0 ) {
//check for keys that didn't make it in.
ObjectReader ka = (ObjectReader) key.attachment();
if ( ka != null ) {
long delta = now - ka.getLastAccess();
if (delta > getTimeout() && (!ka.isAccessed())) {
if (log.isWarnEnabled())
log.warn("Channel key is registered, but has had no interest ops for the last "+getTimeout()+" ms. (cancelled:"+ka.isCancelled()+"):"+key+" last access:"+new java.sql.Timestamp(ka.getLastAccess())+" Possible cause: all threads used, perform thread dump");
ka.setLastAccess(now);
//key.interestOps(SelectionKey.OP_READ);
}//end if
} else {
cancelledKey(key);
}//end if
}//end if
}catch ( CancelledKeyException ckx ) {
cancelledKey(key);
}
}
lastCheck = System.currentTimeMillis();
}
public void remove(final KeyAttachment key, final int ops) {
Runnable r = new Runnable() {
@Override
public void run() {
if ( key == null ) return;
NioChannel nch = key.getChannel();
if ( nch == null ) return;
SocketChannel ch = nch.getIOChannel();
if ( ch == null ) return;
SelectionKey sk = ch.keyFor(selector);
try {
if (sk == null) {
if (SelectionKey.OP_WRITE==(ops&SelectionKey.OP_WRITE)) countDown(key.getWriteLatch());
if (SelectionKey.OP_READ==(ops&SelectionKey.OP_READ))countDown(key.getReadLatch());
} else {
if (sk.isValid()) {
sk.interestOps(sk.interestOps() & (~ops));
if (SelectionKey.OP_WRITE==(ops&SelectionKey.OP_WRITE)) countDown(key.getWriteLatch());
if (SelectionKey.OP_READ==(ops&SelectionKey.OP_READ))countDown(key.getReadLatch());
if (sk.interestOps()==0) {
sk.cancel();
sk.attach(null);
}
}else {
sk.cancel();
sk.attach(null);
}
}
}catch (CancelledKeyException cx) {
if (sk!=null) {
sk.cancel();
sk.attach(null);
}
}
}
};
events.offer(r);
wakeup();
}
public void putEventOps(SelectionKeyImpl sk, int ops) {
synchronized (closeLock) {
if (pollWrapper == null)
throw new ClosedSelectorException();
// make sure this sk has not been removed yet
int index = sk.getIndex();
if (index == -1)
throw new CancelledKeyException();
pollWrapper.putEventOps(index, ops);
}
}
protected void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
if (!lifecycle.started()) {
// ignore
return;
}
if (isCloseConnectionException(e.getCause())) {
logger.trace("close connection exception caught on transport layer [{}], disconnecting from relevant node", e.getCause(), ctx.getChannel());
// close the channel, which will cause a node to be disconnected if relevant
ctx.getChannel().close();
disconnectFromNodeChannel(ctx.getChannel(), e.getCause());
} else if (isConnectException(e.getCause())) {
logger.trace("connect exception caught on transport layer [{}]", e.getCause(), ctx.getChannel());
// close the channel as safe measure, which will cause a node to be disconnected if relevant
ctx.getChannel().close();
disconnectFromNodeChannel(ctx.getChannel(), e.getCause());
} else if (e.getCause() instanceof CancelledKeyException) {
logger.trace("cancelled key exception caught on transport layer [{}], disconnecting from relevant node", e.getCause(), ctx.getChannel());
// close the channel as safe measure, which will cause a node to be disconnected if relevant
ctx.getChannel().close();
disconnectFromNodeChannel(ctx.getChannel(), e.getCause());
} else if (e.getCause() instanceof SizeHeaderFrameDecoder.HttpOnTransportException) {
// in case we are able to return data, serialize the exception content and sent it back to the client
if (ctx.getChannel().isOpen()) {
ChannelBuffer buffer = ChannelBuffers.wrappedBuffer(e.getCause().getMessage().getBytes(Charsets.UTF_8));
ChannelFuture channelFuture = ctx.getChannel().write(buffer);
channelFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
future.getChannel().close();
}
});
}
} else {
logger.warn("exception caught on transport layer [{}], closing connection", e.getCause(), ctx.getChannel());
// close the channel, which will cause a node to be disconnected if relevant
ctx.getChannel().close();
disconnectFromNodeChannel(ctx.getChannel(), e.getCause());
}
}