NIO异步的问题

Question

我最近学了学NIO，但只是入门知识，现在不知道需要上传的源文件是怎么读入到注册到选择器上的套接字通道里的，还有就是异步中可以使用map吗？不知道问题说得够不够具体希望高手们能帮忙解答一下

Answer 1

因为我现在不怎么了解nio可能问题很弱也不太清楚，我看网上说是把通道注册到选择器上，然后侦听这个通道有事件发生时返回一个键，通过这个键可以调用这个通道，然后就可以读写了，可是我不明白怎么把用户要上传的文件放到这个通道里啊，不知道我问的对是不对，希望能帮帮我这个新手啊

Answer 2

encode ---> Channel ---> decode

Answer 3

大体这么个意思吧。
上传前把 file.getName();之类的先发过去。
package net;import java.io.File;
import java.io.FileInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;public class Client {
private SocketChannel socketChannel;
private Selector selector;
private ByteBuffer buffer = ByteBuffer.allocate(1024); public Client() throws IOException {
socketChannel = SocketChannel.open();
InetSocketAddress isa = new InetSocketAddress(InetAddress
.getLocalHost(), 12345);
socketChannel.connect(isa);
socketChannel.configureBlocking(false);
selector = Selector.open();
} public void upload(String fileName) throws IOException {
File file = new File(fileName);
if (file.isDirectory()) {
return;
}

FileInputStream fis = new FileInputStream(file);
fis.getChannel().read(buffer);

socketChannel.register(selector, SelectionKey.OP_READ
| SelectionKey.OP_WRITE); while (selector.select() > 0) {
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = null;
key = (SelectionKey) it.next();
it.remove(); if (key.isReadable()) {
// ....
}
if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
synchronized (buffer) {
buffer.flip();
socketChannel.write(buffer);
buffer.compact();
}
}
}
}
} public static void main(String args[]) {
try {
new Client().upload("/home/zhq/test.txt");
} catch (IOException e) {
e.printStackTrace();
}
}}
package net;import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;public class Server {
private Selector selector = null;
private ServerSocketChannel serverSocketChannel = null;
private int port = 12345; public Server() throws IOException {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(port)); } public void service() throws IOException {
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0) {
Set readyKeys = selector.selectedKeys();
Iterator it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = null;
try {
key = (SelectionKey) it.next();
it.remove(); if (key.isAcceptable()) {
ServerSocketChannel ssc = (ServerSocketChannel) key
.channel();
SocketChannel socketChannel = (SocketChannel) ssc
.accept();
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(1024);
socketChannel.register(selector, SelectionKey.OP_READ
| SelectionKey.OP_WRITE, buffer);
}
if (key.isReadable()) {
receive(key);
}
if (key.isWritable()) {
send(key);
}
} catch (IOException e) {
e.printStackTrace();
try {
if (key != null) {
key.cancel();
key.channel().close();
}
} catch (Exception ex) {
e.printStackTrace();
}
}
}
}
} public void send(SelectionKey key) throws IOException {
//....
} public void receive(SelectionKey key) throws IOException {
ByteBuffer buffer = (ByteBuffer) key.attachment(); SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip(); buffer.limit(buffer.capacity());
buffer.put(readBuff);

FileOutputStream fos = new FileOutputStream("/tmp/test.txt");
buffer.flip();
fos.getChannel().write(buffer);
System.out.println("Deal!");

}
public static void main(String args[]) throws Exception {
new Server().service();
}}

Answer 4

上面的高手我试了你的代码发现有点问题，CPU占用100%要手动停止，你传一张图片试一下。
所以根据你的代码我稍微改动了一下，现在的问题是在服务器端的buffer不知道怎么接不到值。
还请高手帮忙修改一下，这两天看的我头都大了。
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import org.apache.log4j.Logger;public class Server { private Logger logger = Logger.getLogger(Server.class); private int port; private Selector selector; public Server(int port) throws IOException {
this.port = port; ServerSocketChannel sschannel = ServerSocketChannel.open(); sschannel.configureBlocking(false); selector = Selector.open(); sschannel.socket().bind(new InetSocketAddress(port)); sschannel.register(selector, SelectionKey.OP_ACCEPT);
} public void service() {
logger.info("Server started ...");
logger.info("Server listening on port: " + port);
try { while (true) { if (selector.select() > 0) {
Iterator<SelectionKey> it = selector.selectedKeys()
.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key
.channel(); SocketChannel sc = server.accept(); sc.configureBlocking(false); sc.register(selector, SelectionKey.OP_WRITE);
}
if (key.isWritable()) { receive(key); } it.remove();
}
}
}
} catch (Exception e) {
logger.error(e.getMessage());
}
} public void receive(SelectionKey key) throws IOException {
ByteBuffer buffer = (ByteBuffer) key.attachment();
RandomAccessFile fos = new RandomAccessFile(
"d:\\aa\\20081207_ed0ab933dd54d0fad901vh0CBAS2Zebs.jpg", "rw");
buffer.clear();
System.out.println(buffer);
fos.getChannel().write(buffer);
System.out.println("OK!"); } public static void main(String args[]) throws IOException {
Server server = new Server(8088);
server.service(); }
}
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;import org.apache.log4j.Logger;public class Client { private Logger logger = Logger.getLogger(Client.class); private SocketChannel schannel; private FileChannel fchannel; private RandomAccessFile raf; private Selector selector; public Client(String host, int port) throws IOException {
InetSocketAddress addr = new InetSocketAddress(host, port); schannel = SocketChannel.open(); schannel.connect(addr); schannel.configureBlocking(false); selector = Selector.open(); logger.info("connection has been established!...");
} public void upload(String file) throws IOException { try { raf = new RandomAccessFile(file, "r"); fchannel = raf.getChannel(); send(fchannel); } catch (IOException e) {
logger.error(e.getMessage());
} finally {
try {
if (fchannel != null) {
fchannel.close();
}
if (raf != null) {
raf.close();
}
} catch (Exception e) {
logger.error(e.getMessage());
}
}
} public void send(FileChannel channel) throws IOException {
MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_WRITE,
0, channel.size()); schannel.register(selector, SelectionKey.OP_WRITE); while (selector.select() > 0) {
Set<SelectionKey> readyKeys = selector.selectedKeys();
Iterator<SelectionKey> it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = null;
key = (SelectionKey) it.next();
it.remove(); if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
synchronized (buffer) {
buffer.flip();
socketChannel.write(buffer);
buffer.compact();
}
}
}
} } public void close() {
try {
schannel.close();
} catch (IOException e) {
logger.error(e.getMessage());
}
} public static void main(String[] args) { long start = System.currentTimeMillis();
try {
Client client = new Client("localhost", 8088); client.upload("d:\\20081207_ed0ab933dd54d0fad901vh0CBAS2Zebs.jpg");
client.close();
System.out.println("the process of data elapsed "
+ (System.currentTimeMillis() - start) * 1.0 / 1000
+ " second.");
} catch (IOException e) {
e.printStackTrace();
} }
}

Answer 5

恩，谢谢你对我的帮助，我又从新写了一个，但还是穿不过去图片，希望你有时间的时候回来看一眼帮忙改一下吧。import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import org.apache.log4j.Logger;public class Server { private Logger logger = Logger.getLogger(Server.class); private int port; private Selector selector; public Server(int port) throws IOException {
this.port = port; ServerSocketChannel sschannel = ServerSocketChannel.open(); sschannel.configureBlocking(false); selector = Selector.open(); sschannel.socket().bind(new InetSocketAddress(port)); sschannel.register(selector, SelectionKey.OP_ACCEPT);
} public void service() {
logger.info("Server started ...");
logger.info("Server listening on port: " + port); try {
for (;;) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys()
.iterator();
while (iter.hasNext()) {
SelectionKey key = (SelectionKey) iter.next();
iter.remove();
receive(key);
}
}
} catch (IOException e) {
e.printStackTrace();
} } public void receive(SelectionKey key) throws IOException { if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel = server.accept(); channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_READ);
System.out.println(channel + "222222222");
} else if (key.isReadable()) {
SocketChannel channel = (SocketChannel) key.channel(); ByteBuffer buffer = ByteBuffer.allocateDirect(1024 * 708);

System.out.println(buffer + "刚创建");

channel.read(buffer);

System.out.println(buffer + "读到buffer后"); buffer.clear(); RandomAccessFile fraf = new RandomAccessFile("d:\\image\\aa.jpg",
"rw"); FileChannel ffc = fraf.getChannel();
System.out.println(buffer + "3333333");
ffc.write(buffer);
System.out.println(buffer + "4444444");
System.out.println(ffc + "55555");
System.out.println("OK!"); if (ffc != null) {
ffc.close();
}
if (channel != null) {
channel.close();
} } } public static void main(String args[]) throws IOException {
Server server = new Server(8088);
server.service(); }
}import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;import org.apache.log4j.Logger;public class Client { private Logger logger = Logger.getLogger(Client.class); private SocketChannel client; private FileChannel fchannel; private RandomAccessFile raf; private File file; private int length; private Selector selector; private MappedByteBuffer buffer; public Client() {
}; public Client(int port) throws IOException {
InetSocketAddress addr = new InetSocketAddress(InetAddress
.getLocalHost(), port); client = SocketChannel.open();
client.configureBlocking(false);
selector = Selector.open();
client.connect(addr);
client.register(selector, SelectionKey.OP_CONNECT); logger.info("connection has been established!...");
} public void upload(String fileName) throws IOException { try { for (;;) {
selector.select();
Iterator<SelectionKey> iter = selector.selectedKeys()
.iterator(); while (iter.hasNext()) {
SelectionKey key = (SelectionKey) iter.next();
iter.remove();
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key.channel();
if (channel.isConnectionPending())
channel.finishConnect(); file = new File(fileName);
raf = new RandomAccessFile(file, "r");
fchannel = raf.getChannel();
length = (int) fchannel.size();
buffer = fchannel.map(FileChannel.MapMode.READ_ONLY, 0,
length);
buffer.clear(); channel.write(buffer); System.out.println(channel + "1111111"); if (fchannel != null) {
fchannel.close();
} }
}
} } catch (IOException e) {
e.printStackTrace();
} } public void close() {
try {
client.close();
} catch (IOException e) {
logger.error(e.getMessage());
}
} public static void main(String[] args) { long start = System.currentTimeMillis();
try {
Client client = new Client(8088); client.upload("d:\\aa.jpg");
client.close();
System.out.println("the process of data elapsed "
+ (System.currentTimeMillis() - start) * 1.0 / 1000
+ " second.");
} catch (IOException e) {
e.printStackTrace();
} } public File getFile() {
return file;
} public void setFile(File file) {
this.file = file;
} public int getLength() {
return length;
} public void setLength(int length) {
this.length = length;
}
}