Java 多线程使用

Posted 2020-09-19 真菜啊

工作中遇到的问题，记录下解决的思路

 

问题：

　　 对磁盘进行碎片化测试(比如说，磁盘空间是16G)，从64K开始写文件，写满后删除一半，然后写32K 的数据，写满后删除一半。。。直到4K写满删除一般算是结束

第一阶段：

　　使用单独的一个线程进行操作，先写数据，然后删除数据，用循环控制跳出

代码

public class Task extends Thread{
    public void run() {
        int size = 64;
        while(size >=4){
            write(size);
            delete();
            size /= 2;
        }
    }
    
    private void write(int size){
        //省略具体的写数据方法，判断是否写满
    }
    
    private void delete(){
        //省略删除的方法
    }
}

View Code

上述的代码已经实现了功能，但是如果空间很大，怎么半？一个线程写的也太慢了，如何提高效率，写入的速度？

办法就是使用多线程，多个线程同时写，肯定能提升不小的效率

第二阶段：

　　使用多个线程进行操作，代码和第一阶段的代码没有变化，共同写同一个磁盘就行

然后出现了新的问题，多线程操作，到最后肯定会出现剩余的磁盘空间就够一个线程使用(假如是线程A )，其他线程已经写完了该阶段的内容(假如是线程B,C,D)，开始执行删除操作了，此时，这4个线程有的在写入数据，有的在删除数据，会导致线程A一直读到有剩余空间可以写入(因为其他线程在删除文件，腾出新的空间)，这样到最后4K的时候，就出现了A 线程还在写，其他的线程都已经停止好久了。又浪费了好多时间。

　　如何解决呢？我想到了使用线程的同步，A线程写完了就等其他线程，等到所有的线程都写完了，大家一起开始删除，等到大家都删除完成了，再一起开始下一个阶段的写入

　　那该如何等待呢？

　　我用了Object类的wait()和notifyAll()方法。

第三阶段：

　　使用同步解决其他线程结束，就剩余一个线程写的问题，进一步的提高效率

代码

import java.util.HashMap;
import java.util.Map;

/**
 * 状态信息类
 */
public class Status {
    private int finishCount = 0;
    private int threadCount = 0;
    // 存储每个线程的状态
    private Map<Integer, Boolean> maps;

    public Status(int threadCount) {
        this.maps = new HashMap<Integer, Boolean>();
        this.threadCount = threadCount;
    }

    //更新当前线程是否在等待状态，status = true表示已经在等待了
    public synchronized void setStatus(int threadIndex, boolean statu) {
        maps.put(threadIndex, statu);
    }

    public boolean getStatus() {
        boolean result = true;
        for (Map.Entry<Integer, Boolean> entry : maps.entrySet()) {
            result = result && entry.getValue();
        }
        return result;
    }

    // 更新已经完成的线程个数(全部各个阶段执行完调用)
    public synchronized void updateFinishCount() {
        this.finishCount += 1;
    }

    public int getFinishCount() {
        return finishCount;
    }
}

View Code

/**
 * 观察线程
 */
public class Watcher extends Thread{
    
    private int threadCount = 0;
    private Status status;
    
    public Watcher(Status status,int threadCount ) {
        this.status = status;
        this.threadCount = threadCount;
    }
    
    public void run() {
         while(true){
             //检查是否所有status对象上的线程是否都在等待
             if(status.getStatus()){
                 status.notifyAll();
             }
             //检查是否所有线程全部执行完成
             if(status.getFinishCount() == threadCount){
                 break;
             }
         }
    }
}

View Code

/**
 * 具体执行任务的线程
 */
public class Task extends Thread{
    private Status status;
    private int index;
    public Task(Status status,int index) {
        this.status = status;
        this.index =index;
    }
    public void run() {
        int size = 64;
        while(size >=4){
            write(size);
            synchronized (status) {
                status.setStatus(index, true);//设置为等待状态
                try {
                    status.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            status.setStatus(index, false);//取消等待状态
            delete();
            synchronized (status) {
                status.setStatus(index, true);//设置为等待状态
                try {
                    status.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            status.setStatus(index, false);//取消等待状态
            size /= 2;
        }
    }
    
    private void write(int size){
        //省略具体的写数据方法，判断是否写满
    }
    
    private void delete(){
        //省略删除的方法
    }
}

View Code

 

现在看起来是完美了，但是实际的运行过程中，会发现，真的没有控制住线程的同步，还是出现了之前的第二阶段的问题，有一个线程比其他线程慢，而且出现了一个线程没有按照依次递减的顺序执行的古怪情况，我想应该是没有真正的同步造成的。之后又去查找资料，发现Java提供了一个类，就像是为这种情况量身定做的。它就是 CyclicBarrier  ，它自己维护了一个计数器，每当调用一次await()方法，就会阻塞当前的线程，并且计数器减一,计数器的值来源于构造方法，计数器为0的时候，就解除阻塞，更好的是，当计数器为0时，再调用await()方法的时候，会将计数器变成初始值减一，重新开始一个循环。

第四阶段：

　　

**
 * 具体执行任务的线程
 */
public class Task extends Thread {
    private CyclicBarrier cyclicBarrier;

    public Task(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    public void run() {
        int size = 64;
        while (size >= 4) {
            write(size);
            try {
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                //捕获到该异常的话，表示这个线程不用等待了，需要处理一下，唤醒其他阻塞的线程
                e.printStackTrace();
            }
            delete();
            try {
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            size /= 2;
        }
    }

    private void write(int size) {
        // 省略具体的写数据方法，判断是否写满
    }

    private void delete() {
        // 省略删除的方法
    }
}

View Code

 

测试的代码

public class Test {
    private static final int THREAD_COUNT = 4;
    public static void main(String[] args) {
         CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT);
         for(int i=0;i<THREAD_COUNT;i++){
             new Task(barrier).start();
         }
    }
}

View Code