Можно каналы использовать и в многопоточной среде, только разносить по отдельным потокам надо не приём и передачу, а обработку каждого нового подключения:
public class ExecutorNioServer {
private static final int PORT = 1234;
private static final String FILE_NAME = "received_%d.jpg";
public static void main(String args[]) throws IOException {
ExecutorService executor = null;
try {
executor = Executors.newFixedThreadPool(10);
try (ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {
serverSocketChannel.bind(new InetSocketAddress(PORT));
while (true) {
SocketChannel socketChannel = serverSocketChannel.accept();
executor.execute(() -> {
String fileName = String.format(FILE_NAME, System.currentTimeMillis());
try (FileChannel fileChannel = FileChannel.open(Paths.get(fileName),
StandardOpenOption.CREATE,
StandardOpenOption.WRITE);
socketChannel) {
fileChannel.transferFrom(socketChannel, 0, Long.MAX_VALUE);
}
catch (IOException exc) {
exc.printStackTrace();
}
});
}
}
}
finally {
if (executor != null)
executor.shutdown();
}
}
}
Но более интересный и естественный для каналов вариант - это не потоки, а асинхронность:
public class SelectorNioServer {
private static final int PORT = 1234;
private static final String FILE_NAME = "received_%d.jpg";
private static class Storage {
private final AsynchronousFileChannel fileChannel;
private Future<Integer> writeOperaion;
private long position;
public Storage() throws IOException {
String fileName = String.format(FILE_NAME, System.currentTimeMillis());
fileChannel = AsynchronousFileChannel.open(Paths.get(fileName),
StandardOpenOption.CREATE,
StandardOpenOption.WRITE);
}
public void write(ByteBuffer buffer) throws InterruptedException, ExecutionException {
try {
if (writeOperaion != null)
position += writeOperaion.get(5, TimeUnit.SECONDS);
writeOperaion = fileChannel.write(buffer, position);
}
catch (TimeoutException exc) {
close();
System.err.println("Write timeout");
}
}
public void close() {
try {
fileChannel.close();
}
catch (IOException exc) {
exc.printStackTrace();
}
}
}
public static void main(String args[]) throws Exception {
Selector selector = Selector.open();
try (ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {
serverSocketChannel.bind(new InetSocketAddress(PORT));
serverSocketChannel.configureBlocking(false);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
if (key.isAcceptable()) {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ, new Storage());
}
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
Storage storage = (Storage) key.attachment();
ByteBuffer buffer = ByteBuffer.allocate(256);
int read = socketChannel.read(buffer);
if (read != -1) {
buffer.flip();
storage.write(buffer);
}
else {
storage.close();
socketChannel.close();
}
}
iterator.remove();
}
}
}
}
}
Я здесь использовал и мультиплексирование сетевых каналов, и асинхронные файловые каналы, чтобы показать обе возможности реализации асинхронности в NIO.2