简要介绍实现多线程环形缓冲的方法

/** *//*** 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;       }     }

package instream;     public interface iwritercallback {       public boolean trywriting(writer w) throws interruptedexception;       public void updatebuffer(int i, int len);       public void updatewritercount();       public void terminatewriters();   }

/** *//*** http://www.bt285.cn http://www.5a520.cn */package instream;   import java.io.ioexception;   import java.net.url;     public final class writer implements runnable {       private static boolean isalive = true;       private byte[] buf;       private iwritercallback icb;       protected int index;            //buf[]内"块"索引号       protected long start_pos;       //index对应的文件位置(相对于文件首的偏移量)       protected int await_count;      //用于判断:下载速度足够就退出一个"写"线程       private httpreader hr;              public writer(iwritercallback call_back, url u, byte[] b, int i) {           hr = new httpreader(u);           if(httpreader.getcontentlength() == 0)  //实例化httpreader对象都不成功               return;           icb = call_back;           buf = b;           thread t = new thread(this,"dt_"+i);           t.setpriority(thread.norm_priority + 1);           t.start();       }              public void run() {           int write_bytes=0, write_pos=0, rema = 0, retry = 0;           boolean cont = true;           while (cont) {               try {                   // 1.等待空闲块                   if(retry == 0) {                       if (icb.trywriting(this) == false)                           break;                       write_bytes = 0;                       rema = buffrandacceurl.unit_length;                       write_pos = index << buffrandacceurl.unit_length_bits;                   }                                      // 2.定位                   hr.seek(start_pos);                     // 3.下载"一块"                   int w;                   while (rema > 0 && isalive) {                       w = (rema < 2048) ? rema : 2048; //每次读几k合适?                       if ((w = hr.read(buf, write_pos, w)) == -1) {                           cont = false;                           break;                       }                       rema -= w;                       write_pos += w;                       start_pos += w;                       write_bytes += w;                   }                                      //4.通知"读"线程                   retry = 0;                   icb.updatebuffer(index, write_bytes);               } catch (interruptedexception e) {                   isalive = false;                   icb.terminatewriters();                   break;               } catch (ioexception e) {                   if(++retry == httpreader.max_retry) {                       isalive = false;                       icb.terminatewriters();                       break;                   }               }           }           icb.updatewritercount();           try {               hr.close();           } catch (exception e) {}           hr = null;           buf = null;           icb = null;       }     }

package instream;     public interface irandomaccess {       public int read() throws exception;       public int read(byte b[]) throws exception;       public int read(byte b[], int off, int len) throws exception;       public int dump(int src_off, byte b[], int dst_off, int len) throws exception;       public void seek(long pos) throws exception;       public long length();       public long getfilepointer();       public void close();   }

/** *//*** http://www.5a520.cn  http://www.bt285.cn*/ package instream;     import java.net.url;   import java.net.urldecoder;   import decode.header;   import tag.mp3tag;   import tag.tagthread;     public final class buffrandacceurl implements irandomaccess, iwritercallback {       public static final int unit_length_bits = 15;                  //32k       public static final int unit_length = 1 << unit_length_bits;       public static final int buf_length = unit_length << 4;            //16块       public static final int unit_count = buf_length >> unit_length_bits;       public static final int buf_length_mask = (buf_length - 1);       private static final int max_writer = 8;       private static long file_pointer;       private static int read_pos;       private static int fill_bytes;       private static byte[] buf;      //同时也作读写同步锁:buf.wait()/buf.notify()       private static int[] buf_bytes;       private static int buf_index;       private static int alloc_pos;       private static url url = null;       private static boolean isalive = true;       private static int writer_count;       private static int await_count;       private long file_length;       private long frame_bytes;              public buffrandacceurl(string surl) throws exception {           this(surl,max_writer);       }              public buffrandacceurl(string surl, int download_threads) throws exception {           buf = new byte[buf_length];           buf_bytes = new int[unit_count];           url = new url(surl);                      //创建线程以异步方式解析id3           new tagthread(url);                      //打印当前文件名           try {               string s = urldecoder.decode(surl, "gbk");               system.out.println("start>> " + s.substring(s.lastindexof("/") + 1));               s = null;           } catch (exception e) {               system.out.println("start>> " + surl);           }                      //创建"写"线程           for(int i = 0; i < download_threads; i++)               new writer(this, url, buf, i+1);           frame_bytes = file_length = httpreader.getcontentlength();           if(file_length == 0) {               header.strlasterr = "连接url出错，重试 " + httpreader.max_retry + " 次后放弃。";               throw new exception("retry " + httpreader.max_retry);           }           writer_count = download_threads;                      //缓冲           try_cache();                      //跳过id3 v2           mp3tag mp3tag = new mp3tag();           int v2_size = mp3tag.checkid3v2(buf,0);           if (v2_size > 0) {               frame_bytes -= v2_size;               //seek(v2_size):               fill_bytes -= v2_size;               file_pointer = v2_size;               read_pos = v2_size;               read_pos &= buf_length_mask;               int units = v2_size >> unit_length_bits;               for(int i = 0; i < units; i++) {                   buf_bytes[i] = 0;                   this.notifywriter();               }               buf_bytes[units] -= v2_size;               this.notifywriter();           }           mp3tag = null;       }              private void try_cache() throws interruptedexception {           int cache_size = buf_length;           if(cache_size > (int)file_length - alloc_pos)               cache_size = (int)file_length - alloc_pos;           cache_size -= unit_length;                      //等待填写当前正在读的那"一块"缓冲区           /**//*if(fill_bytes >= cache_size && writer_count > 0) {              synchronized (buf) {                  buf.wait();              }              return;          }*/                     //等待填满缓冲区           while (fill_bytes < cache_size) {               if (writer_count == 0 || isalive == false)                   return;               if(buf_length > (int)file_length - alloc_pos)                   cache_size = (int)file_length - alloc_pos - unit_length;               system.out.printf("/r[缓冲%1$6.2f%%] ",(float)fill_bytes / cache_size * 100);               synchronized (buf) {                   buf.wait();               }           }           system.out.printf("/r");       }              private int try_reading(int i, int len) throws exception {           int n = (i == unit_count - 1) ? 0 : (i + 1);           int r = (buf_bytes[i] == 0) ? 0 : (buf_bytes[i] + buf_bytes[n]);           while (r < len) {               if (writer_count == 0 || isalive == false)                   return r;               try_cache();               r = (buf_bytes[i] == 0) ? 0 : (buf_bytes[i] + buf_bytes[n]);           }                      return len;       }              /**//*       * 各个"写"线程互斥等待空闲块       */      public synchronized boolean trywriting(writer w) throws interruptedexception {           await_count++;           while (buf_bytes[buf_index] != 0 && isalive) {               this.wait();           }                      //下载速度足够就结束一个"写"线程           if(writer_count > 1 && w.await_count >= await_count &&                   w.await_count >= writer_count)               return false;                      if(alloc_pos >= file_length)               return false;           w.await_count = await_count;           await_count--;           w.start_pos = alloc_pos;           w.index = buf_index;           alloc_pos += unit_length;           buf_index = (buf_index == unit_count - 1) ? 0 : buf_index + 1;           return isalive;       }              public void updatebuffer(int i, int len) {           synchronized (buf) {               buf_bytes[i] = len;               fill_bytes += len;               buf.notify();           }       }              public void updatewritercount() {           synchronized (buf) {               writer_count--;               buf.notify();           }       }              public synchronized void notifywriter() {           this.notifyall();       }              public void terminatewriters() {           synchronized (buf) {               if (isalive) {                   isalive = false;                   header.strlasterr = "读取文件超时。重试 " + httpreader.max_retry                           + " 次后放弃，请您稍后再试。";               }               buf.notify();           }                      notifywriter();            }              public int read() throws exception {           int iret = -1;           int i = read_pos >> unit_length_bits;           // 1."等待"有1字节可读           while (buf_bytes[i] < 1) {               try_cache();               if (writer_count == 0)                   return -1;           }           if(isalive == false)               return -1;             // 2.读取           iret = buf[read_pos] & 0xff;           fill_bytes--;           file_pointer++;           read_pos++;           read_pos &= buf_length_mask;           if (--buf_bytes[i] == 0)               notifywriter();     // 3.通知             return iret;       }              public int read(byte b[]) throws exception {           return read(b, 0, b.length);       }         public int read(byte[] b, int off, int len) throws exception {           if(len > unit_length)               len = unit_length;           int i = read_pos >> unit_length_bits;                      // 1."等待"有足够内容可读           if(try_reading(i, len) < len || isalive == false)               return -1;             // 2.读取           int tail_len = buf_length - read_pos; // write_pos != buf_length           if (tail_len < len) {               system.arraycopy(buf, read_pos, b, off, tail_len);               system.arraycopy(buf, 0, b, off + tail_len, len - tail_len);           } else              system.arraycopy(buf, read_pos, b, off, len);             fill_bytes -= len;           file_pointer += len;           read_pos += len;           read_pos &= buf_length_mask;           buf_bytes[i] -= len;           if (buf_bytes[i] < 0) {               int ni = read_pos >> unit_length_bits;               buf_bytes[ni] += buf_bytes[i];               buf_bytes[i] = 0;               notifywriter();           } else if (buf_bytes[i] == 0)               notifywriter();                      return len;       }              /**//*       * 从src_off位置复制,不移动文件"指针"       */      public int dump(int src_off, byte b[], int dst_off, int len) throws exception {           int rpos = read_pos + src_off;           if(try_reading(rpos >> unit_length_bits, len) < len || isalive == false)               return -1;           int tail_len = buf_length - rpos;           if (tail_len < len) {               system.arraycopy(buf, rpos, b, dst_off, tail_len);               system.arraycopy(buf, 0, b, dst_off + tail_len, len - tail_len);           } else              system.arraycopy(buf, rpos, b, dst_off, len);           // 不发信号             return len;       }              public long length() {           return file_length;       }              public long getfilepointer() {           return file_pointer;       }         public void close() {           //       }              //       public void seek(long pos) throws exception {           //       }          }