Java多线程环境下如何高效安全处理数据(输入输出流、文件、网络等)(二)

    本博客属原创文章,欢迎转载！转载请务必注明出处:http://guoyunsky.iteye.com/blog/868043

       本博客已迁移本人独立博客: http://www.yun5u.com/

 

         由于不想让博客太难看,所以写到二.这里也会循序渐进,一步步走入所谓的高效和安全.

       上一篇,介绍了思路、设计。博客地址:http://guoyunsky.iteye.com/blog/867469 同时写了个MyOutputStream,用于读取输入流(如IO、网络),但恐怕输入流过大,或者多线程并发读取的时候导致总的内存占用过大.所以采用缓存+文件存放形式。缓存固定一个大小,这样可以控制内存,同时可以重复利用.超过缓存的数据写入到磁盘,避免内存溢出。但如此保存的数据又要读取出来处理,那就又需要一个读取类.将缓存和硬盘中的数据都读出。有人可能会说,怎么会这么麻烦？有这种需求吗？就比如发表博客,网站需要过滤敏感词.那么就需要多个线程读取博客,然后过滤关键字后输出到MyOuputStream.之后还需要什么处理,但就要获取这些数据.于是有了下面一个类,可以以流的形式读取缓存和磁盘中的数据.先贴上代码:

 

Java代码  



import java.io.File;  

import java.io.FileInputStream;  

import java.io.FileNotFoundException;  

import java.io.IOException;  

import java.io.InputStream;  

import java.io.RandomAccessFile;  

  

public class ReuseInputStream extends InputStream {  

    private InputStream diskStream; // 硬盘数据  

    private long position;          // 当前位置  

    private long size;              // 数据大小  

    private byte[] buffer;          // 缓存  

      

      

      

    public ReuseInputStream(long size, byte[] buffer,String backedFileName) throws FileNotFoundException {  

        super();  

        this.size = size;  

        this.buffer = buffer;  

          

        if(size>buffer.length){  

            File backedFile=new File(backedFileName);  

            //RandomAccessFile raf=new RandomAccessFile(backedFile,"r"); // 先埋个伏笔,还有改进之处  

            diskStream=new FileInputStream(backedFile);  

        }  

    }  

      

    @Override  

    public int read() throws IOException {  

        if(position==size){ // 已经满了  

            return -1;  

        }  

          

        if(position<buffer.length){    // 先从缓存中读  

           int c=buffer[(int)position]& 0xFF;   

           position++;  

           return c;  

        }  

          

        // 缓存中已经读取完毕,从文件中读取  

        int c=diskStream.read();  

        if(c>=0){   // 确保文件中也还有数据  

            position++;  

        }  

        return c;  

    }  

  

    @Override  

    public int read(byte[] b, int off, int len) throws IOException {  

        if(position==size){  // 已经满了  

            return -1;  

        }  

          

        if(position<buffer.length){ // 先从缓存中读取,但缓存读完了,不会从硬盘中读取,而是直接返回  

            int toCopy=(int)Math.max(size-position, Math.min(len, buffer.length-position)); // 计算要从缓存中读取的字节数  

            if(toCopy>0){  

                System.arraycopy(buffer, (int)position, b, off, len);  

                position+=toCopy;  

            }  

            return toCopy;  

        }  

          

        int c=diskStream.read(b, off, len); // 从磁盘中读取  

        if(c>0){  

            position+=c;  

        }  

        return c;  

          

    }  

  

    @Override  

    public void close() throws IOException {  

        super.close();  

        if(diskStream!=null){  

            diskStream.close();  

        }  

    }  

      

    // 获得数据大小  

    public long getSize() {  

        return size;  

    }  

      

    // 剩下的数据长度  

    public long getRemain(){  

        return size-position;  

    }  

    // 获得当前位置  

    public long getPosition() {  

        return position;  

    }  

      

    // 获得缓存数据  

    public byte[] getBuffer() {  

        return buffer;  

    }  

      

}  

 

      这个类得提供给MyOutputStream,让它可以返回它里面的数据,于是修改MyOutputStream。这里省略前面的代码,同时可以看测试：

 

Java代码  



package streamUnderThreads;  

  

import java.io.File;  

import java.io.FileNotFoundException;  

import java.io.FileOutputStream;  

import java.io.IOException;  

import java.io.OutputStream;  

  

public class MyOutputStream extends OutputStream {  

    private boolean isOpen;         // 是否已经打开  

    private long size;              // 数据总大小  

    private String backedFileName;  // 超出缓存,要写入到的文件名  

    private OutputStream diskStream;// 超出缓存,写入到文件的OutputStream  

    private byte[] buffer;          // 缓存  

    private long position;          // 当前位置  

    private boolean recording;      // 是否记录数据中  

      

    //省略...  

         

    // 获得数据流  

    public ReuseInputStream getReuseInputStream() throws IOException{  

        return new ReuseInputStream(size,buffer,backedFileName);  

    }  

       

    public static void main(String[] args) {  

        String dir=new File("").getAbsolutePath().concat(File.separator);  

        String fileMemory=dir.concat("fileMemory.txt");  

        String fileDisk=dir.concat("fileDisk.txt");  

        int bufferSize=5;  

        MyOutputStream mosMemory=null;  

        MyOutputStream mosDisk=null;  

        try {  

            mosMemory=new MyOutputStream(bufferSize,fileMemory);  

            mosMemory.open();  

              

            mosDisk=new MyOutputStream(bufferSize,fileDisk);  

            mosDisk.open();  

              

            for(int i=0;i<100;i++){  

                if(i<bufferSize){  

                    mosMemory.write(i);  

                }  

                mosDisk.write(i);  

            }  

              

            mosMemory.close();  

            mosDisk.close();  

              

            System.out.println("mosMemory length:"+mosMemory.getSize());  

            System.out.println("mosDisk length:"+mosDisk.getSize());  

              

            // 返回数据流  

            ReuseInputStream ris=mosDisk.getReuseInputStream();  

            StringBuilder sb=new StringBuilder();  

            int c;  

            while((c=ris.read())!=-1){  

                sb.append(c);  

            }  

            System.out.println(sb.toString());  

              

        } catch (IOException e) {  

            // TODO Auto-generated catch block  

            e.printStackTrace();  

        }finally{  

              

        }  

    }  

  

}