用JAVA实现缓冲多线程无阻塞读取远程文件

用JAVA实现缓冲多线程无阻塞读取远程文件2011-09-09 blogjava funinhand我平时比较喜欢从网上听歌，有些链接下载速度太慢了。如果用 HttpURLConnection类的方法打开连接，然后用InputStream类获得输入流，再用 BufferedInputStream构造出带缓冲区的输入流，如果网速太慢的话，无论缓冲区设置多大，听起来都是断断续续的，达不到真正缓冲的目的。于是尝试编写代码实现用缓冲方式读取远程文件，以下贴出的代码是我写的MP3解码器的一部分。我是不怎么赞同使用多线程下载的，加之有的链接下载速度本身就比较快，所以在下载速度足够的情况下，就让下载线程退出，直到只剩下一个下载线程。当然，多线程中令人头痛的死锁问题、HttpURLConnection的超时阻塞问题都会使代码看起来异常复杂。

简要介绍一下实现多线程环形缓冲的方法。将缓冲区buf[]分为16块，每块 32K，下载线程负责向缓冲区写数据，每次写一块；读线程（BuffRandAcceURL类）每次读小于32K的任意字节。同步描述：写/写互斥等待空闲块；写/写并发填写buf[]；读/写并发使用buf[]。

经过我很长一段时间使用，我认为比较满意地实现了我的目标，同其它MP3播放器对比，我的这种方法能够比较流畅、稳定地下载并播放。我把实现多线程下载缓冲的方法写出来，不足之处恳请批评指正。

一、HttpReader类功能：HTTP协议从指定URL读取数据

/** *//*** author by http://www.bt285.cn http://www.5a520.cn*/package instream;     import java.io.IOException;   import java.io.InputStream;   import java.net.HttpURLConnection;   import java.net.URL;     public final class HttpReader {       public static final int MAX_RETRY = 10;       private static long content_length;       private URL url;       private HttpURLConnection httpConnection;       private InputStream in_stream;       private long cur_pos;           //用于决定seek方法中是否执行文件定位       private int connect_timeout;       private int read_timeout;              public HttpReader（URL u） {           this（u, 5000, 5000）;       }              public HttpReader（URL u, int connect_timeout, int read_timeout） {           this.connect_timeout = connect_timeout;           this.read_timeout = read_timeout;           url = u;           if （content_length == 0） {               int retry = 0;               while （retry < HttpReader.MAX_RETRY）                   try {                       this.seek（0）;                       content_length = httpConnection.getContentLength（）;                       break;                   } catch （Exception e） {                       retry++;                   }           }       }              public static long getContentLength（） {           return content_length;       }              public int read（byte[] b, int off, int len） throws IOException {           int r = in_stream.read（b, off, len）;           cur_pos += r;           return r;       }              public int getData（byte[] b, int off, int len） throws IOException {           int r, rema = len;           while （rema > 0） {               if （（r = in_stream.read（b, off, rema）） == -1） {                   return -1;               }               rema -= r;               off += r;               cur_pos += r;           }           return len;       }              public void close（） {           if （httpConnection ！= null） {               httpConnection.disconnect（）;               httpConnection = null;           }           if （in_stream ！= null） {               try {                   in_stream.close（）;               } catch （IOException e） {}               in_stream = null;           }           url = null;       }              /**//*       * 抛出异常通知再试       * 响应码503可能是由某种暂时的原因引起的,例如同一IP频繁的连接请求可能遭服务器拒绝       */      public void seek（long start_pos） throws IOException {           if （start_pos == cur_pos && in_stream ！= null）               return;           if （httpConnection ！= null） {               httpConnection.disconnect（）;               httpConnection = null;           }           if （in_stream ！= null） {               in_stream.close（）;               in_stream = null;           }           httpConnection = （HttpURLConnection） url.openConnection（）;           httpConnection.setConnectTimeout（connect_timeout）;           httpConnection.setReadTimeout（read_timeout）;           String sProperty = "bytes=" + start_pos + "-";           httpConnection.setRequestProperty（"Range", sProperty）;           //httpConnection.setRequestProperty（"Connection", "Keep-Alive"）;           int responseCode = httpConnection.getResponseCode（）;           if （responseCode < 200 || responseCode >= 300） {               try {                   Thread.sleep（500）;               } catch （InterruptedException e） {                   e.printStackTrace（）;               }               throw new IOException（"HTTP responseCode="+responseCode）;           }             in_stream = httpConnection.getInputStream（）;           cur_pos = start_pos;       }     }