java.nio.channels.AsynchronousSocketChannel#read()源码实例Demo

@Override
public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
    // keep listening
    serverSocketChannel.accept(attachment, this);

    ProxyConnectionState clientState = new ProxyConnectionState();
    clientState.readChannel = clientChannel;
    clientState.buffer = ByteBuffer.allocate(4096);
    clientState.handshakePhase = HandshakePhase.INITIAL;

    try {
        clientChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
    } catch (IOException e) {
        LOG.error("Failed to set TCP_NODELAY after accept, closing channel", e);
        closeChannel(clientChannel);
        return;
    }

    // read from readChannel
    clientChannel.read(clientState.buffer, TIMEOUT_IN_S, TimeUnit.SECONDS, clientState, readHandler);
}
 

/***************************************
 * {@inheritDoc}
 */
@Override
protected boolean performAsyncOperation(
	int													nBytesReceived,
	AsynchronousSocketChannel							rChannel,
	ByteBuffer											rData,
	ChannelCallback<Integer, AsynchronousSocketChannel> rCallback)
	throws IOException
{
	boolean bFinished = false;

	if (nBytesReceived >= 0)
	{
		bFinished = pCheckFinished.test(nBytesReceived, rData);
	}

	if (nBytesReceived != -1 && !bFinished && rData.hasRemaining())
	{
		rChannel.read(rData, rData, rCallback);
	}
	else
	{
		checkErrors(rData, nBytesReceived, bFinished);
		rData.flip();
	}

	return bFinished;
}
 

@Override
public void completed(AsynchronousSocketChannel channel, Server server) {
  try {
    if (log.isDebugEnabled()) {
      log.debug("Aio accept {}", channel);
    }
    if (!channel.isOpen()) {
      log.warn("channel has closed {}", channel);
      return;
    }
    channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

    ChannelContext channelContext = new ChannelContext(server.getGroupContext(), channel);
    channelContext.setServerAddress(server.getServerAddress());
    channelContext.start();

    ByteBuffer buffer = ByteBuffer.allocate(server.getGroupContext().getBufferCapacity());
    buffer.order(ByteOrder.BIG_ENDIAN);
    channel.read(buffer, buffer, channelContext.getReadHandler());

    if (server.getGroupContext().getAioListener() != null) {
      server.getGroupContext().getAioListener().onConnected(channelContext);
    }
  } catch (Exception e) {
    log.error("Aio accept completed error", e);
  } finally {
    server.getServerChannel().accept(server, this);
  }
}
 

@Override
public void completed(Integer result, ByteBuffer byteBuffer)
{
	GroupContext<SessionContext, P, R> groupContext = channelContext.getGroupContext();
	if (result > 0)
	{
		ByteBuffer newByteBuffer = ByteBufferUtils.copy(readByteBuffer, 0, readByteBuffer.position());
		DecodeRunnable<SessionContext, P, R> decodeRunnable = channelContext.getDecodeRunnable();
		decodeRunnable.addMsg(newByteBuffer);
		groupContext.getDecodeExecutor().execute(decodeRunnable);
	} else if (result == 0)
	{
		log.error("读到的数据长度为0");
	} else if (result < 0)
	{
		Aio.close(channelContext, null, "读数据时返回" + result);
	}

	if (AioUtils.checkBeforeIO(channelContext))
	{
		AsynchronousSocketChannel asynchronousSocketChannel = channelContext.getAsynchronousSocketChannel();
		readByteBuffer.position(0);
		readByteBuffer.limit(readByteBuffer.capacity());
		asynchronousSocketChannel.read(readByteBuffer, readByteBuffer, this);
	}

}
 

@Override
public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
    attachment.asyncServerSocketChannel.accept(attachment, this);
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    result.read(buffer, buffer, new ReadCompletionHandler(result));
}
 

/** 
 * @see java.nio.channels.CompletionHandler#completed(java.lang.Object, java.lang.Object)
 * 
 * @param result
 * @param attachment
 * @重写人: tanyaowu
 * @重写时间: 2017年2月26日 下午9:39:18
 * 
 */
@Override
public void completed(Void result, ConnectionCompletionVo<SessionContext, P, R> attachment)
{
	synchronized (attachment)
	{
		try
		{
			boolean isReconnect = attachment.isReconnect();
			ClientChannelContext<SessionContext, P, R> channelContext = attachment.getChannelContext();
			AsynchronousSocketChannel asynchronousSocketChannel = attachment.getAsynchronousSocketChannel();
			AioClient<SessionContext, P, R> aioClient = attachment.getAioClient();
			ClientGroupContext<SessionContext, P, R> clientGroupContext = aioClient.getClientGroupContext();
			Node serverNode = attachment.getServerNode();
			String bindIp = attachment.getBindIp();
			Integer bindPort = attachment.getBindPort();
			ClientAioListener<SessionContext, P, R> clientAioListener = clientGroupContext.getClientAioListener();

			if (isReconnect)
			{
				channelContext.setAsynchronousSocketChannel(asynchronousSocketChannel);
				channelContext.getDecodeRunnable().setCanceled(false);
				channelContext.getHandlerRunnableNormPrior().setCanceled(false);
				//		channelContext.getHandlerRunnableHighPrior().setCanceled(false);
				channelContext.getSendRunnableNormPrior().setCanceled(false);
				//		channelContext.getSendRunnableHighPrior().setCanceled(false);

				clientGroupContext.getCloseds().remove(channelContext);
			} else
			{
				channelContext = new ClientChannelContext<>(clientGroupContext, asynchronousSocketChannel);
				channelContext.setServerNode(serverNode);
				channelContext.getStat().setTimeClosed(SystemTimer.currentTimeMillis());
			}

			channelContext.setBindIp(bindIp);
			channelContext.setBindPort(bindPort);

			channelContext.setReconnCount(0);
			channelContext.setClosed(false);
			
			attachment.setChannelContext(channelContext);

			clientGroupContext.getConnecteds().add(channelContext);

			ReadCompletionHandler<SessionContext, P, R> readCompletionHandler = channelContext.getReadCompletionHandler();
			ByteBuffer readByteBuffer = readCompletionHandler.getReadByteBuffer();//ByteBuffer.allocateDirect(channelContext.getGroupContext().getReadBufferSize());
			readByteBuffer.position(0);
			readByteBuffer.limit(readByteBuffer.capacity());
			asynchronousSocketChannel.read(readByteBuffer, readByteBuffer, readCompletionHandler);

			log.info("connected to {}", serverNode);
			
			try
			{
				clientAioListener.onAfterConnected(channelContext, !channelContext.isClosed(), isReconnect);
			} catch (Exception e1)
			{
				log.error(e1.toString(), e1);
			}
		} catch (Exception e)
		{
			log.error(e.toString(), e);
		}
		
		attachment.notify();
	}
	
	
}
 

void read(AsynchronousSocketChannel socketChannel, String sessionId, AioListener listener) {
    ByteBuffer buffer = aioHelper.getBuffer(sessionId);
    buffer.clear();
    socketChannel.read(buffer, null, BeanFactory.getBean(ReceiveHandler.class).bind(socketChannel, buffer, sessionId, listener));
}
 

/**
 * Synchronously read data from the server. (Needed here for TLS handshake)
 * 
 * @param channel
 *            {@link AsynchronousSocketChannel}
 * @param data
 *            {@link ByteBuffer}
 * @return the number of bytes read
 */
private static Integer read(AsynchronousSocketChannel channel, ByteBuffer data) {
    Future<Integer> f = channel.read(data);
    try {
        return f.get();
    } catch (InterruptedException | ExecutionException ex) {
        throw new CJCommunicationsException(ex);
    }
}
 

/**
 * Synchronously read data from the server. (Needed here for TLS handshake)
 * 
 * @param channel
 *            {@link AsynchronousSocketChannel}
 * @param data
 *            {@link ByteBuffer}
 * @return the number of bytes read
 */
private static Integer read(AsynchronousSocketChannel channel, ByteBuffer data) {
    Future<Integer> f = channel.read(data);
    try {
        return f.get();
    } catch (InterruptedException | ExecutionException ex) {
        throw new CJCommunicationsException(ex);
    }
}