【NIO】Chapter 3. Channels

A Channelis a conduit(管道) that transports data efficiently between byte buffers and the entity on the other end of the channel (usually a file or socket). 

 

Basic channel operations

checking to see if a channel is open (isOpen()) and closing an open channel (close()).

Most, but not all, channels are interruptible. 

Channelare the byte-oriented subinterfaces

Writable-ByteChanneland ReadableByteChannel : channels operate only on byte buffers.

 

two types of channels: file and socket.

FileChannelclass and three socket channel classes:

SocketChannel, ServerSocketChannel, and DatagramChannel.

 

The socket channels have factory methods to create new socket channels directly.

But a FileChannelobject can be obtained only by calling the getChannel()method on an open RandomAccessFile, FileInputStream, or FileOutputStreamobject.

 

Opening Channels

 

SocketChannel sc = SocketChannel.open(); 
sc.connect (new InetSocketAddress ("somehost", someport)); 
ServerSocketChannel ssc = ServerSocketChannel.open(); 
ssc.socket().bind (new InetSocketAddress (somelocalport)); 
DatagramChannel dc = DatagramChannel.open(); 
RandomAccessFile raf = new RandomAccessFile ("somefile", "r"); 
FileChannel fc = raf.getChannel(); 

 

 

Using Channels

Channels can be unidirectionalor bidirectional.

ByteChannel, which extends both ReadableByteChannel and WritableByteChannel.

 

Copy data from one channel to another.

 

package com.java.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

public class ChannelCopy {
	/**
	 * This code copies data from stdin to stdout. Like the 'cat' command, but
	 * without any useful options.
	 */
	public static void main(String[] argv) throws IOException {
		ReadableByteChannel source = Channels.newChannel(System.in);
		WritableByteChannel dest = Channels.newChannel(System.out);
		channelCopy1(source, dest);
		// alternatively, call channelCopy2 (source, dest);
		source.close();
		dest.close();
	}

	/**
	 * Channel copy method 1. This method copies data from the src channel and
	 * writes it to the dest channel until EOF on src. This implementation makes
	 * use of compact() on the temp buffer to pack down the data if the buffer
	 * wasn't fully drained. This may result in data copying, but minimizes
	 * system calls. It also requires a cleanup loop to make sure all the data
	 * gets sent.
	 */
	private static void channelCopy1(ReadableByteChannel src,
			WritableByteChannel dest) throws IOException {
		ByteBuffer buffer = ByteBuffer.allocateDirect(16 * 1024);
		while (src.read(buffer) != -1) {
			// Prepare the buffer to be drained
			buffer.flip();
			// Write to the channel; may block
			dest.write(buffer);
			// If partial transfer, shift remainder down
			// If buffer is empty, same as doing clear()
			buffer.compact();
		}
		// EOF will leave buffer in fill state
		buffer.flip();
		// Make sure that the buffer is fully drained
		while (buffer.hasRemaining()) {
			dest.write(buffer);
		}
	}

	/**
	 * Channel copy method 2. This method performs the same copy, but assures
	 * the temp buffer is empty before reading more data. This never requires
	 * data copying but may result in more systems calls. No post-loop cleanup
	 * is needed because the buffer will be empty when the loop is exited.
	 */
	private static void channelCopy2(ReadableByteChannel src,
			WritableByteChannel dest) throws IOException {
		ByteBuffer buffer = ByteBuffer.allocateDirect(16 * 1024);
		while (src.read(buffer) != -1) {
			// Prepare the buffer to be drained
			buffer.flip();
			// Make sure that the buffer was fully drained
			while (buffer.hasRemaining()) {
				dest.write(buffer);
			}
			// Make the buffer empty, ready for filling
			buffer.clear();
		}
	}
}

 

 

Closing Channels

Unlike buffers, channels cannot be reused. An open channel represents a specific
connection to a specific I/O service and encapsulates the state of that connection. When a
channel is closed, that connection is lost, and the channel is no longer connected to
anything.

Don't confuse interrupting threads sleeping on Channels with those sleeping on Selectors. The former shuts down the channel; the latter does not. However, your thread's interrupt status will be set if it is interrupted while sleeping on a Selector. If that thread then touches a
Channel, that channel will be closed.

 

 

Scatter/gather channels

 It refers to performing a single I/O operation across multiple buffers.

Scatter/gather should be used with direct ByteBuffers to gain the greatest advantage from native I/O, especially if the buffers are long-lived.

 

read

 

ByteBuffer header = ByteBuffer.allocateDirect (10); 
ByteBuffer body = ByteBuffer.allocateDirect (80); 
ByteBuffer [] buffers = { header, body }; 
int bytesRead = channel.read (buffers);

 send

body.clear(); 
body.put("FOO".getBytes()).flip(); // "FOO" as bytes 
header.clear(); 
header.putShort (TYPE_FILE).putLong (body.limit()).flip(); 
long bytesWritten = channel.write (buffers); 

 

 

 

File channels

FileChannelclass can do normal read and write as well as scatter/gather.

 

File channels are always blocking and cannot be placed into nonblocking mode.

 

For file I/O, the true winner is asynchronous I/O, which lets a process request one or more I/O operationsfrom the operating system but does not wait for them to complete. The process is notified at a later time that the requested I/O has completed.

 

FileChannelobjects cannot be created directly.

A FileChannelinstance can be obtained only by calling getChannel()on an open file object
(RandomAccessFile, FileInputStream, or FileOutputStream).

 

Memory-Mapped Files

The new FileChannelclass provides a method, map(), that establishes a virtual memory
mapping between an open fileand a special type of ByteBuffer.

 

Calling map() on a FileChannelcreates a virtual memory mapping backed by a disk file and wraps a
MappedByteBufferobject around that virtual memory space.

 

To map an entire file:

 

buffer = fileChannel.map (FileChannel.MapMode.READ_ONLY, 0, fileChannel.size()); 

 

 

Like conventional file handles, file mappings can be writable or read-only.

The first two mapping modes, MapMode.READ_ONLY and MapMode.READ_WRITE

 

The third mode, MapMode.PRIVATE, indicates that you want a copy-on-write mapping.
This means that any modifications you make via put()will result in a private copy of the
data that only the MappedByteBufferinstance can see. No changes will be made to the
underlying file, and any changes made will be lost when the buffer is garbage collected.

 

All MappedByteBufferobjects are direct. This means that the memory space they occupy
lives outside the JVM heap.

 

 

package com.java.nio;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URLConnection;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
/** 
* Dummy HTTP server using MappedByteBuffers. 
* Given a filename on the command line, pretend to be 
* a web server and generate an HTTP response containing 
* the file content preceded by appropriate headers. The 
* data is sent with a gathering write. 
*/
public class MappedHttp {
	private static final String OUTPUT_FILE = "MappedHttp.out";
	private static final String LINE_SEP = "\r\n";
	private static final String SERVER_ID = "Server: Ronsoft Dummy Server";
	private static final String HTTP_HDR = "HTTP/1.0 200 OK" + LINE_SEP
			+ SERVER_ID + LINE_SEP;
	private static final String HTTP_404_HDR = "HTTP/1.0 404 Not Found"
			+ LINE_SEP + SERVER_ID + LINE_SEP;
	private static final String MSG_404 = "Could not open file: ";

	public static void main(String[] argv) throws Exception {
		if (argv.length < 1) {
			System.err.println("Usage: filename");
			return;
		}
		String file = argv[0];
		ByteBuffer header = ByteBuffer.wrap(bytes(HTTP_HDR));
		ByteBuffer dynhdrs = ByteBuffer.allocate(128);
		ByteBuffer[] gather = { header, dynhdrs, null };
		String contentType = "unknown/unknown";
		long contentLength = -1;
		try {
			FileInputStream fis = new FileInputStream(file);
			FileChannel fc = fis.getChannel();
			MappedByteBuffer filedata = fc.map(MapMode.READ_ONLY, 0, fc.size());
			gather[2] = filedata;
			contentLength = fc.size();
			contentType = URLConnection.guessContentTypeFromName(file);
		} catch (IOException e) {
			// file could not be opened; report problem
			ByteBuffer buf = ByteBuffer.allocate(128);
			String msg = MSG_404 + e + LINE_SEP;
			buf.put(bytes(msg));
			buf.flip();
			// Use the HTTP error response
			gather[0] = ByteBuffer.wrap(bytes(HTTP_404_HDR));
			gather[2] = buf;
			contentLength = msg.length();
			contentType = "text/plain";
		}
		StringBuffer sb = new StringBuffer();
		sb.append("Content-Length: " + contentLength);
		sb.append(LINE_SEP);
		sb.append("Content-Type: ").append(contentType);
		sb.append(LINE_SEP).append(LINE_SEP);
		dynhdrs.put(bytes(sb.toString()));
		dynhdrs.flip();
		FileOutputStream fos = new FileOutputStream(OUTPUT_FILE);
		FileChannel out = fos.getChannel();
		// All the buffers have been prepared; write 'em out
		while (out.write(gather) > 0) {
			// Empty body; loop until all buffers are empty
		}
		out.close();
		System.out.println("output written to " + OUTPUT_FILE);
	}

	// Convert a string to its constituent bytes
	// from the ASCII character set
	private static byte[] bytes(String string) throws Exception {
		return (string.getBytes("US-ASCII"));
	}
}

 

 

Channel-to-Channel Transfers

Bulk transfers of file data from one place to another is so common that a couple of
optimization methods havebeen added to the FileChannelclass to make it even more
efficient:

 

The transferTo()and transferFrom()methods allow you to cross-connect one channel to
another, eliminating the need topass data through an intermediate buffer. These methods
exist only on the FileChannelclass, so one of the channels involved in a
channel-to-channel transfer must be a FileChannel. You can't do direct transfers between
socket channels, but socket channels implement WritableByteChanneland
ReadableByteChannel, so the content of a file can betransferred to a socket with
transferTo(), or data can be read from a socket directly into a file with transferFrom().

 

Channel-to-channel transfers can potentially be extremely fast, especially where the
underlying operating system provides native support. Some operating systems can
perform direct transfers without ever passing the data through user space. This can be a
huge win for high-volume data transfer.

 

 

package com.java.nio;

import java.io.FileInputStream;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.channels.WritableByteChannel;

/**
 * It takes a list of file names as arguments, opens each 
 * in turn and transfers (copies) their content to the given 
 * WritableByteChannel (in this case, stdout). 
 */
public class ChannelTransfer {
	public static void main(String[] argv) throws Exception {
		if (argv.length == 0) {
			System.err.println("Usage: filename ...");
			return;
		}
		catFiles(Channels.newChannel(System.out), argv);
	}

	// Concatenate the content of each of the named files to
	// the given channel. A very dumb version of 'cat'.
	private static void catFiles(WritableByteChannel target, String[] files)
			throws Exception {
		for (int i = 0; i < files.length; i++) {
			FileInputStream fis = new FileInputStream(files[i]);
			FileChannel channel = fis.getChannel();
			channel.transferTo(0, channel.size(), target);
			channel.close();
			fis.close();
		}
	}
}

 

 

 

Socket channels

It's no longer necessary to dedicate a thread to each socket connection (and suffer the context-switching overhead of managing large numbers of threads). Using the new NIO classes, one or a
few threads can manage hundreds or even thousands of active socket connections with little or no performance loss. 

 

Perform readiness selection ofsocket channels using a Selector object.

The preexisting socket channels in java.netare reused for most protocol operations.

 

While every socket channel (in java.nio.channels) has an associated java.netsocket
object, not all sockets have an associated channel.

 

Nonblocking Mode

Readiness selection is a mechanism by which a channel can be queried to determine if it's
ready to perform an operation of interest, such as reading or writing.

 

Setting or resetting a channel's blocking mode is easy.

Simply call configureBlocking() with trueto place it in blocking mode, or falsefor nonblocking mode.

You can determine which mode a socket channel is currently in by invoking isBlocking():

 

SocketChannel sc = SocketChannel.open(); 
sc.configureBlocking (false); // nonblocking 
... 
if ( ! sc.isBlocking()) { 
doSomething ();// do what??? 
} 

 

 

Nonblocking sockets are usually thought of for server-side use because they make it
easier to manage many sockets simultaneously.

But there can also be benefits to using one or a few sockets in nonblocking mode on the client side.

 

ServerSocketChannel

The ServerSocketChannelclass is a channel-based socket listener.

Create a new ServerSocketChannelobject with the static open()factory method, which
returns a channel associated with an unbound java.net.ServerSocketobject.

 

ServerSocketChannel ssc = ServerSocketChannel.open(); 
ServerSocket serverSocket = ssc.socket(); 
// Listen on port 1234 
serverSocket.bind (new InetSocketAddress (1234)); 

 

 

If you choose to invoke accept()on the ServerSocket, it will behave the same as any other ServerSocket: always blocking and returning a java.net.Socketobject.

 

On the other hand, the accept()method of ServerSocketChannelreturns objects of type SocketChannel and is capable of operating in nonblocking mode.

 

A ServerSocketChannelobject can be registered with a Selectorinstance to enable notification when new connections arrive.

 

 

package com.java.nio;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

/**
 * Test nonblocking accept() using ServerSocketChannel. Start this program, then
 * "telnet localhost 1234" to connect to it.
 * 
 * @author Ron Hitchens ([e-mail])
 */
public class ChannelAccept {
	public static final String GREETING = "Hello I must be going.\r\n";

	public static void main(String[] argv) throws Exception {
		int port = 1234; // default
		if (argv.length > 0) {
			port = Integer.parseInt(argv[0]);
		}
		ByteBuffer buffer = ByteBuffer.wrap(GREETING.getBytes());
		ServerSocketChannel ssc = ServerSocketChannel.open();
		ssc.socket().bind(new InetSocketAddress(port));
		ssc.configureBlocking(false);
		while (true) {
			System.out.println("Waiting for connections");
			SocketChannel sc = ssc.accept();
			if (sc == null) {
				// no connections, snooze a while
				Thread.sleep(2000);
			} else {
				System.out.println("Incoming connection from: "
						+ sc.socket().getRemoteSocketAddress());
				buffer.rewind();
				sc.write(buffer);
				sc.close();
			}
		}
	}
}

 

 

SocketChannel

A SocketChannelacts as the client, initiating a connection to a listening server.
It cannot receive until connected and then only from the address to which the connection
was made.

SocketChannel socketChannel = 
SocketChannel.open (new InetSocketAddress ("somehost", somePort)); 

//is equivalent to this: 

SocketChannel socketChannel = SocketChannel.open(); 
socketChannel.connect (new InetSocketAddress ("somehost", somePort)); 

 

While in this intermediate connection-pending state, you should invoke only
finishConnect(), isConnectPending(), or isConnected()on the channel. Once connection
establishment has been successfully completed, isConnected()returns true.

 

InetSocketAddress addr = new InetSocketAddress (host, port); 
SocketChannel sc = SocketChannel.open(); 
sc.configureBlocking (false); 
sc.connect (addr); 
while ( ! sc.finishConnect()) { 
doSomethingElse(); 
} 
doSomethingWithChannel (sc); 
sc.close();

 

If an asynchronous-connection attempt fails, the next invocation of finishConnect()
throws an appropriate checked exception to indicate the nature of the problem.

 

Keep in mind that sockets are stream-oriented, not packet-oriented.

They guarantee that the bytes sent will arrive in the same order but make no promises about maintaining groupings.

 

A sender may write 20 bytes to a socket, and the receiver gets only 3 of those bytes when invoking read(). The remaining 17 bytes may still be in transit. For this reason, it's rarely a good design choice to have multiple, noncooperating threads share the same side of a stream socket.

 

 

 

Pipes

The Pipeclass creates a pair of Channelobjects that provide a loopback mechanism.

The two channels' far ends are connected so that whatever is written down the SinkChannel
appears on the SourceChannel.

任何写入到sinkChannel的数据都可以直接从SourceChannel中读取出来

 

package com.java.nio;

import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.Pipe;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Random;

/**
 * Test Pipe objects using a worker thread.
 * 
 */
public class PipeTest {
	public static void main(String[] argv) throws Exception {
		// Wrap a channel around stdout
		WritableByteChannel out = Channels.newChannel(System.out);
		// Start worker and get read end of channel
		ReadableByteChannel workerChannel = startWorker(10);
		ByteBuffer buffer = ByteBuffer.allocate(100);
		while (workerChannel.read(buffer) >= 0) {
			buffer.flip();
			out.write(buffer);
			buffer.clear();
		}
	}

	// This method could return a SocketChannel or
	// FileChannel instance just as easily
	private static ReadableByteChannel startWorker(int reps) throws Exception {
		Pipe pipe = Pipe.open();
		Worker worker = new Worker(pipe.sink(), reps);
		worker.start();
		return (pipe.source());
	}

	// -----------------------------------------------------------------
	/**
	 * A worker thread object which writes data down a channel. Note: this
	 * object knows nothing about Pipe, uses only a generic WritableByteChannel.
	 */
	private static class Worker extends Thread {
		WritableByteChannel channel;
		private int reps;

		Worker(WritableByteChannel channel, int reps) {
			this.channel = channel;
			this.reps = reps;
		}

		// Thread execution begins here
		public void run() {
			ByteBuffer buffer = ByteBuffer.allocate(100);
			try {
				for (int i = 0; i < this.reps; i++) {
					doSomeWork(buffer);
					// channel may not take it all at once
					while (channel.write(buffer) > 0) {
						// empty
					}
				}
				this.channel.close();
			} catch (Exception e) {
				// easy way out; this is demo code
				e.printStackTrace();
			}
		}

		private String[] products = { "No good deed goes unpunished",
				"To be, or what?", "No matter where you go, there you are",
				"Just say \"Yo\"", "My karma ran over my dogma" };
		private Random rand = new Random();

		private void doSomeWork(ByteBuffer buffer) {
			int product = rand.nextInt(products.length);
			buffer.clear();
			buffer.put(products[product].getBytes());
			buffer.put("\r\n".getBytes());
			buffer.flip();
		}
	}
}

 

 

Channels utility class

A utility class, with the slightly repetitive name of java.nio.channels.Channels,
defines several static factory methods to make it easier for channels to interconnect with
streams and readers/writers.

 

Channel与传统IO中的InputStream, OutputStream, Reader, Writer之间的转换.