《Java 7并发编程实战手册》第六章并发集合

由人民邮电出版社出版的《Java 7并发编程实战手册》终于出版了，译者是俞黎敏和申绍勇，该书将于近期上架。之前并发编程网组织翻译过此书，由于邮电出版社在并发网联系他们之前就找到了译者，所以没有采用并发网的译稿，但邮电出版社将于并发网展开合作，发布该书的样章（样章由并发网挑选，你也可以回帖告诉我们你想看哪一章的样章），并组织赠书活动回馈给活跃读者。活动详情请时刻关注并发网的微博和微信(微信号:ifeves)，最后祝各位用餐愉快！：）

本章将介绍下列内容：

  • 使用非阻塞式线程安全列表
  • 使用阻塞式线程安全列表
  • 使用按优先级排序的阻塞式线程安全列表
  • 使用带有延迟元素的线程安全列表
  • 使用线程安全可遍历映射
  • 生成并发随机数
  • 使用原子变量
  • 使用原子数组

6.1 简介

数据结构（Data Structure）是编程中的基本元素，几乎每个程序都使用一种或多种数据结构来存储和管理数据。Java API提供了包含接口、类和算法的Java集合框架（Java Collection Framework），它实现了可用在程序中的大量数据结构。

当需要在并发程序中使用数据集合时，必须要谨慎地选择相应的实现方式。大多数集合类不能直接用于并发应用，因为它们没有对本身数据的并发访问进行控制。如果一些并发任务共享了一个不适用于并发任务的数据结构，将会遇到数据不一致的错误，并将影响程序的准确运行。这类数据结构的一个例子是ArrayList类。

Java提供了一些可以用于并发程序中的数据集合，它们不会引起任何问题。一般来说，Java提供了两类适用于并发应用的集合。

  • 阻塞式集合（Blocking Collection）：这类集合包括添加和移除数据的方法。当集合已满或为空时，被调用的添加或者移除方法就不能立即被执行，那么调用这个方法的线程将被阻塞，一直到该方法可以被成功执行。
  • 非阻塞式集合（Non-Blocking Collection）：这类集合也包括添加和移除数据的方法。如果方法不能立即被执行，则返回null或抛出异常，但是调用这个方法的线程不会被阻塞。

通过本章的各个小节，你将学会如何在并发应用中使用一些Java集合。

  • 非阻塞式列表对应的实现类：ConcurrentLinkedDeque类；
  • 阻塞式列表对应的实现类：LinkedBlockingDeque 类；
  • 用于数据生成或消费的阻塞式列表对应的实现类：LinkedTransferQueue类；
  • 按优先级排序列表元素的阻塞式列表对应的实现类：PriorityBlockingQueue类；
  • 带有延迟列表元素的阻塞式列表对应的实现类：DelayQueue类；
  • 非阻塞式可遍历映射对应的实现类：ConcurrentSkipListMap类；
  • 随机数字对应的实现类：ThreadLocalRandom类；
  • 原子变量对应的实现类：AtomicLong和AtomicIntegerArray类。

6.2 使用非阻塞式线程安全列表

最基本的集合类型是列表（List）。一个列表包含的元素数量不定，可以在任何位置添加、读取或移除元素。并发列表允许不同的线程在同一时间添加或移除列表中的元素，而不会造成数据不一致。
在本节，将会学到如何在并发程序中使用非阻塞式列表。非阻塞式列表提供了一些操作，如果被执行的操作不能够立即运行（例如，在列表为空时，从列表取出一个元素），方法会抛出异常或返回null。Java 7引入了ConcurrentLinkedDeque类来实现非阻塞式并发列表。
将要实现的范例包括以下两个不同的任务：

  • 添加大量的数据到一个列表中；
  • 从同一个列表中移除大量的数据。

准备工作

本节的范例是在Eclipse IDE里完成的。无论你使用Eclipse还是其他的IDE（比如NetBeans），都可以打开这个IDE并且创建一个新的Java工程。

范例实现

按照接下来的步骤实现本节的范例。
1．创建一个名为AddTask的类，实现Runnable接口。

1 public class AddTask implements Runnable {

2．声明一个私有的ConcurrentLinkedDeque属性list，并指定它的泛型参数是String型的。

1 private ConcurrentLinkedDeque list;

3．实现类的构造器来初始化属性。

1 public AddTask(ConcurrentLinkedDeque list) {
2 this.list=list;
3 }

4．实现run()方法。这个方法将10,000个字符串存放到列表中，这些字符串由当前执行任务的线程的名称和数字组成。

1 @Override
2 public void run() {
3 String name=Thread.currentThread().getName();
4 for (int i=0; i<10000; i++){
5 list.add(name+": Element "+i);
6 }
7 }

5．创建名为PollTask的类，并实现Runnable接口。

1 public class PollTask implements Runnable {

6．声明一个私有的ConcurrentLinkedDeque属性list，并指定它的泛型参数是String型的。

1 private ConcurrentLinkedDeque list;

7．实现类的构造器来初始化属性。

1 public PollTask(ConcurrentLinkedDeque list) {
2 this.list=list;
3 }

8．实现run()方法。这个方法将列表中的10,000个字符串取出，总共取5,000次，每次取两个元素。

1 @Override
2 public void run() {
3 for (int i=0; i<5000; i++) {
4 list.pollFirst();
5 list.pollLast();
6 }
7 }

9．创建范例的主类Main，并添加main()方法。

1 public class Main {
2 public static void main(String[] args) {

10．创建ConcurrentLinkedDeque对象，并指定它的泛型参数是String型的。

1 ConcurrentLinkedDeque list=new
2 ConcurrentLinkedDeque<>();

11．创建线程数组threads，它包含100个线程。

1 Thread threads[]=new Thread[100];

12．创建100个AddTask对象及其对应的运行线程。将每个线程存放到上一步创建的数组中，然后启动线程。

1 for (int i=0; i AddTask task=new AddTask(list);
2 threads[i]=new Thread(task);
3 threads[i].start();
4 }
5 System.out.printf("Main: %d AddTask threads have been
6 launched\n",threads.length);

13．使用join()方法等待线程完成。

1 for (int i=0; i<threads.length; i++) {
2 try {
3 threads[i].join();
4 } catch (InterruptedException e) {
5 e.printStackTrace();
6 }
7 }

14．将列表的元素数量打印到控制台。

1 System.out.printf("Main: Size of the List: %d\n",list.size());

15．创建100个PollTask对象及其对应的运行线程。将每个线程存放到上一步创建的数组中，然后启动线程。

1 for (int i=0; i< threads.length; i++){
2 PollTask task=new PollTask(list);
3 threads[i]=new Thread(task);
4 threads[i].start();
5 }
6 System.out.printf("Main: %d PollTask threads have been
7 launched\n",threads.length);

16．使用join()方法等待线程完成。

1 for (int i=0; i<threads.length; i++) {
2 try {
3 threads[i].join();
4 } catch (InterruptedException e) {
5 e.printStackTrace();
6 }
7 }

17．将列表的元素数量打印到控制台。

1 System.out.printf（"Main: Size of the List: %d\n",list.size()）;

工作原理

本节使用的泛型参数是String类的ConcurrentLinkedDeque对象，用来实现一个非阻塞式并发数据列表。下面的截屏显示了程序的运行结果。

首先，执行100个AddTask任务将元素添加到ConcurrentLinkedDeque对象list中。每个任务使用add()方法向这个列表中插入10,000个元素。add()方法将新元素添加到列表尾部。当所有任务运行完毕，列表中的元素数量将被打印到控制台。在这一刻，列表中有1,000,000个元素。
接下来，执行100个PollTask任务将元素从列表中移除。每个任务使用pollFirst()和pollLast()方法从列表中移除10,000个元素。pollFirst()方法返回并移除列表中的第一个元素，pollLast()方法返回并移除列表中的最后一个元素。如果列表为空，这些方法返回null。当所有任务运行完毕，列表中的元素数量将被打印到控制台。在这一刻，列表中有0个元素。

使用size()方法输出列表中的元素数量。需要注意的是，这个方法返回的值可能不是真实的，尤其当有线程在添加数据或移除数据时，这个方法需要遍历整个列表来计算元素数量，而遍历过的数据可能已经改变。仅当没有任何线程修改列表时，才能保证返回的结果是准确的。

更多信息

ConcurrentLinkedDeque类提供了其他从列表中读取数据的方法。

  • getFirst()和getLast()：分别返回列表中第一个和最后一个元素，返回的元素不会从列表中移除。如果列表为空，这两个方法抛出NoSuchElementExcpetion异常。
  • peek()、peekFirst()和peekLast()：分别返回列表中第一个和最后一个元素，返回的元素不会从列表中移除。如果列表为空，这些方法返回null。
  • remove()、removeFirst()和removeLast()：分别返回列表中第一个和最后一个元素，返回的元素将会从列表中移除。如果列表为空，这些方法抛出NoSuchElementExcpetion异常。

6.3　使用阻塞式线程安全列表

最基本的集合类型是列表。一个列表包含的元素数量不定，可以在任何位置添加、读取或移除元素。并发列表允许不同的线程在同一时间添加或移除列表中的元素，而不会造成数据不一致。
在本节，你会学到如何在并发程序中使用阻塞式列表。阻塞式列表与非阻塞式列表的主要差别是：阻塞式列表在插入和删除操作时，如果列表已满或为空，操作不会被立即执行，而是将调用这个操作的线程阻塞队列直到操作可以执行成功。Java引入了LinkedBlocking Deque类来实现阻塞式列表。
将要实现的范例包括以下两个不同的任务：

  • 添加大量的数据到一个列表中；
  • 从同一个列表中移除大量的数据。

准备工作

本节的范例是在Eclipse IDE里完成的。无论你使用Eclipse还是其他的IDE（比如NetBeans），都可以打开这个IDE并且创建一个新的Java工程。

范例实现

按照接下来的步骤实现本节的范例。
1．创建名为Client的类，并实现Runnable接口。

1 public class Client implements Runnable{

2．声明一个私有的LinkedBlockingDeque属性requestList，并指定它的泛型参数是String型的。

1 private LinkedBlockingDeque requestList;

3．实现类的构造器来初始化属性。

1 public Client (LinkedBlockingDeque requestList) {
2 this.requestList=requestList;
3 }

4．实现run()方法。使用requestList对象的put()方法，每两秒向列表requestList中插入5个字符串。重复3次。

01 @Override
02 public void run() {
03 for (int i=0; i<3; i++) {
04 for (int j=0; j<5; j++) {
05 StringBuilder request=new StringBuilder();
06 request.append(i);
07 request.append(":");
08 request.append(j);
09 try {
10 requestList.put(request.toString());
11 } catch (InterruptedException e) {
12 e.printStackTrace();
13 }
14 System.out.printf("Client: %s at %s.\n",request,new
15 Date());
16 }
17 try {
18 TimeUnit.SECONDS.sleep(2);
19 } catch (InterruptedException e) {
20 e.printStackTrace();
21 }
22 }
23 System.out.printf("Client: End.\n");
24 }

5． 创建范例的主类Main，并添加main()方法。

1 public class Main {
2 public static void main(String[] args) throws Exception {

6． 声明并创建LinkedBlockingDeque属性list，并指定它的泛型参数是String型的。

1 LinkedBlockingDeque list=new LinkedBlockingDeque<>(3);

7．将client作为传入参数创建线程Thread并启动。

1 Client client=new Client(list);
2 Thread thread=new Thread(client);
3 thread.start();

8．使用list对象的take()方法，每300毫秒从列表中取出3个字符串对象，重复5次。在控制台输出字符串。

1 for (int i=0; i for (int j=0; j<3; j++) {
2 String request=list.take();
3 System.out.printf("Main: Request: %s at %s. Size:
4 %d\n",request,new Date(),list.size());
5 }
6 TimeUnit.MILLISECONDS.sleep(300);
7 }

9．输出一条表示程序结束的消息。

1 System.out.printf("Main: End of the program.\n");

工作原理

本节使用的泛型参数是String的LinkedBlockingDeque对象，用来实现一个阻塞式并发数据列表。
Client类使用put()方法将字符串插入到列表中。如果列表已满（列表生成时指定了固定的容量），调用这个方法的线程将被阻塞直到列表中有了可用的空间。

Main类使用take()方法从列表中取字符串。如果列表为空，调用这个方法的线程将被阻塞直到列表不为空（即有可用的元素）。

这个例子中使用了LinkedBlockingDeque对象的两个方法，调用它们的线程可能会被阻塞，在阻塞时如果线程被中断，方法会抛出InterruptedException异常，所以必须捕获和处理这个异常。

更多信息

LinkedBlockingDeque类也提供了其他存取元素的方法，这些方法不会引起阻塞，而是抛出异常或返回null。

  • takeFirst()和takeLast()：分别返回列表中第一个和最后一个元素，返回的元素会从列表中移除。如果列表为空，调用方法的线程将被阻塞直到列表中有可用的元素出现。
  • getFirst()和getLast()：分别返回列表中第一个和最后一个元素，返回的元素不会从列表中移除。如果列表为空，则抛出NoSuchElementExcpetinon异常。
  • peek()、peekFirst()和peekLast()：分别返回列表中第一个和最后一个元素，返回的元素不会从列表中移除。如果列表为空，返回null。
  • poll()、pollFirst()和pollLast()：分别返回列表中第一个和最后一个元素，返回的元素将会从列表中移除。如果列表为空，返回null。
  • add()、addFirst()和addLast(): 分别将元素添加到列表中第一位和最后一位。如果列表已满（列表生成时指定了固定的容量），这些方法将抛出IllegalStateException异常。
  • 文章转自 并发编程网-ifeve.com
时间： 2024-04-20 01:26:31

《Java 7并发编程实战手册》第六章并发集合的相关文章

《C++并发编程实战》——第1章 你好，C++并发世界

第1章 你好,C++并发世界 C++并发编程实战本章主要内容 何谓并发和多线程为什么要在应用程序中使用并发和多线程C++并发支持的发展历程一个简单的C++多线程程序是什么样的这是令C++用户振奋的时刻.距1998年初始的C++标准发布13年后,C++标准委员会给予程序语言和它的支持库一次重大的变革.新的C++标准(也被称为C++11或C++0x)于2011年发布并带来了很多的改变,使得C++的应用更加容易并富有成效. 在C++11标准中一个最重要的新特性就是支持多线程程序.这是C++标准第一次在

《Java 7并发编程实战手册》第四章线程执行器

感谢人民邮电大学授权并发网发布此书样章,新书购买传送门=>当当网 本章将介绍下列内容: 创建线程执行器 创建固定大小的线程执行器 在执行器中执行任务并返回结果 运行多个任务并处理第一个结果 运行多个任务并处理所有结果 在执行器中延时执行任务 在执行器中周期性执行任务 在执行器中取消任务 在执行器中控制任务的完成 在执行器中分离任务的启动与结果的处理 处理在执行器中被拒绝的任务 4.1 简介 通常,使用Java来开发一个简单的并发应用程序时,会创建一些 Runnable 对象,然后创建对应的 Th

《Java 7并发编程实战手册》第五章Fork/Join框架

感谢人民邮电大学授权并发网发布此书样章,新书已上市,购买请进当当网 本章内容包含: 创建Fork/Join线程池 合并任务的结果 异步运行任务 在任务中抛出异常 取消任务 5.1 简介 通常,使用Java来开发一个简单的并发应用程序时,会创建一些Runnable对象,然后创建对应的Thread 对象来控制程序中这些线程的创建.执行以及线程的状态.自从Java 5开始引入了Executor和ExecutorService接口以及实现这两个接口的类(比如ThreadPoolExecutor)之后,使

《GO并发编程实战》—— 锁的使用

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 在本节,我们对Go语言所提供的与锁有关的API进行说明.这包括了互斥锁和读写锁.我们在第6章描述过互斥锁,但却没有提到过读写锁.这两种锁对于传统的并发程序来说都是非常常用和重要的. 互斥锁 互斥锁是传统的并发程序对共享资源进行访问控制的主要手段.它由标准库代码包sync中的Mutex结构体类型代表.sync.Mutex类型(确切地说,是*sync.Mutex类型)只有两个公开方法--Lock和U

《GO并发编程实战》—— 原子操作

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 我们已经知道,原子操作即是进行过程中不能被中断的操作.也就是说,针对某个值的原子操作在被进行的过程当中,CPU绝不会再去进行其它的针对该值的操作.无论这些其它的操作是否为原子操作都会是这样.为了实现这样的严谨性,原子操作仅会由一个独立的CPU指令代表和完成.只有这样才能够在并发环境下保证原子操作的绝对安全. Go语言提供的原子操作都是非侵入式的.它们由标准库代码包sync/atomic中的众多函

《GO并发编程实战》—— 条件变量

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 我们在第6章讲多线程编程的时候详细说明过条件变量的概念.原理和适用场景.因此,我们在本小节仅对sync代码包中与条件变量相关的API进行简单的介绍,并使用它们来改造我们之前实现的*myDataFile类型的相关方法. 在Go语言中,sync.Cond类型代表了条件变量.与互斥锁和读写锁不同,简单的声明无法创建出一个可用的条件变量.为了得到这样一个条件变量,我们需要用到sync.NewCond函数

《GO并发编程实战》—— Concurrent Map

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 我们在本章前面的部分中对Go语言提供的各种传统同步工具和方法进行了逐一的介绍.在本节,我们将运用它们来构造一个并发安全的字典(Map)类型. 我们已经知道,Go语言提供的字典类型并不是并发安全的.因此,我们需要使用一些同步方法对它进行扩展.这看起来并不困难.我们只要使用读写锁将针对一个字典类型值的读操作和写操作保护起来就可以了.确实,读写锁应该是我们首先想到的同步工具.不过,我们还不能确定只使用

《GO并发编程实战》—— 只会执行一次

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 现在,让我们再次聚焦到sync代码包.除了我们介绍过的互斥锁.读写锁和条件变量,该代码包还为我们提供了几个非常有用的API.其中一个比较有特色的就是结构体类型sync.Once和它的Do方法. 与代表锁的结构体类型sync.Mutex和sync.RWMutex一样,sync.Once也是开箱即用的.换句话说,我们仅需对它进行简单的声明即可使用,就像这样: var once sync.Once o

《GO并发编程实战》—— WaitGroup

声明:本文是<Go并发编程实战>的样章,感谢图灵授权并发编程网站发布样章,禁止以任何形式转载此文. 我们在第6章多次提到过sync.WaitGroup类型和它的方法.sync.WaitGroup类型的值也是开箱即用的.例如,在声明 var wg sync.WaitGroup 之后,我们就可以直接正常使用wg变量了.该类型有三个指针方法,即Add.Done和Wait. 类型sync.WaitGroup是一个结构体类型.在它之中有一个代表计数的字段.当一个sync.WaitGroup类型的变量被声