17.1.1 网络基础知识

所谓计算机网络,就是把分布在不同地理区域的计算机与专门的外部设备用通信线路互连成一个规模大、功能强的网络系统,从而使众多的计算机可以方便地互相传递信息,共享硬件、软件、数据信息等资源
计算机网络是现代通信技术与计算机技术相结合的产物,计算机网络可以提供以下一些主要功能。

• 资源共享。
• 信息传输与集中处理。
• 均衡负荷与分布处理
• 综合信息服务。

通过计算机网络可以向全社会提供各种经济信息、科研情报和咨询服务。其中,国际互联网Internet上的全球信息网(WWW,World Wide Web)服务就是一个最典型也是最成功的例子。实际上,今天的网络承载绝大部分大型企业的运转,一个大型的、全球性的企业或组织的日常工作流程都是建立在互联网基础之上的。

计算机网络分类

计算机网络的品种很多,根据各种不同的分类原则,可以得到各种不同类型的计算机网络。

计算机网络通常是按照规模大小和延伸范围来分类的,常见的划分为:局域网(LAN)、城域网(MAN)、厂域网(WAN)。Internet可以视为世界上最大的广域网。
如果按照网络的拓扑结构来划分,可以分为星型网络、总线型网络、环型网络、树型网络、星型环型网络等;

通信协议

如果按照网络的传输介质来划分,可以分为双绞线网、同轴电缆网、光纤网和卫星网等。
计算机网络中实现通信必须有一些约定,这些约定被称为通信协议。通信协议负责对传输速率、传输代码、代码结构、传输控制步骤、岀错控制等制定处理标准。为了让两个节点之间能进行对话,必须在它们之间建立通信工具,使彼此之间能进行信息交换。

通信协议组成

通信协议通常由三部分组成:

• 一是语义部分,用于决定双方对话的类型;
• 二是语法部分,用于决定双方对话的格式;
• 三是变换规则,用于决定通信双方的应答关系。

OSI

国际标准化组织ISO于1978年提出“开放系统互连参考模型”,即著名的OSI(Open System Interconnection)。

开放系统互连参考模型力求将网络简化,并以模块化的方式来设计网络。
开放系统互连参考模型把计算机网络分成

• 物理层、
• 数据链路层、
• 网络层、
• 传输层、
• 会话层、
• 表示层、
• 应用层

这七层

通过十多年的发展和推进,OSI模式已成为各种计算机网络结构的参考标准
图17.1显示了OSI参考模型的推荐分层。

IP

前面介绍过通信协议是网络通信的基础,IP协议则是一种非常重要的通信协议。IP(Internet Protocol)协议又称互联网协议,是支持网间互联的数据报协议。它提供网间连接的完善功能,包括IP数据报规定互联网络范围内的地址格式

TCP

经常与IP协议放在一起的还有TCP(Transmission Control protocol)协议,即传输控制协议,它规定一种可靠的数据信息传递服务。虽然IP和TCP这两个协议功能不尽相同,也可以分开单独使用,但它们是在同一个时期作为一个协议来设计的,并且在功能上也是互补的。因此实际使用中常常把这两个协议统称为TCP/IP协议,TCP/IP协议最早出现在UNIX操作系统中,现在几乎所有的操作系统都支持TCPP协议,因此TCP/IP协议也是Internet中最常用的基础协议。

TCP/IP和OSI分层的关系

按TCP/IP协议模型,网络通常被分为一个四层模型,这个四层模型和前面的OSI七层模型有大致的对应关系,图17.2显示了TCPP分层模型和OSI分层模型之间的对应关系。

17.1.2 IP地址和端口号

IP地址用于唯一地标识网络中的一个通信实体,这个通信实体既可以是一台主机,也可以是一台打印机,或者是路由器的某一个端口。而在基于IP协议网络中传输的数据包,都必须使用P地址来进行标识。
就像写一封信,要标明收信人的通信地址和发信人的地址,而邮政工作人员则通过该地址来决定邮件的去向。类似的过程也发生在计算机网络里,每个被传输的数据包也要包括一个源IP地址和一个目的IP地址,当该数据包在网络中进行传输时,这两个地址要保持不变,以确保网络设备总能根据确定的P地址,将数据包从源通信实体送往指定的目的通信实体。

IP地址是数字型的,IP地址是一个32位(32 bit)整数,但通常为了便于记忆,通常把它分成4个8位的二进制数,每8位之间用圆点隔开,每个8位整数可以转换成一个0~255的十进制整数,因此日常看到的IP地址常常是这种形式:202.9.128.88。

NIC(Internet Network Information Center)统一负责全球Internet IP地址的规划、管理,而Inter NIC、 APNIC、RIPE三大网络信息中心具体负责美国及其他地区的IP地址分配。其中APNIC负责亚太地区的IP管理,我国申请IP地址也要通过APNIC,APNIC的总部设在日本东京大学。

五类IP地址

IP地址被分成了A、B、C、D、E五类,每个类别的网络标识和主机标识各有规则。

• A类:10.0.0.0~10.255.255.255
• B类:172.16.0.0~172.31.255.255
• C类:192.168.0.0~192.168.255.255

端口

IP地址用于唯一地标识网络上的一个通信实体,但一个通信实体可以有多个通信程序同时提供网络服务,此时还需要使用端口。
端口是一个16位的整数,用于表示数据交给哪个通信程序处理。因此,端口就是应用程序与外界交流的出入口,它是一种抽象的软件结构,包括一些数据结构和IO(基本输入输出)缓冲区。
不同的应用程序处理不同端口上的数据,同一台机器上不能有两个程序使用同一个端口,端口号可以从0到65535。

端口分类

通常将端口分为如下三类:

• 公认端口(Well Known Ports):从0到1023,它们紧密绑定(Binding)一些特定的服务
• 注册端口(Registered Ports):从1024到49151,它们松散地绑定一些服务。应用程序通常应该使用这个范围内的端口.
• 动态和或私有端口(Dynamic And/or Private Ports):从49152到65535,这些端口是应用程序使用的动态端口,应用程序一般不会主动使用这些端口

如果把IP地址理解为某个人所在地方的地址(包括街道和门牌号),但仅有地址还是找不到这个人,还需要知道他所在的房号才可以找到这个人。因此如果把应用程序当作人,把计算机网络当作类似邮递员的角色,当一个程序需要发送数据时,需要指定目的地的IP地址和端口,如果指定了正确的IP地址和端口号,计算机网络就可以将数据送给该IP地址和端口所对应的程序。

第17章 网络编程 前言

本章要点

• 计算机网络基础
• IP地址和端口
• 使用InetAddress包装IP地址
• 使用URLEncoder和URLDecoder工具类
• 使用URLConnection访问远程资源
• TCP协议基础
• 使用ServerSocket和Socket
• 使用NIO实现非阻塞式网络通信
• 使用AIO实现异步网络通信
• UDP协议基础
• 使用DatagramSocket发送/接收数据报(DatagramPacket)
• 使用MulticastSocket实现多点广播
• 通过Proxy使用代理服务器
• 通过ProxySelector使用代理服务器

本章将主要介绍Java网络通信的支持,通过这些网络支持类,Java程序可以非常方便地访问互联网上的HTTP服务、FTP服务等,并可以直接取得互联网上的远程资源,还可以向远程资源发送GET、POST请求

介绍网络工具类

本章先简要介绍计算机网络的基础知识,包括酽地址和端口等概念,这些知识是网络编程的基础。本章会详细介绍InetAddress、URLDecoder、URLEncoder、URL和URLConnection等网络工具类,并会深入介绍通过URLConnection发送请求、访问远程资源等操作。

TCP编程

本章将重点介绍Java提供的TCP网络通信支持,包括如何利用ServerSocket建立TCP服务器,利用Socket建立TCP客户端。实际上Java的网络通信非常简单,服务器端通过ServerSocket建立监听,客户端通过Socket连接到指定服务器后,通信双方就可以通过IO流进行通信。本章将以采用逐步迭代的方式开发一个C/S结构多人网络聊天工具为例,向读者介绍基于TCP协议的网络编程。

UDP编程

本章还将重点介绍Java提供的UDP网络通信支持,主要介绍如何使用DatagramSocket来发送、接收数据报(DatagramPacket),并讲解如何使用MulticastSocket来实现多点广播通信。本章也将以开发局域网通信程序为例来介绍MulticastSocket和DatagramSocket的实际用法

代理服务器访问远程资源

本章最后还会介绍利用Proxy和ProxySelector在Java程序中通过代理服务器访问远程资源。

16.10 本章小结

本章主要介绍了Java的多线程编程支持:

• 简要介绍了线程的基本概念,并讲解了线程和进程之间的区别与联系。
• 本章详细讲解了如何创建、启动多线程,并对比了三种创建多线程方式之间的优势和劣势,也详细介绍了线程的生命周期。
• 本章通过示例程序示范了控制线程的几个方法,
• 还详细讲解了线程同步的意义和必要性,并介绍了三种不同的线程同步方法:
  • 同步方法、
  • 同步代码块，
  • 显式使用Lock控制线程同步。
• 本章也介绍了三种实现线程通信的方式:
  • 使用同步监视器的方法实现通信、
  • 显式使用Condition对象实现线程通信，
  • 使用阻塞对象实现线程通信。

此外,本章还介绍了线程组和线程池,由于线程属于创建成本较大的对象,因此程序应该考虑复用线程,线程池是在实际开发中不错的选择.
本章最后介绍了线程相关的工具类,比如ThreadLocal，线程安全的集合类，以及如果使用Collections包装线程不安全的集合类.

本章练习

1. 写2个线程,其中一个线程打印1~52,另一个线程打印A~Z,每两个数字后打印一个字母,也就是打印顺序应该是12A34B56C...5152Z。该习题需要利用多线程通信的知识。
2. 假设车库有3个车位(可以用boolean口数组来表示车库)可以停车,写一个程序模拟多个用户开车离开、停车入库的效果。注意:车位有车时不能停车

16.9.4 Java9新增的发布-订阅框架

Java9新增了一个发布-订阅框架,该框架是基于异步响应流的。这个发布-订阅框架可以非常方便地处理异步线程之间的流数据交换(比如两个线程之间需要交换数据)。而且这个发布-订阅框架不需要使用数据中心来缓冲数据,同时具有非常高效的性能。

Flow类静态内部接口

这个发布订阅框架使用Flow类的4个静态内部接口作为核心API

Flow.Publisher接口方法

Flow.Publisher作为生产者,负责发布数据项,并注册订阅者。Flow.Publisher接口定义了如下方法来注册订阅者。

Flow.Subscriber接口方法

Flow.Subscriber接口定义了如下方法

Flow.Publisher的SubmissionPublisher实现类

为了处理一些通用发布者的场景,Java9为Flow.Publisher提供了一个SubmissionPublisher实现类,它可向当前订阅者异步提交非空的数据项,直到它被关闭。每个订阅者都能以相同的顺序接收到新提交的数据项。
程序创建SubmissionPublisher对象时,需要传入一个线程池作为底层支撑;该类也提供了一个无参数的构造器,该构造器使用ForkJoinPool.commonPool()方法来提交发布者,以此实现发布者向订阅者提供数据项的异步特性。

程序示例 使用SubmissionPublisher作为发布者

下面程序示范了使用SubmissionPublisher作为发布者的用法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

import java.util.concurrent.Flow.*;
import java.util.*;
import java.util.concurrent.*;

public class PubSubTest {
    public static void main(String[] args) {
        // 创建一个SubmissionPublisher作为发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        // 创建订阅者
        MySubscriber<String> subscriber = new MySubscriber<>();
        // 注册订阅者
        publisher.subscribe(subscriber);
        // 发布几个数据项
        System.out.println("开发发布数据...");
        List.of("Java", "Kotlin", "Go", "Erlang", "Swift", "Lua").forEach(im -> {
            // 提交数据
            publisher.submit(im);
            try {
                Thread.sleep(500);
            } catch (Exception ex) {
            }
        });
        // 发布结束
        publisher.close();
        // 发布结束后，为了让发布者线程不会死亡，暂停线程
        synchronized ("fkjava") {
            try {
                "fkjava".wait();
            } catch (Exception ex) {
            }
        }
    }
}

// 创建订阅者
class MySubscriber<T> implements Subscriber<T> {
    // 发布者与订阅者之间的纽带
    private Subscription subscription;

    @Override // 订阅时触发该方法
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        // 开始请求数据
        subscription.request(1);
    }

    @Override // 接收到数据时触发该方法
    public void onNext(T item) {
        System.out.println("获取到数据: " + item);
        // 请求下一条数据
        subscription.request(1);
    }

    @Override // 订阅出错时触发该方法
    public void onError(Throwable t) {
        t.printStackTrace();
        synchronized ("fkjava") {
            "fkjava".notifyAll();
        }
    }

    @Override // 订阅结束时触发该方法
    public void onComplete() {
        System.out.println("订阅结束");
        synchronized ("fkjava") {
            "fkjava".notifyAll();
        }
    }
}

上面程序中先创建SubmissionPublisher对象,该对象可作为发布者;然后创建订阅者对象,该订阅者类是一个自定义类;接着注册订阅者。
完成上面步骤之后,程序即可调用SubmissionPublisher对象的submit()方法来发布数据项,发布者通过该方法发布数据项。
上面程序实现了一个自定义的订阅者,该订阅者实现了Subscriber接口的4个方法,重点就是实现onNext()方法,当订阅者获取到数据时就会触发这个onNext()方法,订阅者通过该方法接收数据。
至于订阅者接收到数据项之后的处理,则取决于程序的业务需求
运行该程序,可以看到订阅者逐项获得数据的过程。

1
2
3
4
5
6
7
8

开发发布数据...
获取到数据: Java
获取到数据: Kotlin
获取到数据: Go
获取到数据: Erlang
获取到数据: Swift
获取到数据: Lua
订阅结束

16.9.3 线程安全的集合类

实际上从Java5开始,在java.util.concurrent包下提供了大量支持高效并发访问的集合接口和实现类,如图16.7所示。

线程安全集合分类

从图16.17所示的类图可以看出,这些线程安全的集合类可分为如下两类

• 以Concurrent开头的集合类,如ConcurrentHashMap、ConcurrentSkipListMap、ConcurrentSkipListSet、ConcurrentLinkedQueue,ConcurrentLinkedDeque.
• 以CopyOnWrite开头的集合类,如CopyOnWriteArrayList、CopyOnWriteArraySet

Concurrent开头的集合类

其中以Concurrent开头的集合类代表了支持并发访问的集合,它们可以支持多个线程并发写入访问,这些写入线程的所有操作都是线程安全的,但读取操作不必锁定。以Concurrent开头的集合类采用了更复杂的算法来保证永远不会锁住整个集合,因此在并发写入时有较好的性能。

ConcurrentLinkedQueue

当多个线程共享访问一个公共集合时,ConcurrentLinkedQueue是一个恰当的选择。ConcurrentLinkedQueue不允许使用null元素。ConcurrentLinkedQueue实现了多线程的高效访问,多个线程访问ConcurrentLinkedQueue集合时无须等待。

ConcurrentHashMap

在默认情况下,ConcurrentHashMap支持16个线程并发写入,当有超过16个线程并发向该Map中写入数据时,可能有一些线程需要等待。实际上,程序通过设置concurrencyLevel构造参数(默认值为16)来支持更多的并发写入线程

java.util.concurrent包下的集合和java.util包的集合在迭代时的区别

使用java.util包下的Collection作为集合对象时,如果该集合对象创建迭代器后集合元素发生改变,则会引发ConcurrentModificationException异常。
与前面介绍的HashMap和普通集合不同的是,因为ConcurrentLinkedQueue和ConcurrentHashMap支持多线程并发访问,所以当使用迭代器来遍历集合元素时,该迭代器可能无法反映出创建迭代器之后所做的修改,并且程序不会抛出任何异常。

java8对ConcurrentHashMap的扩展

Java8扩展了ConcurrentHashMap的功能,Java8为该类新增了30多个新方法,这些方法可借助于Stream和Lambda表达式支持执行聚集操作。

Java8的ConcurrentHashMap新的的方法

ConcurrentHashMap新增的方法大致可分为如下三类。

• forEach系列(forEach,ForEachKey,ForEachValue,ForEachEntry)
• search系列(search,SearchKeys,SearchValues,SearchEntries)
• reduce系列(reduce,reduceToDouble,reduceToLong,reduceKeys,reduceValues)

除此之外,ConcurrentHashMap还新增了mappingCount()、newKeySet()等方法,增强后的ConcurrentHashMap更适合作为缓存实现类使用。

CopyOn开头的集合

由于CopyOnWriteArraySet的底层封装了CopyOnWriteArrayList,因此它的实现机制完全类似于CopyOnWriteArrayList集合。
对于CopyOnWriteArrayList集合,正如它的名字所暗示的,它釆用复制底层数组的方式来实现写操作

• 当线程对CopyOnWriteArrayList集合执行读取操作时,线程将会直接读取集合本身,无须加锁与阻塞。
• 当线程对CopyOnWriteArraylist集合执行写入操作时(包括调用add()、remove()、set()等方法),该集合会在底层复制一份新的数组,接下来对新的数组执行写入操作。
  • 由于对**CopyOnWriteArrayList集合的写入操作都是对数组的副本执行操作,因此它是线程安全的**

需要指出的是,由于CopyOnWriteArrayList执行写入操作时需要频繁地复制数组,性能比较差,但由于读操作与写操作不是操作同一个数组,而且读操作也不需要加锁,因此读操作就很快、很安全。
由此可见,CopyOnWriteArrayList适合用在读取操作远远大于写入操作的场景中,例如缓存等。

16.9.2 包装线程不安全的集合

前面介绍Java集合时所讲的ArrayList、Linkedlist、HashSet、TreeSet、HashMap、TreeMap等都是线程不安全的,也就是说,当多个并发线程向这些集合中存、取元素时,就可能会破坏这些集合的数据完整性。

Collections的synchronizedXxx方法

如果程序中有多个线程可能访问以上这些集合,就可以使用Collections提供的类方法把这些集合包装成线程安全的集合。
Collections提供了如下几个静态方法。

代码 使用线程安全的HashMap

例如需要在多线程中使用线程安全的HashMap对象,则可以采用如下代码:

1
2

//使用Collectionss的synchronizedMap方法将一个普通的HashMap包装成线程安全的类
HashMap m=Collections.synchronizedMap(new HashMap());

创建后就包装

如果需要把某个集合包装成线程安全的集合,则应该在创建之后立即包装

16.9 线程相关类

Java还为线程安全提供了一些工具类,如ThreadLocal类,它代表一个线程局部变量,通过把数据放在ThreadLocal中就可以让每个线程创建一个该变量的副本,从而避免并发访问的线程安全问题。除此之外,Java5还新增了大量的线程安全类。

16.9.1 Threadlocal类

早在JDK1.2推出之时,Java就为多线程编程提供了一个ThreadLocal类;从Java5.0以后,Java引入了泛型支持,Java为该ThreadLocal类增加了泛型支持,即:ThreadLocal<T>。通过使用ThreadLocal类可以简化多线程编程时的并发访问,使用这个ThreadLocal类可以很简捷地隔离多线程程序的竞争资源.
ThreadLocal,是ThreadLocalVariable(线程局部变量)的意思,也许将它命名为ThreadLocalVar更加合适。

线程局部变量的作用

线程局部变量(ThreadLocal)的功用其实非常简单,就是为每一个使用该变量的线程都提供一个变量值的副本,使每一个线程都可以独立地改变自己的副本,而不会和其他线程的副本冲突。从线程的角度看,就好像每一个线程都完全拥有该变量一样。

ThreadLocal常用方法

ThreadLocal类的用法非常简单,它只提供了如下三个public方法。

程序 ThreadLocal示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

class Account {
    /*
     * 定义一个ThreadLocal类型的变量，该变量将是一个线程局部变量 每个线程都会保留该变量的一个副本
     */
    private ThreadLocal<String> name = new ThreadLocal<>();

    // 定义一个初始化name成员变量的构造器
    public Account(String str) {
        this.name.set(str);
        // 下面代码用于访问当前线程的name副本的值
        System.out.println("---" + this.name.get());
    }

    // name的setter和getter方法
    public String getName() {
        return name.get();
    }

    public void setName(String str) {
        this.name.set(str);
    }
}

class MyTest extends Thread {
    // 定义一个Account类型的成员变量
    private Account account;

    public MyTest(Account account, String name) {
        super(name);
        this.account = account;
    }

    public void run() {
        // 循环10次
        for (int i = 0; i < 10; i++) {
            // 当i == 6时输出将账户名替换成当前线程名
            if (i == 6) {
                account.setName(getName());
            }
            // 输出同一个账户的账户名和循环变量
            System.out.println(account.getName() + " 账户的i值：" + i);
        }
    }
}

public class ThreadLocalTest {
    public static void main(String[] args) {
        // 启动两条线程，两条线程共享同一个Account
        Account at = new Account("初始名");
        /*
         * 虽然两条线程共享同一个账户，即只有一个账户名
         * 但由于账户名是ThreadLocal类型的，所以每条线程 都完全拥有各自的账户名副本，
         * 所以从i == 6之后，将看到两条 线程访问同一个账户时看到不同的账户名。
         */
        new MyTest(at, "线程甲").start();
        new MyTest(at, "线程乙").start();
    }
}

由于程序中的账户名是一个ThreadLocal变量,所以虽然程序中只有一个Account对象,但两个子线程将会产生两个账户名(主线程也持有一个账户名的副本)。两个线程进行循环时都会在i==6时将账户名 改为 线程名,这样就可以看到两个线程拥有两个账户名的情形:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

---初始名
null 账户的i值：0
null 账户的i值：0
null 账户的i值：1
null 账户的i值：1
null 账户的i值：2
null 账户的i值：2
null 账户的i值：3
null 账户的i值：3
null 账户的i值：4
null 账户的i值：5
null 账户的i值：4
线程乙 账户的i值：6
null 账户的i值：5
线程乙 账户的i值：7
线程甲 账户的i值：6
线程乙 账户的i值：8
线程甲 账户的i值：7
线程乙 账户的i值：9
线程甲 账户的i值：8
线程甲 账户的i值：9

从上面程序可以看出,实际上账户名有三个副本,主线程一个,另外启动的两个线程各一个,它们的值互不干扰,每个线程完全拥有自己的ThreadLocal变量,这就是ThreadLocal的用途。

ThreadLocal和加锁同步的区别

ThreadLocal和其他所有的同步机制一样,都是为了解决多线程中对同一变量的访问冲突.

• 在普通的同步机制中,是通过对象加锁来实现多个线程对同一变量的安全访问的。该变量是多个线程共享的,所以要使用这种同步机制,需要很细致地分析在什么时候对变量进行读写,什么时候需要锁定某个对象,什么时候释放该对象的锁等。在这种情况下,系统并没有将这份资源复制多份,只是采用了安全机制来控制对这份资源的访问而已。
• ThreadLocal从另一个角度来解决多线程的并发访问,ThreadLocal将需要并发访问的资源复制多份,每个线程拥有一份资源,每个线程都拥有自己的资源副本,从而也就没有必要对该变量进行同步了,ThreadLocal提供了线程安全的共享对象,在编写多线程代码时,可以把不安全的整个变量封装进ThreadLocal,或者把该对象与线程相关的状态使用ThreadLocal保存。

ThreadeLocal不能替代同步

ThreadLocal并不能替代同步机制,两者面向的问题领域不同。

• 同步机制是为了同步多个线程对相同资源的并发访问,是多个线程之间进行通信的有效方式;
• 而ThreadLocal是为了隔离多个线程的数据共享,从根本上避免多个线程之间对共享资源(变量)的竞争,也就不需要对多个线程进行同步了。

什么时候用同步机制 什么时候使用ThreadLocal

通常建议:

• 如果多个线程之间需要共享资源,以达到线程之间的通信功能,就使用同步机制;
• 如果仅仅需要隔离多个线程之间的共享冲突,则可以使用ThreadLocal

16.8.2 Java8增强的ForkJoinPool

现在计算机大多已向多CPU方向发展,即使普通PC,甚至小型智能设备(如手机)、多核处理器也已被广泛应用。在未来的日子里,处理器的核心数将会发展到更多。
虽然硬件上的多核CPU已经十分成熟,但很多应用程序并未为这种多核CPU做好准备,因此并不能很好地利用多核CPU的性能优势。
为了充分利用多CPU、多核CPU的性能优势,计算机软件系统应该可以充分“挖掘”每个CPU的计算能力,绝不能让某个CPU处于“空闲”状态。为了充分利用多CPU、多核CPU的优势,可以考虑把一个任务拆分成多个“小任务”,把多个“小任务”放到多个处理器核心上并行执行;当多个“小任务”执行完成之后,再将这些执行结果合并起来即可。

Java7提供的ForkJoinPool

Java7提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算,再把多个“小任务”的结果合并成总的计算结果。
ForkJoinPool是ExecutorService的实现类,因此是一种特殊的线程池。

ForkJoinPool构造器

ForkJoinPool提供了如下两个常用的构造器。

Java8对ForkJoinPool的扩展

Java8进一步扩展了ForkJoinPool的功能,Java8为ForkJoinPool增加了通用池功能。

ForkJoinPool类通过如下两个静态方法提供通用池功能。

ForkJoinTask

创建了ForkJoinPool实例之后,就可调用ForkJoinPool的submit(ForkJoinTask task)或invoke(ForkJoinTask task)方法来执行指定任务了。

• 其中**ForkJoinTask代表一个可以并行、合并的任务**。
• 这个ForkJoinTask是一个抽象类,它还有两个抽象子类:RecursiveAction和RecursiveTask。其中
  • RecursiveTask代表有返回值的任务,
  • RecursiveAction代表没有返回值的任务。

图16.14显示了ForkJoinPool、ForkJoinTask等类的类图。

无返回值的大任务分解

程序 使用ForkJoinPool将大任务拆分

下面以执行没有返回值的“大任务”(简单地打印0-300的数值)为例,程序将一个“大任务”拆分成多个“小任务”,并将任务交给ForkJoinPool来执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import java.util.concurrent.*;

// 继承RecursiveAction来实现"可分解"的任务
class PrintTask extends RecursiveAction {
    // 每个“小任务”只最多只打印50个数
    private static final int THRESHOLD = 50;
    private int start;
    private int end;

    // 打印从start到end的任务
    public PrintTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        // 当end与start之间的差小于THRESHOLD时，开始打印
        if (end - start < THRESHOLD) {
            for (int i = start; i < end; i++) {
                System.out.println(Thread.currentThread().getName() + "的i值：" + i);
            }
        } else {
            // 如果当end与start之间的差大于THRESHOLD时，即要打印的数超过50个
            // 将大任务分解成两个小任务。
            int middle = (start + end) / 2;
            PrintTask left = new PrintTask(start, middle);
            PrintTask right = new PrintTask(middle, end);
            // 并行执行两个“小任务”
            left.fork();
            right.fork();
        }
    }
}

public class ForkJoinPoolTest {
    public static void main(String[] args) throws Exception {
        ForkJoinPool pool = new ForkJoinPool();
        // 提交可分解的PrintTask任务
        pool.submit(new PrintTask(0, 300));
        pool.awaitTermination(2, TimeUnit.SECONDS);
        // 关闭线程池
        pool.shutdown();
    }
}

上面程序中的代码:

1
2
3
4
5
6
7
8

// 如果当end与start之间的差大于THRESHOLD时，即要打印的数超过50个
// 将大任务分解成两个小任务。
int middle = (start + end) / 2;
PrintTask left = new PrintTask(start, middle);
PrintTask right = new PrintTask(middle, end);
// 并行执行两个“小任务”
left.fork();
right.fork();

实现了对指定打印任务的分解,分解后的任务分别调用fork()方法开始并行执行运行上面程序,可以看到下所示的结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

ForkJoinPool-1-worker-1的i值：262
ForkJoinPool-1-worker-3的i值：37 
ForkJoinPool-1-worker-4的i值：0  
ForkJoinPool-1-worker-2的i值：112
ForkJoinPool-1-worker-4的i值：1  
ForkJoinPool-1-worker-3的i值：38 
ForkJoinPool-1-worker-1的i值：263
ForkJoinPool-1-worker-3的i值：39 
ForkJoinPool-1-worker-4的i值：2  
ForkJoinPool-1-worker-2的i值：113
ForkJoinPool-1-worker-4的i值：3  
ForkJoinPool-1-worker-3的i值：40 
ForkJoinPool-1-worker-1的i值：264
ForkJoinPool-1-worker-3的i值：41 
ForkJoinPool-1-worker-4的i值：4  
ForkJoinPool-1-worker-2的i值：114
ForkJoinPool-1-worker-4的i值：5  
......
ForkJoinPool-1-worker-2的i值：257
ForkJoinPool-1-worker-2的i值：258
ForkJoinPool-1-worker-2的i值：259
ForkJoinPool-1-worker-2的i值：260
ForkJoinPool-1-worker-2的i值：261

从执行结果来看,ForkJoinPool启动了4个线程来执行这个打印任务—这是因为测试计算机的CPU是4核的。不仅如此,读者可以看到程序虽然打印了0-299这300个数字,但并不是连续打印的,这是因为程序将这个打印任务进行了分解,分解后的任务会并行执行,所以不会按顺序从0打印到299。

有返回值的大任务分解

上面定义的任务是一个没有返回值的打印任务,如果大任务是有返回值的任务,则可以让任务继承RecursiveTask<T>,其中泛型参数T就代表了该任务的返回值类型。

程序 使用RecursiveTask对数组求和

下面程序示范了使用RecursiveTask对一个长度为100的数组的元素值进行累加。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

import java.util.concurrent.*;
import java.util.*;

// 继承RecursiveTask来实现"可分解"的任务
class CalTask extends RecursiveTask<Integer> {
    // 每个“小任务”只最多只累加20个数
    private static final int THRESHOLD = 20;
    private int arr[];
    private int start;
    private int end;

    // 累加从start到end的数组元素
    public CalTask(int[] arr, int start, int end) {
        this.arr = arr;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum = 0;
        // 当end与start之间的差小于THRESHOLD时，开始进行实际累加
        if (end - start < THRESHOLD) {
            for (int i = start; i < end; i++) {
                sum += arr[i];
            }
            return sum;
        } else {
            // 如果当end与start之间的差大于THRESHOLD时，即要累加的数超过20个时
            // 将大任务分解成两个小任务。
            int middle = (start + end) / 2;
            CalTask left = new CalTask(arr, start, middle);
            CalTask right = new CalTask(arr, middle, end);
            // 并行执行两个“小任务”
            left.fork();
            right.fork();
            // 把两个“小任务”累加的结果合并起来
            return left.join() + right.join(); // ①
        }
    }
}

public class Sum {
    public static void main(String[] args) throws Exception {
        int[] arr = new int[100];
        Random rand = new Random();
        int total = 0;
        // 初始化100个数字元素
        for (int i = 0, len = arr.length; i < len; i++) {
            int tmp = rand.nextInt(20);
            // 对数组元素赋值，并将数组元素的值添加到sum总和中。
            total += (arr[i] = tmp);
        }
        System.out.println(total);
        // 创建一个通用池
        ForkJoinPool pool = ForkJoinPool.commonPool();
        // 提交可分解的CalTask任务
        Future<Integer> future = pool.submit(new CalTask(arr, 0, arr.length));
        System.out.println(future.get());
        // 关闭线程池
        pool.shutdown();
    }
}

上面程序与前一个程序基本相似,同样是将任务进行了分解,并调用分解后的任务的fork()方法使它们并行执行。与前一个程序不同的是,现在任务是带返回值的,因此程序还在①号代码处将两个分解后的“小任务”的返回值进行了合并。
运行上面程序,将可以看到程序通过CalTask计算出来的总和,与初始化数组元素时统计出来的总和总是相等,这表明程序一切正常。

16.8 线程池

系统启动一个新线程的成本是比较高的,因为它涉及与操作系统交互。在这种情形下,使用线程池可以很好地提高性能,尤其是当程序中需要创建大量生存期很短暂的线程时,更应该考虑使用线程池。
与数据库连接池类似的是,线程池在系统启动时即创建大量空闲的线程,程序将一个Runnable对象或Callable对象传给线程池,线程池就会启动一个空闲的线程来执行它们的run()或call()方法,当run()或call()方法执行结束后,该线程并不会死亡,而是再次返回线程池中成为空闲状态,等待执行下一个Runnable对象的run()(或cal()方法。
除此之外,使用线程池可以有效地控制系统中并发线程的数量,当系统中包含大量并发线程时,会导致系统性能剧烈下降,甚至导致JVM崩溃,而线程池的最大线程数参数可以控制系统中并发线程数不超过此数。

16.8.1 Java8改进的线程池

Executors类

在Java5以前,开发者必须手动实现自己的线程池;从Java5开始,Java内建支持线程池。Java5新增了一个Executors工厂类来产生线程池,该工厂类包含如下几个静态工厂方法来创建线程池

创建ExecutorService线程池

ExecutorService对象对象代表一个线程池,它可以执行Runnable对象或Callable对象所代表的线程;

创建ScheduledExecutorService线程池

ScheduledExecutorService是ExecutorService的子类,它可以在指定延迟后执行线程任务;

java8 新增的workStealing池

这两个方法则是Java8新增的,这两个方法可充分利用多CPU并行的能力。这两个方法生成的workStealing池,都相当于后台线程池,如果所有的前台线程都死亡了,workStealing池中的线程会自动死亡。
由于目前计算机硬件的发展日新月异,即使普通用户使用的电脑通常也都是多核CPU,因此Java8在线程支持上也增加了利用多CPU并行的能力,这样可以更好地发挥底层硬件的性能。

尽快执行线程池ExecutorService

ExecutorService代表尽快执行线程的线程池(只要线程池中有空闲线程,就立即执行线程任务),程序只要将一个Runnable对象或Callable对象(代表线程任务)提交给该线程池,该线程池就会尽快执行该任务。

ExecutorService里提供了如下三个方法。

延迟周期线程池ScheduledExecutorService

ScheduledExecutorService代表可在指定延迟后或周期性地执行线程任务的线程池,它提供了如下4个方法。

关闭线程池

shutdown方法

用完一个线程池后,应该调用该线程池的shutdown方法,该方法将启动线程池的关闭序列,调用shutdown()方法后的线程池不再接收新任务,但会将以前所有已提交任务执行完成。当线程池中的所有任务都执行完成后,池中的所有线程都会死亡;

shutdownNow方法

另外也可以调用线程池的shutdownNow()方法来关闭线程池,该方法试图停止所有正在执行的活动任务,暂停处理正在等待的任务,并返回等待执行的任务列表。

使用线程池步骤

使用线程池来执行线程任务的步骤如下:

1. 调用Executors类的静态工厂方法创建一个ExecutorService对象,该对象代表一个线程池。
2. 创建Runnable实现类或Callable实现类的实例,作为线程执行任务。
3. 调用ExecutorService对象的submit方法来提交Runnable实例或Callable实例。
4. 当不想提交任何任务时,调用ExecutorService对象的shutdown()方法来关闭线程池。

程序 使用线程池

下面程序使用线程池来执行指定Runnable对象所代表的任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import java.util.concurrent.*;

public class ThreadPoolTest {
    public static void main(String[] args) throws Exception {
        // 创建足够的线程来支持4个CPU并行的线程池
        // 创建一个具有固定线程数（6）的线程池
        ExecutorService pool = Executors.newFixedThreadPool(6);
        // 使用Lambda表达式创建Runnable对象
        Runnable target = () -> {
            for (int i = 0; i < 100; i++) {
                System.out.println(Thread.currentThread().getName() + "的i值为:" + i);
            }
        };
        // 向线程池中提交两个线程
        pool.submit(target);
        pool.submit(target);
        // 关闭线程池
        pool.shutdown();
    }
}

上面程序中创建Runnable实现类与最开始创建线程池并没有太大差别,创建了Runnable实现类之后程序没有直接创建线程、启动线程来执行该Runnable任务,而是通过线程池来执行该任务,使用线程池来执行Runnable任务的代码如程序中粗体字代码所示。运行上面程序,将看到两个线程交替执行的效果,如下所示:

1
2
3
4
5
6
7
8
9
10
11

pool-1-thread-1的i值为:0
pool-1-thread-2的i值为:0 
pool-1-thread-1的i值为:1 
pool-1-thread-2的i值为:1 
pool-1-thread-2的i值为:2 
......
pool-1-thread-1的i值为:98
pool-1-thread-2的i值为:97
pool-1-thread-1的i值为:99
pool-1-thread-2的i值为:98
pool-1-thread-2的i值为:99

16.7 线程组和未处理的异常

线程组

Java使用ThreadGroup来表示线程组,它可以对一批线程进行分类管理,Java允许程序直接对线程组进行控制。对线程组的控制相当于同时控制这批线程。
用户创建的所有线程都属于指定线程组,如果程序没有显式指定线程属于哪个线程组,则该线程属于默认线程组。

子线程默认和父线程同组

在默认情况下,子线程和创建它的父线程处于同一个线程组内,例如A线程创建了B线程,并且没有指定B线程的线程组,则B线程属于A线程所在的线程组。

一天是不良人一辈子都是不良人

一旦某个线程加入了指定线程组之后,该线程将一直属于该线程组,直到该线程死亡,线程运行中途不能改变它所属的线程组。

Thread类线程组构造器

Thread类提供了如下几个构造器来设置新创建的线程属于哪个线程组。

获取当前线程的线程组

因为中途不可改变线程所属的线程组,所以Thread类没有提供setThreadGroup()方法来改变线程所属的线程组,但提供了一个getThreadGroup()方法来返回该线程所属的线程组,getThreadGroup()方法的返回值是ThreadGroup对象,表示一个线程组。

ThreadGroup构造器

ThreadGroup类提供了如下两个简单的构造器来创建实例。

上面两个构造器在创建线程组实例时都必须为其指定一个名字,也就是说,线程组总会具有一个字符串类型的名字,该名字可通过调用ThreadGroup的getName()方法来获取,但不允许改变线程组的名字。

ThreadGroup方法

ThreadGroup类提供了如下几个常用的方法来操作整个线程组里的所有线程。

程序 线程组

下面程序创建了几个线程,它们分别属于不同的线程组,程序还将一个线程组设置成后台线程组。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class MyThread extends Thread {
    // 提供指定线程名的构造器
    public MyThread(String name) {
        super(name);
    }

    // 提供指定线程名、线程组的构造器
    public MyThread(ThreadGroup group, String name) {
        super(group, name);
    }

    public void run() {
        for (int i = 0; i < 20; i++) {
            System.out.println(getName() + " 线程的i变量" + i);
        }
    }
}

public class ThreadGroupTest {
    public static void main(String[] args) {
        // 获取主线程所在的线程组，这是所有线程默认的线程组
        ThreadGroup mainGroup = Thread.currentThread().getThreadGroup();
        System.out.println("主线程组的名字：" + mainGroup.getName());
        System.out.println("主线程组是否是后台线程组：" + mainGroup.isDaemon());
        new MyThread("主线程组的线程").start();
        ThreadGroup tg = new ThreadGroup("新线程组");
        // 设为后台线程
        tg.setDaemon(true);
        System.out.println("tg线程组是否是后台线程组：" + tg.isDaemon());
        MyThread tt = new MyThread(tg, "tg组的线程甲");
        tt.start();
        new MyThread(tg, "tg组的线程乙").start();
    }
}

运行效果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

主线程组的名字：main
主线程组是否是后台线程组：false
tg线程组是否是后台线程组：true 
主线程组的线程 线程的i变量0    
主线程组的线程 线程的i变量1    
tg组的线程甲 线程的i变量0      
tg组的线程乙 线程的i变量0      
主线程组的线程 线程的i变量2    
...
tg组的线程乙 线程的i变量19
tg组的线程甲 线程的i变量10
...
主线程组的线程 线程的i变量19
tg组的线程甲 线程的i变量18
tg组的线程甲 线程的i变量19

线程组异常处理

处理线程组内抛出的异常

ThreadGroup内还定义了一个很有用的uncaughtException方法:

从Java5开始,Java加强了线程的异常处理,如果线程执行过程中抛出了一个未处理异常,JVM在结束该线程之前会自动查找是否有对应的Thread.UncaughtExceptionHandler对象,如果找到该处理器对象,则会调用该对象的uncaughtException(Thread t, Throwable e)方法来处理该异常。

Thread.UncaughtExceptionHandler是Thread类的一个静态内部接口,该接口内只有一个方法:void uncaughtException(Thread t, Throwable e),该方法中的t代表出现异常的线程,而e代表该线程抛出的异常。

设置线程异常处理器

Thread类提供了如下两个方法来设置异常处理器。

ThreadGroup类实现了ThreadUncaughtHandler接口,所以每个线程所属的线程组将会作为默认的异常处理器。当一个线程抛出未处理异常时,

• 首先,JVM会査找该异常对应的异常处理器(setUncaughtExceptionHandler方法设置的异常处理器),如果找到该异常处理器,则将调用该异常处理器处理该异常;
• 否则,JVM将会调用该线程所属的线程组对象的uncaughtException方法来处理该异常线程组处理异常的默认流程如下。
  • 如果该线程组有父线程组,则调用父线程组的uncaughtException()方法来处理该异常
  • 如果该线程实例所属的线程类有默认的异常处理器(由setDefaultUncaughtExceptionHandler方法设置的异常处理器),那么就调用该异常处理器来处理该异常。
  • 如果该异常对象是ThreadDeath的对象,则不做任何处理;否则,将异常跟踪栈的信息打印到System.Err错误输出流,并结束该线程。

程序 为线程设置异常处理器

下面程序为主线程设置了异常处理器,当主线程运行抛出未处理异常时,该异常处理器将会起作用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

// 定义自己的异常处理器
class MyExHandler implements Thread.UncaughtExceptionHandler {
    // 实现uncaughtException方法，该方法将处理线程的未处理异常
    public void uncaughtException(Thread t, Throwable e) {
        System.out.println(t + " 线程出现了异常：" + e);
    }
}

public class ExHandler {
    public static void main(String[] args) {
        // 设置主线程的异常处理器
        Thread.currentThread().setUncaughtExceptionHandler(new MyExHandler());
        int a = 5 / 0; // ①
        System.out.println("程序正常结束！");
    }
}

上面程序的主方法中为主线程设置了异常处理器,而①号代码处将引发一个未处理异常,则该异常处理器会负责处理该异常。运行该程序,会看到如下输出:

1

Thread[main,5,main] 线程出现了异常：java.lang.ArithmeticException: / by zero

异常处理器与通过catch捕获异常的不同

从上面程序的执行结果来看,虽然程序中粗体字代码指定了异常处理器对未捕获的异常进行处理,而且该异常处理器也确实起作用了,但程序依然不会正常结束。这说明异常处理器与通过catch捕获异常是不同的:

• 当使用catch捕获异常时,异常不会向上传播给上一级调用者;
• 但使用异常处理器对异常进行处理之后,异常依然会传播给上一级调用者