A concurrent program has multiple logical threads of control. These threads may or may not run in parallel.
A parallel program may or may not have more than one logical thread of control.
并发是问题域中的概念:程序需要设计成能够处理多个同时(几乎同时)发生的事件;而并行则是方法域中的概念:通过将问题中的多个部分并发执行,来加速解决问题.
并发是同一时间 应对(dealing with) 多件事件的处理能力;并行是同一时间 动手做(doing) 多件事情的能力
可以同时处理多个问题,但是每次只做一件事,是并发.
我妻子是一位教师。与众多教师一样,她极其善于处理多个任务。她虽然每次只能做一件事,但可以并发地处理多个任务。比如,在听一位学生朗读的时候,她可以暂停学生的朗读,以维持课堂秩序,或者回答学生的问题。这是并发,但并不并行(因为仅有她一个人,某一时刻只能进行一件事)。 但如果还有一位助教,则她们中一位可以聆听朗读,而同时另一位可以回答问题。这种方式既是并发,也是并行。 假设班级设计了自己的贺卡并要批量制作。一种方法是让每位学生制作五枚贺卡。这种方法是并行,而(从整体看)不是并发,因为这个过程整体来说只有一个任务。
并发和并行经常被混淆的一个原因是,传统的“线程与锁”模型并没有显式支持并行。如果要用线程与锁模型为多核进行开发,唯一的选择就是写一个并发的程序,让其并行地运行在多核上。
并发程序通常是不确定的, 并行程序可能是确定的. 使用一门直接支持并行的编程语言可以写出并行程序,而不会引入不入不 确定性.
终于来到了大家所默认的并行形式——多处理器。从程序员的角度来看,多处理器架构最明 显的分类特征是其内存模型(共享内存模型或分布式内存模型)。
共享内存模型(通过内存通信)
分布式内存模型( 通过网络通信 )
`java class Counter { private int count = 0; public synchronized void increment() { ++count; } ➤ public int getCount() { return count; } }
` - 这段代码没有竞态条件的bug 但是又内存可见性的bug 因为getCount()没有加锁,调用getCount()可能获得一个失效的值
```java 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) {} }
} ``` - 创建五个哲学家实例,这个程序可以愉快的运行很久,但到某个时刻一切会停下来:如果所有哲学家同时进餐,就会死锁(相邻的几个同时准备进餐还不至于会死锁) - 一个线程想使用多把锁的时候就要考虑死锁的可能,有一个简单的规则还有避免死锁:总是按照一个全局的固定的顺序获取多把锁,其中一种实现如下:
```java 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) {} }
} ```
程序解释:当所有人同时决定进餐的时候,ABCD左手分别拿起1234号筷子(对于他们小的编号的筷子还是在左手),这和上面的程序没啥不同,但是差别就在这个E,他左边的筷子是大编号,所以他左手拿的是1,然而1被A拿了,所以他就一只筷子都拿不到,所以D可以正常进餐,就不会死锁
局限:获取锁的代码写的比较集中的话,有利于维护这个全局顺序,但是对于规模比较大的程序,使用锁的地方比较零散,各处都遵守这个顺序就显得不太实际.
技巧:使用对象的散列值作为锁的全局顺序
优点:适用于所有java对象,不用为锁对象专门定义并维护一个顺序,
缺点:但是对象的散列值不能保证唯一性(虽然几率很小), 不是迫不得已不要使用
`java if(System.identityHashCode(left) < System.identityHashCode(right)) { first = left; second = right; } else { first = right; second = left; }
`
`java private synchronized void updateProgress(int n) { for (ProgressListener listener: listeners) // listeners是累的一个field listener.onProgress(n); }
`
上面的方法乍一看好像没啥问题,但是这个方法调用了onProgress()方法,我们对这个方法一无所知, 要是他里面还有一把锁,就可能会死锁
解决方案:在遍历前对listeners进行保护性复制(defensive copy),再针对这份副本进行遍历 ```java
private void updateProgress(int n) { ArrayList
这个方案一石多鸟,不仅调用外星方法的时候不用加锁,而且还大大减少了代码持有锁的时间(前面是对方法加synchronized,这里是对语句块)
```java Lock lock = new ReentrantLock(); lock.lock(); try{ //使用共享资源
} finally { //使用finally确保锁释放 lock.unlock(); } ```
```java 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"); } } };
```
```java 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) {} } } ```
交替锁可以只锁住链表的一部分,允许不涉及被锁部分的其他线程自由访问链表.插入新的链表节点时,需要将待插入位置两边的节点加锁.首先锁住链表的前两个节点,如果这两个节点之间不是待插入的位置,那么就解锁第一个,并锁住第三个,以此类推,知道找到待插入位置并插入新的节点,最后解锁两边的节点
这种交替型的加锁和解锁顺序无法用内置锁实现,使用ReentrantLock可以
```java 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!");
} } ```
`java ReentrantLock lock = new ReentrantLock(); Condition condition = lock.newCondition(); lock.lock(); try { while (! « condition is true » ) { condition.await(); } //使用共享资源 } finally { lock.unlock(); }
`
```java 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 ); }
} ```
```java //Downloader.java private CopyOnWriteArrayList
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); } ```
```java //生产者 将page加到队尾 class Parser implements Runnable { private BlockingQueue
public Parser(BlockingQueue
public void run() { try { Iterable
public Counter(BlockingQueue
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); } }
```
```java public static void main(String[] args) throws Exception { ArrayBlockingQueue
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");
} ```
`java 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(); } }
`
`java 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();
`
`java //改用 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; } }
`
```java class Counter implements Runnable {
private BlockingQueue
public Counter(BlockingQueue
public void run() { try { while(true) { Page page = queue.take(); if (page.isPoisonPill()) break; Iterable
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