java - 跟踪执行线程

Question

我想弄清楚如何跟踪我的应用程序生成的所有线程。最初，我以为我已经使用 CyclicBarrier 解决了这个问题，但是我看到线程在我的 await 调用之后执行。

下面是工作伪代码:

public class ThreadTesterRunner {

    public static void main(String[] args) throws InterruptedException {

        final CyclicBarrier cb = new CyclicBarrier(1);
        ThreadRunner tr = new ThreadRunner(cb);
        Thread t = new Thread(tr, "Thread Runner");
        t.start();

        boolean process = true;
        // wait until all threads process, then print reports
        while (process){
            if(tr.getIsFinished()){
                System.out.println("Print metrics");
                process = false;
            }
            Thread.sleep(1000);
        }
    }
}


class ThreadRunner implements Runnable {
    static int timeOutTime = 2;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final CyclicBarrier barrier;
    private boolean isFinished=false;

    public ThreadRunner(CyclicBarrier cb) {
        this.barrier = cb;
    }

    public void run(){
        try {
            boolean stillLoop = true; int i = 0;
            while (stillLoop){
                int size;
                Future<Integer> future = null;
                try {
                    future = executorService.submit(new Reader()); // sleeps
                    size = future.get();
                } catch (InterruptedException | ExecutionException ex) {
                    // handle Errs
                }

                if(i == 3){
                    stillLoop = false;
                    this.barrier.await();
                    this.isFinished=true;
                }
                //System.out.println("i = "+i+"  Size is: "+size+"\r");
                i++;
            }
        } catch (InterruptedException | BrokenBarrierException e1) {
            e1.printStackTrace();
        }
    }

    public boolean getIsFinished(){
        return this.isFinished;
    }
}

class Reader implements Callable {
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    @Override
    public Object call() throws Exception {
        System.out.println("Reading...");
        Thread.sleep(2000);
        executorService.submit(new Writer());
        return 1000;
    }
}

class Writer implements Callable {
    @Override
    public Void call() throws Exception {
        Thread.sleep(4000);
        System.out.println("Wrote");    
        return null;
    }
}

谁能建议一种在所有线程运行后只打印“打印指标”的方法？

Answer 1

您似乎没有做任何事情来协调您的 Reader 和 Writer 线程，而这正是您要等待的线程。如果您将同步屏障传递给这些线程，以便它们可以在完成时注册并发出信号，那么它就可以正常工作。

这是为此重写的版本，使用 Phaser 而不是 CyclicBarrier。请注意，每个 Reader 和 Writer 在构造时注册自己，并在执行完成时通知同步屏障:

public class ThreadTesterRunner {
    public static void main(String[] args) throws InterruptedException {
        final Phaser cb = new Phaser();
        ThreadRunner tr = new ThreadRunner(cb);
        Thread t = new Thread(tr, "Thread Runner");
        t.start();

        boolean process = true;
        // wait until all threads process, then print reports
        while (process){
            if(tr.getIsFinished()){
                System.out.println("Print metrics");
                process = false;
            }
            //else {
            //  System.out.println("Waiting:  registered=" + cb.getRegisteredParties() + ", arrived=" + cb.getArrivedParties() + ", unarrived=" + cb.getUnarrivedParties());
            //}
            Thread.sleep(1000);
        }
    }
}


class ThreadRunner implements Runnable {
    static int timeOutTime = 2;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final Phaser barrier;
    private boolean isFinished=false;

    public ThreadRunner(Phaser phaser) {
        this.barrier = phaser;
    }

    public void run(){
        try {
            boolean stillLoop = true; int i = 0;
            while (stillLoop){
                int size;
                Future<Integer> future = null;
                try {
                    future = executorService.submit(new Reader(this.barrier)); // sleeps
                    size = future.get();
                } catch (InterruptedException | ExecutionException ex) {
                    // handle Errs
                }

                if(i == 3){
                    stillLoop = false;
                    this.barrier.awaitAdvance(0);
                    this.isFinished=true;
                }
                //System.out.println("i = "+i+"  Size is: "+size+"\r");
                i++;
            }
        } catch (Exception e1) {
            e1.printStackTrace();
        }
    }

    public boolean getIsFinished(){
        return this.isFinished;
    }
}

class Reader implements Callable {
    private Phaser barrier;
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    public Reader(Phaser phase) {
        phase.register();
        this.barrier = phase;
    }

    @Override
    public Object call() throws Exception {
        System.out.println("Reading...");
        Thread.sleep(2000);
        executorService.submit(new Writer(this.barrier));
        this.barrier.arrive();
        return 1000;
    }
}

class Writer implements Callable {
    private Phaser barrier;

    public Writer(Phaser phase) {
        phase.register();
        this.barrier = phase;
    }

    @Override
    public Void call() throws Exception {
        Thread.sleep(4000);
        System.out.println("Wrote");
        this.barrier.arrive();
        return null;
    }
}