10-ForkJoin与享元模式
ForkJoin
Fork/Join：线程池的实现，体现是分治思想，适用于能够进行任务拆分的 CPU 密集型运算，用于并行计算
任务拆分：将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列都可以用分治思想进行求解
-
Fork/Join 在分治的基础上加入了多线程，把每个任务的分解和合并交给不同的线程来完成，提升了运算效率
-
ForkJoin 使用 ForkJoinPool 来启动，是一个特殊的线程池，默认会创建与 CPU 核心数大小相同的线程池
-
任务有返回值继承 RecursiveTask，没有返回值继承 RecursiveAction
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool(4);
System.out.println(pool.invoke(new MyTask(5)));
//拆分 5 + MyTask(4) --> 4 + MyTask(3) -->
}
// 1~ n 之间整数的和
class MyTask extends RecursiveTask<Integer> {
private int n;
public MyTask(int n) {
this.n = n;
}
@Override
public String toString() {
return "MyTask{" + "n=" + n + '}';
}
@Override
protected Integer compute() {
// 如果 n 已经为 1，可以求得结果了
if (n == 1) {
return n;
}
// 将任务进行拆分(fork)
MyTask t1 = new MyTask(n - 1);
t1.fork();
// 合并(join)结果
int result = n + t1.join();
return result;
}
}继续拆分优化：
class AddTask extends RecursiveTask<Integer> {
int begin;
int end;
public AddTask(int begin, int end) {
this.begin = begin;
this.end = end;
}
@Override
public String toString() {
return "{" + begin + "," + end + '}';
}
@Override
protected Integer compute() {
// 5, 5
if (begin == end) {
return begin;
}
// 4, 5 防止多余的拆分 提高效率
if (end - begin == 1) {
return end + begin;
}
// 1 5
int mid = (end + begin) / 2; // 3
AddTask t1 = new AddTask(begin, mid); // 1,3
t1.fork();
AddTask t2 = new AddTask(mid + 1, end); // 4,5
t2.fork();
int result = t1.join() + t2.join();
return result;
}
}ForkJoinPool 实现了工作窃取算法来提高 CPU 的利用率：
- 每个线程都维护了一个双端队列，用来存储需要执行的任务
- 工作窃取算法允许空闲的线程从其它线程的双端队列中窃取一个任务来执行
- 窃取的必须是最晚的任务，避免和队列所属线程发生竞争，但是队列中只有一个任务时还是会发生竞争
享元模式
享元模式（Flyweight pattern）： 用于减少创建对象的数量，以减少内存占用和提高性能，这种类型的设计模式属于结构型模式，它提供了减少对象数量从而改善应用所需的对象结构的方式
异步模式：让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务，也可将其归类为分工模式，典型实现就是线程池
工作机制：享元模式尝试重用现有的同类对象，如果未找到匹配的对象，则创建新对象
自定义连接池：
public static void main(String[] args) {
Pool pool = new Pool(2);
for (int i = 0; i < 5; i++) {
new Thread(() -> {
Connection con = pool.borrow();
try {
Thread.sleep(new Random().nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
pool.free(con);
}).start();
}
}
class Pool {
//连接池的大小
private final int poolSize;
//连接对象的数组
private Connection[] connections;
//连接状态数组 0表示空闲 1表示繁忙
private AtomicIntegerArray states; //int[] -> AtomicIntegerArray
//构造方法
public Pool(int poolSize) {
this.poolSize = poolSize;
this.connections = new Connection[poolSize];
this.states = new AtomicIntegerArray(new int[poolSize]);
for (int i = 0; i < poolSize; i++) {
connections[i] = new MockConnection("连接" + (i + 1));
}
}
//使用连接
public Connection borrow() {
while (true) {
for (int i = 0; i < poolSize; i++) {
if (states.get(i) == 0) {
if (states.compareAndSet(i, 0, 1)) {
System.out.println(Thread.currentThread().getName() + " borrow " + connections[i]);
return connections[i];
}
}
}
//如果没有空闲连接，当前线程等待
synchronized (this) {
try {
System.out.println(Thread.currentThread().getName() + " wait...");
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
//归还连接
public void free(Connection con) {
for (int i = 0; i < poolSize; i++) {
if (connections[i] == con) {//判断是否是同一个对象
states.set(i, 0);//不用cas的原因是只会有一个线程使用该连接
synchronized (this) {
System.out.println(Thread.currentThread().getName() + " free " + con);
this.notifyAll();
}
break;
}
}
}
}
class MockConnection implements Connection {
private String name;
//.....
}