并发编程Java并发编程并发编程

七周七并发(线程与锁)

2016-08-31  本文已影响103人  Aaaaaaaaaron

1.概述

1.1并发还是并行(Concurrent or Parallel)

1.2 并行架构

位级（bit-level）并行

指令级（instruction-level）并行

数据级（data）并行 数据级并行

任务级（task-level）并行

1.3 并发：不只是多核

1.4 七个模型

线程与锁

函数式编程

Clojure之道——分离标识与状态

actor

通信顺序进程（Communicating Sequential Processes，CSP）

数据级并行

Lambda架构

2.线程与锁

2.1 简单粗暴

2.2 第一天:互斥和内存模型

竞态条件

内存可见性

class Counter {
    private int count = 0;
    public synchronized void increment() { ++count; } ➤
    public int getCount() { return count; }
}

死锁

class Philosopher extends Thread {
    private Chopstick left, right;
    private Random random;

    public Philosopher(Chopstick left, Chopstick right) {
        this.left = left; this.right = right;
                random = new Random();
    }

    public void run() {
        try {
            while(true) {
                    Thread.sleep(random.nextInt(1000)); // Think for a while
                synchronized(left) { // Grab left chopstick //
                    synchronized(right) { // Grab right chopstick // 15
                        Thread.sleep(random.nextInt(1000)); // Eat for a while
                    }
                }
            }
        } catch(InterruptedException e) {}
    }
}

class Philosopher extends Thread {
    private Chopstick first, second;
    private Random random;
    private int thinkCount;

    public Philosopher(Chopstick left, Chopstick right) {
        if(left.getId() < right.getId()) {
            first = left; second = right;
        } else {
            first = right; second = left;
        }
        random = new Random();
    }

    public void run() {
        try {
            while(true) {
                ++thinkCount;
                if (thinkCount % 10 == 0)
                    System.out.println("Philosopher " + this + " has thought " + thinkCount + " times");
                Thread.sleep(random.nextInt(1000));     // Think for a while
                synchronized(first) {                   // Grab first chopstick
                    synchronized(second) {                // Grab second chopstick
                        Thread.sleep(random.nextInt(1000)); // Eat for a while
                    }
                }
            }
        } catch(InterruptedException e) {}
    }
}

if(System.identityHashCode(left) < System.identityHashCode(right)) {
            first = left; second = right;
} else {
            first = right; second = left;
}

来自外星方法的危害

  private synchronized void updateProgress(int n) {
    for (ProgressListener listener: listeners) // listeners是累的一个field
      listener.onProgress(n);
  }


  private void updateProgress(int n) {
    ArrayList<ProgressListener> listenersCopy;
    synchronized(this) {
      listenersCopy = (ArrayList<ProgressListener>)listeners.clone();
    }
    for (ProgressListener listener: listenersCopy)
      listener.onProgress(n);
  }

避免危害的准则

2.3第二天:超越内置锁

Lock lock = new ReentrantLock();
lock.lock();
try{
  //使用共享资源

} finally { //使用finally确保锁释放
  lock.unlock();
}

可中断的锁

    final ReentrantLock l1 = new ReentrantLock();
    final ReentrantLock l2 = new ReentrantLock();


    Thread t1 = new Thread() {
      public void run() {
        try {
          l1.lockInterruptibly();
          Thread.sleep(1000);
          l2.lockInterruptibly();
        } catch (InterruptedException e) { System.out.println("t1 interrupted"); }
      }
    };

超时

class Philosopher extends Thread {
  private ReentrantLock leftChopstick, rightChopstick;
  private Random random;
  private int thinkCount;

  public Philosopher(ReentrantLock leftChopstick, ReentrantLock rightChopstick) {
    this.leftChopstick = leftChopstick; this.rightChopstick = rightChopstick;
    random = new Random();
  }

  public void run() {
    try {
      while(true) {
        ++thinkCount;
        if (thinkCount % 10 == 0)
          System.out.println("Philosopher " + this + " has thought " + thinkCount + " times");
        Thread.sleep(random.nextInt(1000)); // Think for a while
        leftChopstick.lock();
        try {
          if (rightChopstick.tryLock(1000, TimeUnit.MILLISECONDS)) {
            // Got the right chopstick
            try {
              Thread.sleep(random.nextInt(1000)); // Eat for a while
            } finally { rightChopstick.unlock(); }
          } else {
            // Didn't get the right chopstick - give up and go back to thinking
            System.out.println("Philosopher " + this + " timed out");
          }
        } finally { leftChopstick.unlock(); }
      }
    } catch(InterruptedException e) {}
  }
}

交替锁(hand-over-hand locking)

class ConcurrentSortedList {
  private class Node {
    int value;
    Node prev;
    Node next;
    ReentrantLock lock = new ReentrantLock();

    Node() {}

    Node(int value, Node prev, Node next) {
      this.value = value; this.prev = prev; this.next = next;
    }
  }

  private final Node head;
  private final Node tail;

  public ConcurrentSortedList() {
    head = new Node(); tail = new Node();
    head.next = tail; tail.prev = head;
  }

  //insert方法是有序的 遍历列表直到找到第一个值小于等于新插入的值得节点,在这个位置插入
  public void insert(int value) {
    Node current = head;
    current.lock.lock();
    Node next = current.next;
    try {
      while (true) {
        next.lock.lock();
        try {
          if (next == tail || next.value < value) {
            Node node = new Node(value, current, next);
            next.prev = node;
            current.next = node;
              //!!!这里return要在两个finally都执行完后才会执行啊!!!但只是finally里的.不过要是return换成exit(0)就直接退出了

            return;
          }
        } finally { current.lock.unlock(); }
        current = next;
        next = current.next;
      }
    } finally { next.lock.unlock(); }
  }

  public int size() {
    Node current = tail;
    int count = 0;

    while (current.prev != head) {
      ReentrantLock lock = current.lock;
      lock.lock();
      try {
        ++count;
        current = current.prev;
      } finally { lock.unlock(); }
    }

    return count;
  }

  public boolean isSorted() {
    Node current = head;
    while (current.next.next != tail) {
      current = current.next;
      if (current.value < current.next.value)
        return false;
    }
    return true;
  }
}

class LinkedList {
  public static void main(String[] args) throws InterruptedException {
    final ConcurrentSortedList list = new ConcurrentSortedList();
    final Random random = new Random();

    class TestThread extends Thread {
      public void run() {
        for (int i = 0; i < 10000; ++i)
          list.insert(random.nextInt());
      }
    }

    class CountingThread extends Thread {
      public void run() {
        while (!interrupted()) {
          System.out.print("\r" + list.size());
          System.out.flush();
        }
      }
    }

    Thread t1 = new TestThread();
    Thread t2 = new TestThread();
    Thread t3 = new CountingThread();
    //注意一下这里的用法 这里先join再interrupted的用法
    t1.start(); t2.start(); t3.start();
    t1.join(); t2.join();
    t3.interrupt();
 
    System.out.println("\r" + list.size());
 
    if (list.size() != 20000)
      System.out.println("*** Wrong size!");
 
    if (!list.isSorted())
      System.out.println("*** Not sorted!");
  }
}

条件变量

ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lock.lock();
try {
  while (! « condition is true » ) {
    condition.await();
  }
  //使用共享资源
} finally { lock.unlock(); }

class Philosopher extends Thread {

    private boolean eating;
    private Philosopher left;
    private Philosopher right;
    private ReentrantLock table;
    private Condition condition;
    private Random random;
    private int thinkCount;

    public Philosopher ( ReentrantLock table ) {
        eating = false;
        this.table = table;
        condition = table.newCondition();
        random = new Random();
    }

    public void setLeft ( Philosopher left ) {
        this.left = left;
    }

    public void setRight ( Philosopher right ) {
        this.right = right;
    }

    public void run () {
        try {
            while ( true ) {
                think();
                eat();
            }
        }
        catch ( InterruptedException e ) {
        }
    }

    private void think () throws InterruptedException {
        table.lock();
        try {
            eating = false;
            left.condition.signal();
            right.condition.signal();
        } finally {
            table.unlock();
        }
        ++thinkCount;
        if ( thinkCount % 10 == 0 ) {
            System.out.println( "Philosopher " + this + " has thought " + thinkCount + " times" );
        }
        Thread.sleep( 1000 );
    }

    private void eat () throws InterruptedException {
        table.lock();
        try {
            while ( left.eating || right.eating ) {
                condition.await();
            }
            eating = true;
        } finally {
            table.unlock();
        }
        Thread.sleep( 1000 );
    }
}

原子变量

2.4 站在巨人的肩膀上

写入时复制

  1. 内存占用问题
  2. 数据一致性问题:CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的的数据，马上能读到，请不要使用CopyOnWrite容器。
//Downloader.java
  private CopyOnWriteArrayList<ProgressListener> listeners;


  public void addListener(ProgressListener listener) {
    listeners.add(listener);
  }
  public void removeListener(ProgressListener listener) {
    listeners.remove(listener);
  }
  private void updateProgress(int n) {
    for (ProgressListener listener: listeners)
      listener.onProgress(n);
  }

一个完整的程序

版本一的并行:生产者和消费者(串行的我略过了)

//生产者 将page加到队尾
class Parser implements Runnable {
  private BlockingQueue<Page> queue;


  public Parser(BlockingQueue<Page> queue) {
    this.queue = queue;
  }
  
  public void run() {
    try {
      Iterable<Page> pages = new Pages(100000, "enwiki.xml");
      for (Page page: pages)
        queue.put(page);
    } catch (Exception e) { e.printStackTrace(); }
  }
}
//消费者
class Counter implements Runnable {
  private BlockingQueue<Page> queue;
  private Map<String, Integer> counts;

  public Counter(BlockingQueue<Page> queue,
                 Map<String, Integer> counts) {
    this.queue = queue;
    this.counts = counts;
  }


  public void run() {
    try {
      while(true) {
        Page page = queue.take();
        if (page.isPoisonPill())
          break;


        Iterable<String> words = new Words(page.getText());
        for (String word: words)
          countWord(word);
      }
    } catch (Exception e) { e.printStackTrace(); }
  }


  private void countWord(String word) {
    Integer currentCount = counts.get(word);
    if (currentCount == null)
      counts.put(word, 1);
    else
      counts.put(word, currentCount + 1);
  }
}



  public static void main(String[] args) throws Exception {
    ArrayBlockingQueue<Page> queue = new ArrayBlockingQueue<Page>(100);
    HashMap<String, Integer> counts = new HashMap<String, Integer>();
 
    Thread counter = new Thread(new Counter(queue, counts));
    Thread parser = new Thread(new Parser(queue));
    long start = System.currentTimeMillis();
 
    counter.start();
    parser.start();
    parser.join();
    queue.put(new PoisonPill());
    counter.join();
    long end = System.currentTimeMillis();
    System.out.println("Elapsed time: " + (end - start) + "ms");
  }

第二个版本:多个消费者

  private void countWord(String word) {
    lock.lock();
    try {
      Integer currentCount = counts.get(word);
      if (currentCount == null)  counts.put(word, 1);
      else  counts.put(word, currentCount + 1);
    } finally { lock.unlock(); }
  }

    ExecutorService executor = Executors.newCachedThreadPool();
    for (int i = 0; i < NUM_COUNTERS; ++i)
      executor.execute(new Counter(queue, counts));
    Thread parser = new Thread(new Parser(queue));
    parser.start();
    parser.join();
    for (int i = 0; i < NUM_COUNTERS; ++i)
      queue.put(new PoisonPill());
    executor.shutdown();

  //改用 ConcurrentHashMap
  private void countWord(String word) {
    while (true) { //理解一下这里的循环 如果下面的操作没有成功的话,就重试
      Integer currentCount = counts.get(word);
      if (currentCount == null) {
        if (counts.putIfAbsent(word, 1) == null) //如果没有与1关联 则关联,有原子性
          break;
      } else if (counts.replace(word, currentCount, currentCount + 1)) {
        break;
      }
    }

class Counter implements Runnable {

  private BlockingQueue<Page> queue;
  private ConcurrentMap<String, Integer> counts;
  private HashMap<String, Integer> localCounts;

  public Counter(BlockingQueue<Page> queue,
                 ConcurrentMap<String, Integer> counts) {
    this.queue = queue;
    this.counts = counts;
    localCounts = new HashMap<String, Integer>();
  }

  public void run() {
    try {
      while(true) {
        Page page = queue.take();
        if (page.isPoisonPill())
          break;
        Iterable<String> words = new Words(page.getText());
        for (String word: words)
          countWord(word);
      }
      //所以计数的那个可以是普通的map 他只在自己的线程里
      mergeCounts();
    } catch (Exception e) { e.printStackTrace(); }
  }

  private void countWord(String word) {
    Integer currentCount = localCounts.get(word);
    if (currentCount == null)
      localCounts.put(word, 1);
    else
      localCounts.put(word, currentCount + 1);
  }

  private void mergeCounts() {
    for (Map.Entry<String, Integer> e: localCounts.entrySet()) {
      String word = e.getKey();
      Integer count = e.getValue();
      while (true) {
        Integer currentCount = counts.get(word);
        if (currentCount == null) {
          if (counts.putIfAbsent(word, count) == null)
            break;
        } else if (counts.replace(word, currentCount, currentCount + count)) {
          break;
        }
      }
    }
  }
}

第三天总结

2.5 复习

上一篇 下一篇

猜你喜欢

热点阅读