基于Java回顾之多线程同步的使用详解

首先阐述什么是同步,不同步有什么问题,然后讨论可以采取哪些措施控制同步,接下来我们会仿照回顾网络通信时那样,构建一个服务器端的“线程池”,JDK为我们提供了一个很大的concurrent工具包,最后我们会对里面的内容进行探索。

为什么要线程同步?

说到线程同步,大部分情况下, 我们是在针对“单对象多线程”的情况进行讨论,一般会将其分成两部分,一部分是关于“共享变量”,一部分关于“执行步骤”。

共享变量

当我们在线程对象(Runnable)中定义了全局变量,run方法会修改该变量时,如果有多个线程同时使用该线程对象,那么就会造成全局变量的值被同时修改,造成错误。我们来看下面的代码:


代码如下:

共享变量造成同步问题
 class MyRunner implements Runnable
 {
     public int sum = 0;

public void run()
     {
         System.out.println(Thread.currentThread().getName() + " Start.");
         for (int i = 1; i <= 100; i++)
         {
             sum += i;
         }
         try {
             Thread.sleep(500);
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
         System.out.println(Thread.currentThread().getName() + " --- The value of sum is " + sum);
         System.out.println(Thread.currentThread().getName() + " End.");
     }
 }

private static void sharedVaribleTest() throws InterruptedException
 {
     MyRunner runner = new MyRunner();
     Thread thread1 = new Thread(runner);
     Thread thread2 = new Thread(runner);
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

这个示例中,线程用来计算1到100的和是多少,我们知道正确结果是5050(好像是高斯小时候玩过这个?),但是上述程序返回的结果是10100,原因是两个线程同时对sum进行操作。

执行步骤

我们在多个线程运行时,可能需要某些操作合在一起作为“原子操作”,即在这些操作可以看做是“单线程”的,例如我们可能希望输出结果的样子是这样的:


代码如下:

线程1:步骤1
 线程1:步骤2
 线程1:步骤3
 线程2:步骤1
 线程2:步骤2
 线程2:步骤3

如果同步控制不好,出来的样子可能是这样的:


代码如下:

线程1:步骤1
线程2:步骤1
线程1:步骤2
线程2:步骤2
线程1:步骤3
线程2:步骤3

这里我们也给出一个示例代码:


代码如下:

执行步骤混乱带来的同步问题
 class MyNonSyncRunner implements Runnable
 {
     public void run() {
         System.out.println(Thread.currentThread().getName() + " Start.");
         for(int i = 1; i <= 5; i++)
         {
             System.out.println(Thread.currentThread().getName() + " Running step " + i);
             try
             {
                 Thread.sleep(50);
             }
             catch(InterruptedException ex)
             {
                 ex.printStackTrace();
             }
         }
         System.out.println(Thread.currentThread().getName() + " End.");
     }
 }

private static void syncTest() throws InterruptedException
 {
     MyNonSyncRunner runner = new MyNonSyncRunner();
     Thread thread1 = new Thread(runner);
     Thread thread2 = new Thread(runner);
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

如何控制线程同步

既然线程同步有上述问题,那么我们应该如何去解决呢?针对不同原因造成的同步问题,我们可以采取不同的策略。

控制共享变量

我们可以采取3种方式来控制共享变量。

将“单对象多线程”修改成“多对象多线程”

上文提及,同步问题一般发生在“单对象多线程”的场景中,那么最简单的处理方式就是将运行模型修改成“多对象多线程”的样子,针对上面示例中的同步问题,修改后的代码如下:


代码如下:

解决共享变量问题方案一
 private static void sharedVaribleTest2() throws InterruptedException
 {
     Thread thread1 = new Thread(new MyRunner());
     Thread thread2 = new Thread(new MyRunner());
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

我们可以看到,上述代码中两个线程使用了两个不同的Runnable实例,它们在运行过程中,就不会去访问同一个全局变量。
将“全局变量”降级为“局部变量”

既然是共享变量造成的问题,那么我们可以将共享变量改为“不共享”,即将其修改为局部变量。这样也可以解决问题,同样针对上面的示例,这种解决方式的代码如下:


代码如下:

解决共享变量问题方案二
 class MyRunner2 implements Runnable
 {
     public void run()
     {
         System.out.println(Thread.currentThread().getName() + " Start.");
         int sum = 0;
         for (int i = 1; i <= 100; i++)
         {
             sum += i;
         }
         try {
             Thread.sleep(500);
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
         System.out.println(Thread.currentThread().getName() + " --- The value of sum is " + sum);
         System.out.println(Thread.currentThread().getName() + " End.");
     }
 }

private static void sharedVaribleTest3() throws InterruptedException
 {
     MyRunner2 runner = new MyRunner2();
     Thread thread1 = new Thread(runner);
     Thread thread2 = new Thread(runner);
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

我们可以看出,sum变量已经由全局变量变为run方法内部的局部变量了。
使用ThreadLocal机制

ThreadLocal是JDK引入的一种机制,它用于解决线程间共享变量,使用ThreadLocal声明的变量,即使在线程中属于全局变量,针对每个线程来讲,这个变量也是独立的。

我们可以用这种方式来改造上面的代码,如下所示:


代码如下:

解决共享变量问题方案三
 class MyRunner3 implements Runnable
 {
     public ThreadLocal<Integer> tl = new ThreadLocal<Integer>();

public void run()
     {
         System.out.println(Thread.currentThread().getName() + " Start.");
         for (int i = 0; i <= 100; i++)
         {
             if (tl.get() == null)
             {
                 tl.set(new Integer(0));
             }
             int sum = ((Integer)tl.get()).intValue();
             sum+= i;
             tl.set(new Integer(sum));
             try {
                 Thread.sleep(10);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
         }

System.out.println(Thread.currentThread().getName() + " --- The value of sum is " + ((Integer)tl.get()).intValue());
         System.out.println(Thread.currentThread().getName() + " End.");
     }
 }

private static void sharedVaribleTest4() throws InterruptedException
 {
     MyRunner3 runner = new MyRunner3();
     Thread thread1 = new Thread(runner);
     Thread thread2 = new Thread(runner);
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

综上三种方案,第一种方案会降低多线程执行的效率,因此,我们推荐使用第二种或者第三种方案。

控制执行步骤

说到执行步骤,我们可以使用synchronized关键字来解决它。


代码如下:

执行步骤问题解决方案
 class MySyncRunner implements Runnable
 {
     public void run() {
         synchronized(this)
         {
             System.out.println(Thread.currentThread().getName() + " Start.");
             for(int i = 1; i <= 5; i++)
             {
                 System.out.println(Thread.currentThread().getName() + " Running step " + i);
                 try
                 {
                     Thread.sleep(50);
                 }
                 catch(InterruptedException ex)
                 {
                     ex.printStackTrace();
                 }
             }
             System.out.println(Thread.currentThread().getName() + " End.");
         }
     }
 }

private static void syncTest2() throws InterruptedException
 {
     MySyncRunner runner = new MySyncRunner();
     Thread thread1 = new Thread(runner);
     Thread thread2 = new Thread(runner);
     thread1.setDaemon(true);
     thread2.setDaemon(true);
     thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

在线程同步的话题上,synchronized是一个非常重要的关键字。它的原理和数据库中事务锁的原理类似。我们在使用过程中,应该尽量缩减synchronized覆盖的范围,原因有二:1)被它覆盖的范围是串行的,效率低;2)容易产生死锁。我们来看下面的示例:


代码如下:

synchronized示例
 private static void syncTest3() throws InterruptedException
 {
     final List<Integer> list = new ArrayList<Integer>();

Thread thread1 = new Thread()
     {
         public void run()
         {
             System.out.println(Thread.currentThread().getName() + " Start.");
             Random r = new Random(100);
             synchronized(list)
             {
                 for (int i = 0; i < 5; i++)
                 {
                     list.add(new Integer(r.nextInt()));
                 }
                 System.out.println("The size of list is " + list.size());
             }
             try
             {
                 Thread.sleep(500);
             }
             catch(InterruptedException ex)
             {
                 ex.printStackTrace();
             }
             System.out.println(Thread.currentThread().getName() + " End.");
         }
     };

Thread thread2 = new Thread()
     {
         public void run()
         {
             System.out.println(Thread.currentThread().getName() + " Start.");
             Random r = new Random(100);
             synchronized(list)
             {
                 for (int i = 0; i < 5; i++)
                 {
                     list.add(new Integer(r.nextInt()));
                 }
                 System.out.println("The size of list is " + list.size());
             }
             try
             {
                 Thread.sleep(500);
             }
             catch(InterruptedException ex)
             {
                 ex.printStackTrace();
             }
             System.out.println(Thread.currentThread().getName() + " End.");
         }
     };

thread1.start();
     thread2.start();
     thread1.join();
     thread2.join();
 }

我们应该把需要同步的内容集中在一起,尽量不包含其他不相关的、消耗大量资源的操作,示例中线程休眠的操作显然不应该包括在里面。

构造线程池

我们在<基于Java回顾之网络通信的应用分析>中,已经构建了一个Socket连接池,这里我们在此基础上,构建一个线程池,完成基本的启动、休眠、唤醒、停止操作。

基本思路还是以数组的形式保持一系列线程,通过Socket通信,客户端向服务器端发送命令,当服务器端接收到命令后,根据收到的命令对线程数组中的线程进行操作。

Socket客户端的代码保持不变,依然采用构建Socket连接池时的代码,我们主要针对服务器端进行改造。

首先,我们需要定义一个线程对象,它用来执行我们的业务操作,这里简化起见,只让线程进行休眠。


代码如下:

定义线程对象
 enum ThreadStatus
 {
     Initial,
     Running,
     Sleeping,
     Stopped
 }

enum ThreadTask
 {
     Start,
     Stop,
     Sleep,
     Wakeup
 }

class MyThread extends Thread
 {
     public ThreadStatus status = ThreadStatus.Initial;
     public ThreadTask task;
     public void run()
     {
         status = ThreadStatus.Running;
         while(true)
         {
             try {
                 Thread.sleep(3000);
                 if (status == ThreadStatus.Sleeping)
                 {
                     System.out.println(Thread.currentThread().getName() + " 进入休眠状态。");
                     this.wait();
                 }
             } catch (InterruptedException e) {
                 System.out.println(Thread.currentThread().getName() + " 运行过程中出现错误。");
                 status = ThreadStatus.Stopped;
             }
         }
     }
 }

然后,我们需要定义一个线程管理器,它用来对线程池中的线程进行管理,代码如下:


代码如下:

定义线程池管理对象
 class MyThreadManager
 {
     public static void manageThread(MyThread[] threads, ThreadTask task)
     {
         for (int i = 0; i < threads.length; i++)
         {
             synchronized(threads[i])
             {
                 manageThread(threads[i], task);
             }
         }
         System.out.println(getThreadStatus(threads));
     }

public static void manageThread(MyThread thread, ThreadTask task)
     {
         if (task == ThreadTask.Start)
         {
             if (thread.status == ThreadStatus.Running)
             {
                 return;
             }
             if (thread.status == ThreadStatus.Stopped)
             {
                 thread = new MyThread();
             }
             thread.status = ThreadStatus.Running;
             thread.start();

}
         else if (task == ThreadTask.Stop)
         {
             if (thread.status != ThreadStatus.Stopped)
             {
                 thread.interrupt();
                 thread.status = ThreadStatus.Stopped;
             }
         }
         else if (task == ThreadTask.Sleep)
         {
             thread.status = ThreadStatus.Sleeping;
         }
         else if (task == ThreadTask.Wakeup)
         {
             thread.notify();
             thread.status = ThreadStatus.Running;
         }
     }

public static String getThreadStatus(MyThread[] threads)
     {
         StringBuffer sb = new StringBuffer();
         for (int i = 0; i < threads.length; i++)
         {
             sb.append(threads[i].getName() + "的状态:" + threads[i].status).append("\r\n");
         }
         return sb.toString();
     }
 }

最后,是我们的服务器端,它不断接受客户端的请求,每收到一个连接请求,服务器端会新开一个线程,来处理后续客户端发来的各种操作指令。


代码如下:

定义服务器端线程池对象
 public class MyThreadPool {

public static void main(String[] args) throws IOException
     {
         MyThreadPool pool = new MyThreadPool(5);
     }

private int threadCount;
     private MyThread[] threads = null;

public MyThreadPool(int count) throws IOException
     {
         this.threadCount = count;
         threads = new MyThread[count];
         for (int i = 0; i < threads.length; i++)
         {
             threads[i] = new MyThread();
             threads[i].start();
         }
         Init();
     }

private void Init() throws IOException
     {
         ServerSocket serverSocket = new ServerSocket(5678);
         while(true)
         {
             final Socket socket = serverSocket.accept();
             Thread thread = new Thread()
             {
                 public void run()
                 {
                     try
                     {
                         System.out.println("检测到一个新的Socket连接。");
                         BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                         PrintStream ps = new PrintStream(socket.getOutputStream());
                         String line = null;
                         while((line = br.readLine()) != null)
                         {
                             System.out.println(line);
                             if (line.equals("Count"))
                             {
                                 System.out.println("线程池中有5个线程");
                             }
                             else if (line.equals("Status"))
                             {
                                 String status = MyThreadManager.getThreadStatus(threads);
                                 System.out.println(status);
                             }
                             else if (line.equals("StartAll"))
                             {
                                 MyThreadManager.manageThread(threads, ThreadTask.Start);
                             }
                             else if (line.equals("StopAll"))
                             {
                                 MyThreadManager.manageThread(threads, ThreadTask.Stop);
                             }
                             else if (line.equals("SleepAll"))
                             {
                                 MyThreadManager.manageThread(threads, ThreadTask.Sleep);
                             }
                             else if (line.equals("WakeupAll"))
                             {
                                 MyThreadManager.manageThread(threads, ThreadTask.Wakeup);
                             }
                             else if (line.equals("End"))
                             {
                                 break;
                             }
                             else
                             {
                                 System.out.println("Command:" + line);
                             }
                             ps.println("OK");
                             ps.flush();
                         }
                     }
                     catch(Exception ex)
                     {
                         ex.printStackTrace();
                     }
                 }
             };
             thread.start();
         }
     }
 }

探索JDK中的concurrent工具包

为了简化开发人员在进行多线程开发时的工作量,并减少程序中的bug,JDK提供了一套concurrent工具包,我们可以用它来方便的开发多线程程序。
线程池

我们在上面实现了一个非常“简陋”的线程池,concurrent工具包中也提供了线程池,而且使用非常方便。

concurrent工具包中的线程池分为3类:ScheduledThreadPool、FixedThreadPool和CachedThreadPool。

首先我们来定义一个Runnable的对象


代码如下:

定义Runnable对象
 class MyRunner implements Runnable
 {
     public void run() {
         System.out.println(Thread.currentThread().getName() + "运行开始");
         for(int i = 0; i < 1; i++)
         {
             try
             {
                 System.out.println(Thread.currentThread().getName() + "正在运行");
                 Thread.sleep(200);
             }
             catch(Exception ex)
             {
                 ex.printStackTrace();
             }
         }
         System.out.println(Thread.currentThread().getName() + "运行结束");
     }
 }

可以看出,它的功能非常简单,只是输出了线程的执行过程。

ScheduledThreadPool

这和我们平时使用的ScheduledTask比较类似,或者说很像Timer,它可以使得一个线程在指定的一段时间内开始运行,并且在间隔另外一段时间后再次运行,直到线程池关闭。

示例代码如下:


代码如下:

ScheduledThreadPool示例
 private static void scheduledThreadPoolTest()
 {
     final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);

MyRunner runner = new MyRunner();

final ScheduledFuture<?> handler1 = scheduler.scheduleAtFixedRate(runner, 1, 10, TimeUnit.SECONDS);
     final ScheduledFuture<?> handler2 = scheduler.scheduleWithFixedDelay(runner, 2, 10, TimeUnit.SECONDS);

scheduler.schedule(new Runnable()
     {
         public void run()
         {
             handler1.cancel(true);
             handler2.cancel(true);
             scheduler.shutdown();
         }
     }, 30, TimeUnit.SECONDS
     );
 }

FixedThreadPool

这是一个指定容量的线程池,即我们可以指定在同一时间,线程池中最多有多个线程在运行,超出的线程,需要等线程池中有空闲线程时,才能有机会运行。

来看下面的代码:


代码如下:

FixedThreadPool示例
 private static void fixedThreadPoolTest()
 {
     ExecutorService exec = Executors.newFixedThreadPool(3);
     for(int i = 0; i < 5; i++)
     {
         MyRunner runner = new MyRunner();
         exec.execute(runner);
     }
     exec.shutdown();
 }

注意它的输出结果:


代码如下:

pool-1-thread-1运行开始
pool-1-thread-1正在运行
pool-1-thread-2运行开始
pool-1-thread-2正在运行
pool-1-thread-3运行开始
pool-1-thread-3正在运行
pool-1-thread-1运行结束
pool-1-thread-1运行开始
pool-1-thread-1正在运行
pool-1-thread-2运行结束
pool-1-thread-2运行开始
pool-1-thread-2正在运行
pool-1-thread-3运行结束
pool-1-thread-1运行结束
pool-1-thread-2运行结束

可以看到从始至终,最多有3个线程在同时运行。
CachedThreadPool

这是另外一种线程池,它不需要指定容量,只要有需要,它就会创建新的线程。

它的使用方式和FixedThreadPool非常像,来看下面的代码:


代码如下:

CachedThreadPool示例
 private static void cachedThreadPoolTest()
 {
     ExecutorService exec = Executors.newCachedThreadPool();
     for(int i = 0; i < 5; i++)
     {
         MyRunner runner = new MyRunner();
         exec.execute(runner);
     }
     exec.shutdown();
 }

它的执行结果如下:


代码如下:

pool-1-thread-1运行开始
pool-1-thread-1正在运行
pool-1-thread-2运行开始
pool-1-thread-2正在运行
pool-1-thread-3运行开始
pool-1-thread-3正在运行
pool-1-thread-4运行开始
pool-1-thread-4正在运行
pool-1-thread-5运行开始
pool-1-thread-5正在运行
pool-1-thread-1运行结束
pool-1-thread-2运行结束
pool-1-thread-3运行结束
pool-1-thread-4运行结束
pool-1-thread-5运行结束

可以看到,它创建了5个线程。
处理线程返回值

在有些情况下,我们需要使用线程的返回值,在上述的所有代码中,线程这是执行了某些操作,没有任何返回值。

如何做到这一点呢?我们可以使用JDK中的Callable<T>和CompletionService<T>,前者返回单个线程的结果,后者返回一组线程的结果。
返回单个线程的结果

还是直接看代码吧:


代码如下:

Callable示例
 private static void callableTest() throws InterruptedException, ExecutionException
 {
     ExecutorService exec = Executors.newFixedThreadPool(1);
     Callable<String> call = new Callable<String>()
     {
         public String call()
         {
             return "Hello World.";
         }
     };
     Future<String> result = exec.submit(call);
     System.out.println("线程的返回值是" + result.get());
     exec.shutdown();
 }

执行结果如下:


代码如下:

线程的返回值是Hello World.

返回线程池中每个线程的结果

这里需要使用CompletionService<T>,代码如下:


代码如下:

CompletionService示例
 private static void completionServiceTest() throws InterruptedException, ExecutionException
 {
     ExecutorService exec = Executors.newFixedThreadPool(10);
     CompletionService<String> service = new ExecutorCompletionService<String>(exec);
     for (int i = 0; i < 10; i++)
     {
         Callable<String> call = new Callable<String>()
         {
             public String call() throws InterruptedException
             {
                 return Thread.currentThread().getName();
             }
         };
         service.submit(call);
     }

Thread.sleep(1000);
     for(int i = 0; i < 10; i++)
     {
         Future<String> result = service.take();
         System.out.println("线程的返回值是" + result.get());
     }
     exec.shutdown();
 }

执行结果如下:


代码如下:

线程的返回值是pool-2-thread-1
线程的返回值是pool-2-thread-2
线程的返回值是pool-2-thread-3
线程的返回值是pool-2-thread-5
线程的返回值是pool-2-thread-4
线程的返回值是pool-2-thread-6
线程的返回值是pool-2-thread-8
线程的返回值是pool-2-thread-7
线程的返回值是pool-2-thread-9
线程的返回值是pool-2-thread-10

实现生产者-消费者模型

对于生产者-消费者模型来说,我们应该都不会陌生,通常我们都会使用某种数据结构来实现它。在concurrent工具包中,我们可以使用BlockingQueue来实现生产者-消费者模型,如下:


代码如下:

BlockingQueue示例
 public class BlockingQueueSample {

public static void main(String[] args)
     {
         blockingQueueTest();
     }

private static void blockingQueueTest()
     {
         final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
         final int maxSleepTimeForSetter = 10;
         final int maxSleepTimerForGetter = 10;

Runnable setter = new Runnable()
         {
             public void run()
             {
                 Random r = new Random();
                 while(true)
                 {
                     int value = r.nextInt(100);
                     try
                     {
                         queue.put(new Integer(value));
                         System.out.println(Thread.currentThread().getName() + "---向队列中插入值" + value);
                         Thread.sleep(r.nextInt(maxSleepTimeForSetter) * 1000);
                     }
                     catch(Exception ex)
                     {
                         ex.printStackTrace();
                     }
                 }
             }
         };

Runnable getter = new Runnable()
         {
             public void run()
             {
                 Random r = new Random();
                 while(true)
                 {
                     try
                     {
                         if (queue.size() == 0)
                         {
                             System.out.println(Thread.currentThread().getName() + "---队列为空");
                         }
                         else
                         {
                             int value = queue.take().intValue();
                             System.out.println(Thread.currentThread().getName() + "---从队列中获取值" + value);
                         }
                         Thread.sleep(r.nextInt(maxSleepTimerForGetter) * 1000);
                     }
                     catch(Exception ex)
                     {
                         ex.printStackTrace();
                     }
                 }
             }
         };

ExecutorService exec = Executors.newFixedThreadPool(2);
         exec.execute(setter);
         exec.execute(getter);
     }
 }

我们定义了两个线程,一个线程向Queue中添加数据,一个线程从Queue中取数据。我们可以通过控制maxSleepTimeForSetter和maxSleepTimerForGetter的值,来使得程序得出不同的结果。

可能的执行结果如下:


代码如下:

pool-1-thread-1---向队列中插入值88
pool-1-thread-2---从队列中获取值88
pool-1-thread-1---向队列中插入值75
pool-1-thread-2---从队列中获取值75
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-1---向队列中插入值50
pool-1-thread-2---从队列中获取值50
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-2---队列为空
pool-1-thread-1---向队列中插入值51
pool-1-thread-1---向队列中插入值92
pool-1-thread-2---从队列中获取值51
pool-1-thread-2---从队列中获取值92

因为Queue中的值和Thread的休眠时间都是随机的,所以执行结果也不是固定的。

使用信号量来控制线程

JDK提供了Semaphore来实现“信号量”的功能,它提供了两个方法分别用于获取和释放信号量:acquire和release,示例代码如下:


代码如下:

SemaPhore示例
 private static void semaphoreTest()
 {
     ExecutorService exec = Executors.newFixedThreadPool(10);
     final Semaphore semp = new Semaphore(2);

for (int i = 0; i < 10; i++)
     {
         Runnable runner = new Runnable()
         {
             public void run()
             {
                 try
                 {
                     semp.acquire();
                     System.out.println(new Date() + " " + Thread.currentThread().getName() + "正在执行。");
                     Thread.sleep(5000);
                     semp.release();
                 }
                 catch(Exception ex)
                 {
                     ex.printStackTrace();
                 }
             }
         };
         exec.execute(runner);
     }

exec.shutdown();
 }

执行结果如下:


代码如下:

Tue May 07 11:22:11 CST 2013 pool-1-thread-1正在执行。
Tue May 07 11:22:11 CST 2013 pool-1-thread-2正在执行。
Tue May 07 11:22:17 CST 2013 pool-1-thread-3正在执行。
Tue May 07 11:22:17 CST 2013 pool-1-thread-4正在执行。
Tue May 07 11:22:22 CST 2013 pool-1-thread-5正在执行。
Tue May 07 11:22:22 CST 2013 pool-1-thread-6正在执行。
Tue May 07 11:22:27 CST 2013 pool-1-thread-7正在执行。
Tue May 07 11:22:27 CST 2013 pool-1-thread-8正在执行。
Tue May 07 11:22:32 CST 2013 pool-1-thread-10正在执行。
Tue May 07 11:22:32 CST 2013 pool-1-thread-9正在执行。

可以看出,尽管线程池中创建了10个线程,但是同时运行的,只有2个线程。
控制线程池中所有线程的执行步骤

在前面,我们已经提到,可以用synchronized关键字来控制单个线程中的执行步骤,那么如果我们想要对线程池中的所有线程的执行步骤进行控制的话,应该如何实现呢?

我们有两种方式,一种是使用CyclicBarrier,一种是使用CountDownLatch。

CyclicBarrier使用了类似于Object.wait的机制,它的构造函数中需要接收一个整型数字,用来说明它需要控制的线程数目,当在线程的run方法中调用它的await方法时,它会保证所有的线程都执行到这一步,才会继续执行后面的步骤。

示例代码如下:


代码如下:

CyclicBarrier示例
 class MyRunner2 implements Runnable
 {
     private CyclicBarrier barrier = null;
     public MyRunner2(CyclicBarrier barrier)
     {
         this.barrier = barrier;
     }

public void run() {
         Random r = new Random();
         try
         {
             for (int i = 0; i < 3; i++)
             {
                 Thread.sleep(r.nextInt(10) * 1000);
                 System.out.println(new Date() + "--" + Thread.currentThread().getName() + "--第" + (i + 1) + "次等待。");
                 barrier.await();
             }
         }
         catch(Exception ex)
         {
             ex.printStackTrace();
         }
     }

}

private static void cyclicBarrierTest()
 {
     CyclicBarrier barrier = new CyclicBarrier(3);

ExecutorService exec = Executors.newFixedThreadPool(3);
     for (int i = 0; i < 3; i++)
     {
         exec.execute(new MyRunner2(barrier));
     }
     exec.shutdown();
 }

执行结果如下:


代码如下:

Tue May 07 11:31:20 CST 2013--pool-1-thread-2--第1次等待。
Tue May 07 11:31:21 CST 2013--pool-1-thread-3--第1次等待。
Tue May 07 11:31:24 CST 2013--pool-1-thread-1--第1次等待。
Tue May 07 11:31:24 CST 2013--pool-1-thread-1--第2次等待。
Tue May 07 11:31:26 CST 2013--pool-1-thread-3--第2次等待。
Tue May 07 11:31:30 CST 2013--pool-1-thread-2--第2次等待。
Tue May 07 11:31:32 CST 2013--pool-1-thread-1--第3次等待。
Tue May 07 11:31:33 CST 2013--pool-1-thread-3--第3次等待。
Tue May 07 11:31:33 CST 2013--pool-1-thread-2--第3次等待。

可以看出,thread-2到第1次等待点时,一直等到thread-1到达后才继续执行。

CountDownLatch则是采取类似”倒计时计数器”的机制来控制线程池中的线程,它有CountDown和Await两个方法。示例代码如下:


代码如下:

CountDownLatch示例
 private static void countdownLatchTest() throws InterruptedException
 {
     final CountDownLatch begin = new CountDownLatch(1);
     final CountDownLatch end = new CountDownLatch(5);
     ExecutorService exec = Executors.newFixedThreadPool(5);
     for (int i = 0; i < 5; i++)
     {
         Runnable runner = new Runnable()
         {
             public void run()
             {
                 Random r = new Random();
                 try
                 {
                     begin.await();
                     System.out.println(Thread.currentThread().getName() + "运行开始");
                     Thread.sleep(r.nextInt(10)*1000);
                     System.out.println(Thread.currentThread().getName() + "运行结束");
                 }
                 catch(Exception ex)
                 {
                     ex.printStackTrace();
                 }
                 finally
                 {
                     end.countDown();
                 }
             }
         };
         exec.execute(runner);
     }
     begin.countDown();
     end.await();
     System.out.println(Thread.currentThread().getName() + "运行结束");
     exec.shutdown();
 }

执行结果如下:


代码如下:

pool-1-thread-1运行开始
pool-1-thread-5运行开始
pool-1-thread-2运行开始
pool-1-thread-3运行开始
pool-1-thread-4运行开始
pool-1-thread-2运行结束
pool-1-thread-1运行结束
pool-1-thread-3运行结束
pool-1-thread-5运行结束
pool-1-thread-4运行结束
main运行结束

(0)

相关推荐

  • 深入理解JAVA多线程之线程间的通信方式

    一,介绍 本总结我对于JAVA多线程中线程之间的通信方式的理解,主要以代码结合文字的方式来讨论线程间的通信,故摘抄了书中的一些示例代码. 二,线程间的通信方式 ①同步 这里讲的同步是指多个线程通过synchronized关键字这种方式来实现线程间的通信. 参考示例: public class MyObject { synchronized public void methodA() { //do something.... } synchronized public void methodB()

  • Java多线程中线程间的通信实例详解

    Java多线程中线程间的通信 一.使用while方式来实现线程之间的通信 package com.ietree.multithread.sync; import java.util.ArrayList; import java.util.List; public class MyList { private volatile static List list = new ArrayList(); public void add() { list.add("apple"); } publ

  • java 多线程-线程通信实例讲解

    线程通信的目标是使线程间能够互相发送信号.另一方面,线程通信使线程能够等待其他线程的信号. 通过共享对象通信 忙等待 wait(),notify()和 notifyAll() 丢失的信号 假唤醒 多线程等待相同信号 不要对常量字符串或全局对象调用 wait() 通过共享对象通信 线程间发送信号的一个简单方式是在共享对象的变量里设置信号值.线程 A 在一个同步块里设置 boolean 型成员变量 hasDataToProcess 为 true,线程 B 也在同步块里读取 hasDataToProc

  • java多线程实现服务器端与多客户端之间的通信

    用java语言构建一个网络服务器,实现客户端和服务器之间通信,实现客户端拥有独立线程,互不干扰. 应用多线程来实现服务器与多线程之间的通信的基本步骤 服务器端创建ServerSocket,循环调用accept()等待客户端链接 客户端创建一个Socket并请求和服务器端链接 服务器端接受客户端请求,创建socekt与该客户端建立专线链接 建立链接的socket在一个单独的线程上对话 服务器继续等待新的链接 服务器端Server.java package test.concurrent.socke

  • 浅谈Java多线程实现及同步互斥通讯

    Java多线程深入理解本文主要从三个方面了解和掌握多线程: 1. 多线程的实现方式,通过继承Thread类和通过实现Runnable接口的方式以及异同点. 2. 多线程的同步与互斥中synchronized的使用方法. 3. 多线程的通讯中的notify(),notifyAll(),及wait(),的使用方法,以及简单的生成者和消费者的代码实现. 下面来具体的讲解Java中的多线程: 一:多线程的实现方式 通过继承Threa类来实现多线程主要分为以下三步: 第一步:继承 Thread,实现Thr

  • 详解Java多线程编程中的线程同步方法

    1.多线程的同步: 1.1.同步机制: 在多线程中,可能有多个线程试图访问一个有限的资源,必须预防这种情况的发生.所以引入了同步机制:在线程使用一个资源时为其加锁,这样其他的线程便不能访问那个资源了,直到解锁后才可以访问. 1.2.共享成员变量的例子: 成员变量与局部变量: 成员变量: 如果一个变量是成员变量,那么多个线程对同一个对象的成员变量进行操作,这多个线程是共享一个成员变量的. 局部变量: 如果一个变量是局部变量,那么多个线程对同一个对象进行操作,每个线程都会有一个该局部变量的拷贝.他们

  • 五种Java多线程同步的方法

    为什么要线程同步 因为当我们有多个线程要同时访问一个变量或对象时,如果这些线程中既有读又有写操作时,就会导致变量值或对象的状态出现混乱,从而导致程序异常.举 个例子,如果一个银行账户同时被两个线程操作,一个取100块,一个存钱100块.假设账户原本有0块,如果取钱线程和存钱线程同时发生,会出现什么结果 呢?取钱不成功,账户余额是100.取钱成功了,账户余额是0.那到底是哪个呢?很难说清楚.因此多线程同步就是要解决这个问题. 一.不同步时的代码 Bank.java package threadTe

  • 基于Java回顾之多线程同步的使用详解

    首先阐述什么是同步,不同步有什么问题,然后讨论可以采取哪些措施控制同步,接下来我们会仿照回顾网络通信时那样,构建一个服务器端的"线程池",JDK为我们提供了一个很大的concurrent工具包,最后我们会对里面的内容进行探索. 为什么要线程同步? 说到线程同步,大部分情况下, 我们是在针对"单对象多线程"的情况进行讨论,一般会将其分成两部分,一部分是关于"共享变量",一部分关于"执行步骤". 共享变量 当我们在线程对象(Run

  • 基于Java回顾之I/O的使用详解

    工作后,使用的技术随着项目的变化而变化,时而C#,时而Java,当然还有其他一些零碎的技术.总体而言,C#的使用时间要更长一些,其次是Java.我本身对语言没有什么倾向性,能干活的语言,就是好语言.而且从面向对象的角度来看,我觉得C#和Java对我来说,没什么区别. 这篇文章主要回顾Java中和I/O操作相关的内容,I/O也是编程语言的一个基础特性,Java中的I/O分为两种类型,一种是顺序读取,一种是随机读取. 我们先来看顺序读取,有两种方式可以进行顺序读取,一种是InputStream/Ou

  • 基于Java中进制的转换函数详解

    十进制转成十六进制: Integer.toHexString(int i) 十进制转成八进制 Integer.toOctalString(int i) 十进制转成二进制 Integer.toBinaryString(int i) 十六进制转成十进制 Integer.valueOf("FFFF",16).toString() 八进制转成十进制 Integer.valueOf("876",8).toString() 二进制转十进制 Integer.valueOf(&qu

  • 基于Java中throw和throws的区别(详解)

    系统自动抛出的异常 所有系统定义的编译和运行异常都可以由系统自动抛出,称为标准异常,并且 Java 强烈地要求应用程序进行完整的异常处理,给用户友好的提示,或者修正后使程序继续执行. 语句抛出的异常 用户程序自定义的异常和应用程序特定的异常,必须借助于 throws 和 throw 语句来定义抛出异常. throw是语句抛出一个异常. 语法:throw (异常对象); throw e; throws是方法可能抛出异常的声明.(用在声明方法时,表示该方法可能要抛出异常) 语法:[(修饰符)](返回

  • 基于Java回顾之多线程详解

    线程是操作系统运行的基本单位,它被封装在进程中,一个进程可以包含多个线程.即使我们不手动创造线程,进程也会有一个默认的线程在运行. 对于JVM来说,当我们编写一个单线程的程序去运行时,JVM中也是有至少两个线程在运行,一个是我们创建的程序,一个是垃圾回收. 线程基本信息 我们可以通过Thread.currentThread()方法获取当前线程的一些信息,并对其进行修改. 我们来看以下代码: 复制代码 代码如下: 查看并修改当前线程的属性 String name = Thread.currentT

  • 基于一个应用程序多线程误用的分析详解

    一.需求和初步实现很简单的一个windows服务:客户端连接邮件服务器,下载邮件(含附件)并保存为.eml格式,保存成功后删除服务器上的邮件.实现的伪代码大致如下: 复制代码 代码如下: public void Process()        {            var recordCount = 1000;//每次取出邮件记录数            while (true)            {                using (var client = new Pop

  • 基于java解析JSON的三种方式详解

    本文实例分析了基于java解析JSON的三种方式.分享给大家供大家参考,具体如下: 一.什么是JSON? JSON是一种取代XML的数据结构,和xml相比,它更小巧但描述能力却不差,由于它的小巧所以网络传输数据将减少更多流量从而加快速度. JSON就是一串字符串 只不过元素会使用特定的符号标注. {} 双括号表示对象 [] 中括号表示数组 "" 双引号内是属性或值 : 冒号表示后者是前者的值(这个值可以是字符串.数字.也可以是另一个数组或对象) 所以 {"name"

  • Java中的AQS同步队列问题详解

    目录 AQS 同步队列 1.AQS 介绍 1.1.类图关系 1.2.节点剖析 2.AQS 实现原理 2.1.队列初始化 2.2.追加节点 3.AQS 唤醒动作 AQS 同步队列 1.AQS 介绍 AQS 是 AbstractQueuedSynchronizer 的缩写,他是一个抽象同步类,为 JUC 包下的大多数同步工具提供了核心实现,例如 ReentrantLock 的底层就是使用同步队列.AQS 提供一套基础的机制来实现线程的同步.阻塞与唤醒.等待队列等功能,也就是想要深入学习线程工具类,这

  • java(jsp)整合discuz同步登录功能详解

    最近做了一个资源库系统的项目,老师说可以搭建开源论坛替代自己开发社交模块,正好在开源中国上看到了一个利用discuz的UCenter功能实现同步登录的开源项目(https://code.google.com/p/discuz-ucenter-api-for-java/),不禁大喜,于是花了几个小时照着教程操作了一遍,居然很轻松的成功了,特写此文以做纪念.. Uenter是Comsenz旗下各个产品之间信息直接传递的一个桥梁,通过UCenter站长可以无缝整合Comsenz系列产品,实现用户的一站

  • Spring中基于Java的配置@Configuration和@Bean用法详解

    一.首先,需要xml中进行少量的配置来启动Java配置: <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://ww

随机推荐