詳解Java編程中線程同步以及定時(shí)啟動(dòng)線程的方法

更新時(shí)間：2016年01月05日 09:21:23 作者：zhangjunhd

這篇文章主要介紹了詳解Java編程中線程同步以及定時(shí)啟動(dòng)線程的方法, 講到了wait()與notify()方法以及阻塞隊(duì)列等知識(shí),需要的朋友可以參考下

使用wait()與notify()實(shí)現(xiàn)線程間協(xié)作
1. wait()與notify()/notifyAll()
調(diào)用sleep()和yield()的時(shí)候鎖并沒(méi)有被釋放，而調(diào)用wait()將釋放鎖。這樣另一個(gè)任務(wù)（線程）可以獲得當(dāng)前對(duì)象的鎖，從而進(jìn)入它的synchronized方法中。可以通過(guò)notify()/notifyAll()，或者時(shí)間到期，從wait()中恢復(fù)執(zhí)行。
只能在同步控制方法或同步塊中調(diào)用wait()、notify()和notifyAll()。如果在非同步的方法里調(diào)用這些方法，在運(yùn)行時(shí)會(huì)拋出IllegalMonitorStateException異常。
2.模擬單個(gè)線程對(duì)多個(gè)線程的喚醒
模擬線程之間的協(xié)作。Game類有2個(gè)同步方法prepare()和go()。標(biāo)志位start用于判斷當(dāng)前線程是否需要wait()。Game類的實(shí)例首先啟動(dòng)所有的Athele類實(shí)例，使其進(jìn)入wait()狀態(tài)，在一段時(shí)間后，改變標(biāo)志位并notifyAll()所有處于wait狀態(tài)的Athele線程。
Game.java

package concurrency;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

class Athlete implements Runnable {
  private final int id;
  private Game game;

  public Athlete(int id, Game game) {
   this.id = id;
   this.game = game;
  }

  public boolean equals(Object o) {
   if (!(o instanceof Athlete))
    return false;
   Athlete athlete = (Athlete) o;
   return id == athlete.id;
  }

  public String toString() {
   return "Athlete<" + id + ">";
  }

  public int hashCode() {
   return new Integer(id).hashCode();
  }

  public void run() {
   try {
    game.prepare(this);
   } catch (InterruptedException e) {
    System.out.println(this + " quit the game");
   }
  }
 }

public class Game implements Runnable {
  private Set<Athlete> players = new HashSet<Athlete>();
  private boolean start = false;

  public void addPlayer(Athlete one) {
   players.add(one);
  }

  public void removePlayer(Athlete one) {
   players.remove(one);
  }

  public Collection<Athlete> getPlayers() {
   return Collections.unmodifiableSet(players);
  }

  public void prepare(Athlete athlete) throws InterruptedException {
   System.out.println(athlete + " ready!");
   synchronized (this) {
    while (!start)
    wait();
    if (start)
     System.out.println(athlete + " go!");
   }
  }

  public synchronized void go() {
   notifyAll();
  }
  
  public void ready() {
   Iterator<Athlete> iter = getPlayers().iterator();
   while (iter.hasNext())
    new Thread(iter.next()).start();
  }

  public void run() {
   start = false;
   System.out.println("Ready......");
   System.out.println("Ready......");
   System.out.println("Ready......");
   ready();
   start = true;
   System.out.println("Go!");
   go();
  }

  public static void main(String[] args) {
   Game game = new Game();
   for (int i = 0; i < 10; i++)
    game.addPlayer(new Athlete(i, game));
   new Thread(game).start();
  }
}

結(jié)果：

Ready......
Ready......
Ready......
Athlete<0> ready!
Athlete<1> ready!
Athlete<2> ready!
Athlete<3> ready!
Athlete<4> ready!
Athlete<5> ready!
Athlete<6> ready!
Athlete<7> ready!
Athlete<8> ready!
Athlete<9> ready!
Go!
Athlete<9> go!
Athlete<8> go!
Athlete<7> go!
Athlete<6> go!
Athlete<5> go!
Athlete<4> go!
Athlete<3> go!
Athlete<2> go!
Athlete<1> go!
Athlete<0> go!

3.模擬忙等待過(guò)程
MyObject類的實(shí)例是被觀察者，當(dāng)觀察事件發(fā)生時(shí)，它會(huì)通知一個(gè)Monitor類的實(shí)例（通知的方式是改變一個(gè)標(biāo)志位）。而此Monitor類的實(shí)例是通過(guò)忙等待來(lái)不斷的檢查標(biāo)志位是否變化。
BusyWaiting.java

import java.util.concurrent.TimeUnit;

class MyObject implements Runnable {
  private Monitor monitor;

  public MyObject(Monitor monitor) {
   this.monitor = monitor;
  }

  public void run() {
   try {
    TimeUnit.SECONDS.sleep(3);
    System.out.println("i'm going.");
    monitor.gotMessage();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}

class Monitor implements Runnable {
  private volatile boolean go = false;

  public void gotMessage() throws InterruptedException {
   go = true;
  }

  public void watching() {
   while (go == false)
    ;
   System.out.println("He has gone.");
  }

  public void run() {
   watching();
  }
}

public class BusyWaiting {
  public static void main(String[] args) {
   Monitor monitor = new Monitor();
   MyObject o = new MyObject(monitor);
   new Thread(o).start();
   new Thread(monitor).start();
  }
}

結(jié)果：

i'm going.
He has gone.

4.使用wait()與notify()改寫上面的例子
下面的例子通過(guò)wait()來(lái)取代忙等待機(jī)制，當(dāng)收到通知消息時(shí)，notify當(dāng)前Monitor類線程。
Wait.java

package concurrency.wait;

import java.util.concurrent.TimeUnit;

class MyObject implements Runnable {
  private Monitor monitor;

  public MyObject(Monitor monitor) {
   this.monitor = monitor;
  }

定時(shí)啟動(dòng)線程
這里提供兩種在指定時(shí)間后啟動(dòng)線程的方法。一是通過(guò)java.util.concurrent.DelayQueue實(shí)現(xiàn)；二是通過(guò)java.util.concurrent.ScheduledThreadPoolExecutor實(shí)現(xiàn)。
1. java.util.concurrent.DelayQueue
類DelayQueue是一個(gè)無(wú)界阻塞隊(duì)列，只有在延遲期滿時(shí)才能從中提取元素。它接受實(shí)現(xiàn)Delayed接口的實(shí)例作為元素。
<<interface>>Delayed.java

package java.util.concurrent;
import java.util.*;
public interface Delayed extends Comparable<Delayed> {
  long getDelay(TimeUnit unit);
}

getDelay()返回與此對(duì)象相關(guān)的剩余延遲時(shí)間，以給定的時(shí)間單位表示。此接口的實(shí)現(xiàn)必須定義一個(gè) compareTo 方法，該方法提供與此接口的 getDelay 方法一致的排序。

DelayQueue隊(duì)列的頭部是延遲期滿后保存時(shí)間最長(zhǎng)的 Delayed 元素。當(dāng)一個(gè)元素的getDelay(TimeUnit.NANOSECONDS) 方法返回一個(gè)小于等于 0 的值時(shí)，將發(fā)生到期。
2.設(shè)計(jì)帶有時(shí)間延遲特性的隊(duì)列
類DelayedTasker維護(hù)一個(gè)DelayQueue<DelayedTask> queue，其中DelayedTask實(shí)現(xiàn)了Delayed接口，并由一個(gè)內(nèi)部類定義。外部類和內(nèi)部類都實(shí)現(xiàn)Runnable接口，對(duì)于外部類來(lái)說(shuō)，它的run方法是按定義的時(shí)間先后取出隊(duì)列中的任務(wù)，而這些任務(wù)即內(nèi)部類的實(shí)例，內(nèi)部類的run方法定義每個(gè)線程具體邏輯。

這個(gè)設(shè)計(jì)的實(shí)質(zhì)是定義了一個(gè)具有時(shí)間特性的線程任務(wù)列表，而且該列表可以是任意長(zhǎng)度的。每次添加任務(wù)時(shí)指定啟動(dòng)時(shí)間即可。
DelayedTasker.java

package com.zj.timedtask;

import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.Collection;
import java.util.Collections;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class DelayedTasker implements Runnable {
  DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();

  public void addTask(DelayedTask e) {
    queue.put(e);
  }

  public void removeTask() {
    queue.poll();
  }

  public Collection<DelayedTask> getAllTasks() {
    return Collections.unmodifiableCollection(queue);
  }

  public int getTaskQuantity() {
    return queue.size();
  }

  public void run() {
    while (!queue.isEmpty())
      try {
       queue.take().run();
      } catch (InterruptedException e) {
       System.out.println("Interrupted");
      }
    System.out.println("Finished DelayedTask");
  }

  public static class DelayedTask implements Delayed, Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private final int delta;
    private final long trigger;

    public DelayedTask(int delayInSeconds) {
      delta = delayInSeconds;
      trigger = System.nanoTime() + NANOSECONDS.convert(delta, SECONDS);
    }

    public long getDelay(TimeUnit unit) {
      return unit.convert(trigger - System.nanoTime(), NANOSECONDS);
    }

    public int compareTo(Delayed arg) {
      DelayedTask that = (DelayedTask) arg;
      if (trigger < that.trigger)
       return -1;
      if (trigger > that.trigger)
       return 1;
      return 0;
    }

    public void run() {
      //run all that you want to do
      System.out.println(this);
    }

    public String toString() {
      return "[" + delta + "s]" + "Task" + id;
    }
  }

  public static void main(String[] args) {
    Random rand = new Random();
    ExecutorService exec = Executors.newCachedThreadPool();
    DelayedTasker tasker = new DelayedTasker();
    for (int i = 0; i < 10; i++)
      tasker.addTask(new DelayedTask(rand.nextInt(5)));
    exec.execute(tasker);
    exec.shutdown();
  }
}

結(jié)果：

[0s]Task 1
[0s]Task 2
[0s]Task 3
[1s]Task 6
[2s]Task 5
[3s]Task 8
[4s]Task 0
[4s]Task 4
[4s]Task 7
[4s]Task 9
Finished DelayedTask

3. java.util.concurrent.ScheduledThreadPoolExecutor
該類可以另行安排在給定的延遲后運(yùn)行任務(wù)（線程），或者定期（重復(fù)）執(zhí)行任務(wù)。在構(gòu)造子中需要知道線程池的大小。最主要的方法是：

[1] schedule
public ScheduledFuture<?> schedule(Runnable command, long delay,TimeUnit unit)
創(chuàng)建并執(zhí)行在給定延遲后啟用的一次性操作。
指定者：
-接口 ScheduledExecutorService 中的 schedule;
參數(shù)：
-command - 要執(zhí)行的任務(wù) ;
-delay - 從現(xiàn)在開(kāi)始延遲執(zhí)行的時(shí)間 ;
-unit - 延遲參數(shù)的時(shí)間單位 ;
返回：
-表示掛起任務(wù)完成的 ScheduledFuture，并且其 get() 方法在完成后將返回 null。

[2] scheduleAtFixedRate
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command,long initialDelay,long period,TimeUnit unit)
創(chuàng)建并執(zhí)行一個(gè)在給定初始延遲后首次啟用的定期操作，后續(xù)操作具有給定的周期；也就是將在 initialDelay 后開(kāi)始執(zhí)行，然后在 initialDelay+period 后執(zhí)行，接著在 initialDelay + 2 * period 后執(zhí)行，依此類推。如果任務(wù)的任何一個(gè)執(zhí)行遇到異常，則后續(xù)執(zhí)行都會(huì)被取消。否則，只能通過(guò)執(zhí)行程序的取消或終止方法來(lái)終止該任務(wù)。如果此任務(wù)的任何一個(gè)執(zhí)行要花費(fèi)比其周期更長(zhǎng)的時(shí)間，則將推遲后續(xù)執(zhí)行，但不會(huì)同時(shí)執(zhí)行。
指定者：
-接口 ScheduledExecutorService 中的 scheduleAtFixedRate;
參數(shù)：
-command - 要執(zhí)行的任務(wù) ;
-initialDelay - 首次執(zhí)行的延遲時(shí)間 ;
-period - 連續(xù)執(zhí)行之間的周期 ;
-unit - initialDelay 和 period 參數(shù)的時(shí)間單位 ;
返回：
-表示掛起任務(wù)完成的 ScheduledFuture，并且其 get() 方法在取消后將拋出異常。
4.設(shè)計(jì)帶有時(shí)間延遲特性的線程執(zhí)行者
類ScheduleTasked關(guān)聯(lián)一個(gè)ScheduledThreadPoolExcutor，可以指定線程池的大小。通過(guò)schedule方法知道線程及延遲的時(shí)間，通過(guò)shutdown方法關(guān)閉線程池。對(duì)于具體任務(wù)（線程）的邏輯具有一定的靈活性（相比前一中設(shè)計(jì)，前一種設(shè)計(jì)必須事先定義線程的邏輯，但可以通過(guò)繼承或裝飾修改線程具體邏輯設(shè)計(jì)）。
ScheduleTasker.java

package com.zj.timedtask;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ScheduleTasker {
  private int corePoolSize = 10;
  ScheduledThreadPoolExecutor scheduler;

  public ScheduleTasker() {
    scheduler = new ScheduledThreadPoolExecutor(corePoolSize);
  }

  public ScheduleTasker(int quantity) {
    corePoolSize = quantity;
    scheduler = new ScheduledThreadPoolExecutor(corePoolSize);
  }

  public void schedule(Runnable event, long delay) {
    scheduler.schedule(event, delay, TimeUnit.SECONDS);
  }

  public void shutdown() {
    scheduler.shutdown();
  }

  public static void main(String[] args) {
    ScheduleTasker tasker = new ScheduleTasker();
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[1s]Task 1");
      }
    }, 1);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[2s]Task 2");
      }
    }, 2);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[4s]Task 3");
      }
    }, 4);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[10s]Task 4");
      }
    }, 10);

    tasker.shutdown();
  }
}

結(jié)果：

[1s]Task 1
[2s]Task 2
[4s]Task 3
[10s]Task 4
  public void run() {
   try {
    TimeUnit.SECONDS.sleep(3);
    System.out.println("i'm going.");
    monitor.gotMessage();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}

class Monitor implements Runnable {
  private volatile boolean go = false;

  public synchronized void gotMessage() throws InterruptedException {
   go = true;
   notify();
  }

  public synchronized void watching() throws InterruptedException {
   while (go == false)
    wait();
   System.out.println("He has gone.");
  }

  public void run() {
   try {
    watching();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}

public class Wait {
  public static void main(String[] args) {
   Monitor monitor = new Monitor();
   MyObject o = new MyObject(monitor);
   new Thread(o).start();
   new Thread(monitor).start();
  }
}

結(jié)果：