Java线程同步 2

CyclicBarrier用于设置一个Barrier，达到Barrier的线程将等待，当所有线程都达到Barrier时，所有线程继续执行。
<?xml:namespace prefix = o ns = "urn:schemas-microsoft-com:office:office" /> 

1.       创建CyclicBarrier时通过构造函数指定线程的数量n。
2.       每个线程调用await方法等待其他线程。
3.       当有n个await方法被调用时，所有处于await的线程都将继续执行。

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest

{

    private static CyclicBarrier barrier = null;

    

    public static void main(String[] args)

    {

        // 初始化

        barrier = new CyclicBarrier(10);

        

        // 启动10个线程

        System.out.println("main: starting 10 threads.");

        for (int i = 0; i < 10; i++)

        {

            Thread th = new Thread(new TestRunnable(), "thread " + i);

            th.start();

        }

    }

    

    private static class TestRunnable implements Runnable

    {

        @Override

        public void run()

        {

            String thName = Thread.currentThread().getName();

            try

            {

                // 随机等待5-14秒

                int n = (int)(Math.random() * 10 + 5);

                System.out.printf("%s: sleep for %d seconds. /n", thName, n);

                Thread.sleep(n * 1000);

                

                // 等待其他线程

                System.out.printf("%s: waiting for other threads. /n", thName);

                barrier.await();

                

                // 结束

                System.out.printf("%s: end. /n", thName);

            }

            catch (InterruptedException e)

            {

                System.out.printf("%s: interrupted. /n", thName);

                return;

            }

            catch (BrokenBarrierException e)

            {

                System.out.printf("%s: broken barrier. /n", thName);

                return;

            }

        }

    }

}

 

1.       在调用await方法时设置超时时间。
2.       当有一个await方法超时，所有处于await等待的线程将收到BrokenBarrierException，超时的线程自身将收到TimeoutException。
3.       BrokenBarrier之后的所有await调用将直接抛出BrokenBarrierException。
 

 

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;

public class CyclicBarrierBrokenTest

{

    private static CyclicBarrier barrier = null;

    

    public static void main(String[] args)

    {

        // 初始化

        barrier = new CyclicBarrier(10);

        

        // 启动10个线程

        System.out.println("main: starting 10 threads.");

        for (int i = 0; i < 10; i++)

        {

            Thread th = new Thread(new TestRunnable(), "thread " + i);

            th.start();

        }

    }

    

    private static class TestRunnable implements Runnable

    {

        @Override

        public void run()

        {

            String thName = Thread.currentThread().getName();

            try

            {

                // 随机等待5-14秒

                int n = (int)(Math.random() * 10 + 5);

                System.out.printf("%s: sleep for %d seconds. /n", thName, n);

                Thread.sleep(n * 1000);

                

                // 等待其他线程

                System.out.printf("%s: waiting for other threads. /n", thName);

                barrier.await(5, TimeUnit.SECONDS);

                

                // 结束

                System.out.printf("%s: end. /n", thName);

            }

            catch (InterruptedException e)

            {

                System.out.printf("%s: interrupted. /n", thName);

                return;

            }

            catch (BrokenBarrierException e)

            {

                System.out.printf("%s: broken barrier. /n", thName);

                return;

            }

            catch (TimeoutException e)

            {

                System.out.printf("%s: timeout. /n", thName);

                return;

            }

        }

    }

}

当调用await线程达到CyclicBarrier的设置之后，CyclicBarrier将被重置，重新等待n个await调用。
 

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierResetTest

{

    private static CyclicBarrier barrier = null;

    

    public static void main(String[] args) throws Exception

    {

        barrier = new CyclicBarrier(2);

        

        Thread th1 = new Thread(new TestRunnable(), "thread 1");

        Thread th2 = new Thread(new TestRunnable(), "thread 2");

        

        th1.start();

        th2.start();

        

        Thread.sleep(60 * 1000);

        

        th1.interrupt();

        th2.interrupt();

        

        th1.join();

        th2.join();

    }

    

    private static class TestRunnable implements Runnable

    {

        @Override

        public void run()

        {

            String thName = Thread.currentThread().getName();

            System.out.printf("%s: start. /n", thName);

            

            try

            {

                while (true)

                {

                    // 随机产生1-100的整数。

                    int c = (int)(Math.random() * 100) + 1;

                    

                    // 随机等待0-4秒。

                    int t = (int)(Math.random() * 5);

                    System.out.printf("%s: sleep for %d seconds./n", thName, t);

                    Thread.sleep(t * 1000);

                    

                    // 等待另1线程。

                    barrier.await();

                    

                    // 输出产生的整数。

                    System.out.printf("%s: %d /n", thName, c);

                    

                    // 等待其他线程输出。

                    Thread.sleep(100);

                }

            }

            catch (InterruptedException e)

            {

                System.out.printf("%s: interrupted. /n", thName);

            }

            catch (BrokenBarrierException e)

            {

                e.printStackTrace();

            }

            

            System.out.printf("%s: end. /n", thName);

        }

    }

}

 

1.       在创建CyclicBarrier的时候指定通过Barrier时需要执行的语句。
2.       通过构造方法的Runnable参数指定。
3.       Runnable中的语句将由最后达到Barrier的线程执行。

 

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierActionTest

{

    private static CyclicBarrier barrier = null;

    

    public static void main(String[] args) throws Exception

    {

        barrier = new CyclicBarrier(10, new BarrierAction());

        

        for (int i = 0; i < 10; i++)

        {

            Thread th = new Thread(new TestRunnable(), "thread " + i);

            th.start();

        }

    }

    

    private static class TestRunnable implements Runnable

    {

        @Override

        public void run()

        {

            try

            {

                String thName = Thread.currentThread().getName();

                // System.out.printf("%s: start. /n", thName);

                

                // 随机等待0-9秒

                int t = (int)(Math.random() * 10);

                System.out.printf("%s: sleep for %d seconds. /n", thName, t);

                Thread.sleep(t * 1000);

                

                // 等待barrier

                barrier.await();

                System.out.printf("%s: end. /n", thName);

            }

            catch (InterruptedException e)

            {

                e.printStackTrace();

                return;

            }

            catch (BrokenBarrierException e)

            {

                e.printStackTrace();

                return;

            }

        }

    }

    

    private static class BarrierAction implements Runnable

    {

        @Override

        public void run()

        {

            String thName = Thread.currentThread().getName();

            

            System.out.printf("%s: barrier action. /n", thName);

        }

    }

}