java异步socket实现文件下载（转）

嵌入式Linux

（1）服务器端NIOServer.java
import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
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.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;
/**
*
* 测试文件下载的NIOServer
*
* @author tenyears.cn
*/
public class NIOServer
{
static int BLOCK = 4096;
// 处理与客户端的交互
public class HandleClient
{
  protected FileChannel channel;
  protected ByteBuffer buffer;
  
  public HandleClient() throws IOException
  {
   this.channel = new FileInputStream(filename).getChannel();
   this.buffer = ByteBuffer.allocate(BLOCK);
  }
  public ByteBuffer readBlock()
  {
   try
   {
    buffer.clear();
    int count = channel.read(buffer);
    buffer.flip();
    if (count  iter = selector.selectedKeys().iterator();
    while (iter.hasNext())
    {
     SelectionKey key = iter.next();
     iter.remove();
     handleKey(key);
    }
   }
  }
  catch (IOException e)
  {
   e.printStackTrace();
  }
}

// 处理事件
protected void handleKey(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);
  }
  else if (key.isReadable())
  { // 读信息
   SocketChannel channel = (SocketChannel) key.channel();
   int count = channel.read(clientBuffer);
   if (count > 0)
   {
    clientBuffer.flip();
    CharBuffer charBuffer = decoder.decode(clientBuffer);
    System.out.println("Client >>" + charBuffer.toString());
    SelectionKey wKey = channel.register(selector,SelectionKey.OP_WRITE);
    wKey.attach(new HandleClient());
   }
   else
    channel.close();
   clientBuffer.clear();
  }
  else if (key.isWritable())
  { // 写事件
   SocketChannel channel = (SocketChannel) key.channel();
   HandleClient handle = (HandleClient) key.attachment();
   ByteBuffer block = handle.readBlock();
   
   if (block != null)
    channel.write(block);
   else
   {
    handle.close();
    channel.close();
   }
  }
}

public static void main(String[] args)
{
  int port = 12345;
  try
  {
   NIOServer server = new NIOServer(port);
   System.out.println("Listenning on " + port);
   while (true)
   {
    server.listen();
   }
  }
  catch (IOException e)
  {
   e.printStackTrace();
  }
}
}

(2)客户端程序：NIOClient.java
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
            
/**
* 文件下载客户端
* @author tenyears.cn
*/
public class NIOClient
{
static int SIZE = 100;
static InetSocketAddress ip = new InetSocketAddress("localhost",12345);
static CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();
//下载线程
static class Download implements Runnable
{
  protected int index;
  public Download(int index)
  {
   this.index = index;
  }

  public void run()
  {
   try
   {
    long start = System.currentTimeMillis();
    SocketChannel client = SocketChannel.open();
    client.configureBlocking(false);
    Selector selector = Selector.open();
    client.register(selector, SelectionKey.OP_CONNECT);
    client.connect(ip);
    ByteBuffer buffer = ByteBuffer.allocate(8 * 1024);
    int total = 0;
    FOR: for (;;)
    {
     selector.select();
     Iterator iter = selector.selectedKeys().iterator();
     while (iter.hasNext())
     {
      SelectionKey key = iter.next();
      iter.remove();
      if (key.isConnectable())
      {
       SocketChannel channel = (SocketChannel) key.channel();
       if (channel.isConnectionPending())
       channel.finishConnect();
       channel.write(encoder.encode(CharBuffer.wrap("Hello from " + index)));
       channel.register(selector, SelectionKey.OP_READ);
      }
      else if (key.isReadable())
      {
       SocketChannel channel = (SocketChannel) key.channel();
       int count = channel.read(buffer);
       if (count > 0)
       {
        total += count;
        buffer.clear();
       }
       else
       {
        client.close();
        break FOR;
       }
      }
     }
    }
    double last = (System.currentTimeMillis() - start) * 1.0 / 1000;
    System.out.println("Thread " + index + " downloaded " + total
    + "bytes in " + last + "s.");
   }
   catch (IOException e)
   {
    e.printStackTrace();
   }
  }
}

public static void main(String[] args) throws IOException
{
  ExecutorService exec = Executors.newFixedThreadPool(SIZE);
  for (int index = 0; index


本文来自ChinaUnix博客，如果查看原文请点：http://blog.chinaunix.net/u3/93928/showart_1995686.html