技术问答-7 java中的集合(2)-Collection
作者：互联网
Collection
一、方法概览
| 返回值 | 方法 | 说明 |
|---|---|---|
| int | size() | 集合大小 |
| boolean | isEmpty() | 是不是包含元素(return size == 0 ) |
| Iteratir | iterator() | 获取迭代器 |
| Object[] | toArray() | 转换成Object数组 |
| T[] | toArray(T[] a) | 转换成指定类型的数组 |
| boolean | add(E e) | 添加元素 |
| boolean | remove(Object o) | 删除指定元素 |
| boolean | containsAll(Collection<?> c) | 是否包含结合中所有元素 |
| boolean | addAll(Collection<? extends E> c ) | 添加集合c中所有元素 |
| boolean | removeAll(Collection<?> c ) | 移出包含的c结合中的值 |
| default boolean | removeIf(Predicate<? super E> filter ) | 遍历数组移除符合条件的数据（JDK1.8） |
| boolean | retainAll(Collection<?> c) | 仅保留此collection中那些也包含在指定collection的元素 |
| void clear() | 清空集合 | |
| Spliterator | spliterator() | 分割数组（多线程操作JDK1.8） |
| Stream | stream() | 流操作（JDK1.8） |
| default Stream | parallelStream() | 并行执行的流操作 |
二、什么是流Stream（此流非彼流）
- Stream是java8新增特性，被人称之为流
- Stream不是集合元素，它不是数据结构，因此它不保存数据，它是有关算法和计算的，它更像一个高级版本的Iterator.原始版本的Iterator只能显示的一个一个遍历元素并对其执行某种操作，而Stream只需要给出需要对其包含的元素执行什么操作，比如“过滤长度为10的字符串”等 Stream会隐式的在内部进行遍历，做出相应的数据操作。
- Stream与迭代器一样是单向的，不可重复的，数据只遍历一次
- Stream可以进行并行化操作，迭代器只能命令式的、串行化操作
- 当时用串行化方式去遍历时，每个item读完后才能读取下一个item。而是用并行去遍历时，数据会被分割成多个段，其中每一个段都在不同的线程中处理，然后将结果一起输出。
- Stream的并行操作依赖于java7引入的Fork/Join框架来拆分任务处理过程。
- JAVA并行API演变历程
1）1.0-1.4 java.lang.Thread
2 ) 1.5 java.util.concurrent
3 ) 1.6 Phasers 等
4） 1.7 Fork/Join 框架
5）1.8 Lambda - Stream的另一个大特点是，数据源本身可以是无限的
二、parallelStream
- 什么是 parallelStream
parallelStream 是一个并行执行的流，他通过默认的ForkJoinPool ，可能提高多线程任务的速度 - parallelStream作用
Stream具有平行处理的能力，处理的过程会分而治之，也就是将一个大任务切分成多个小任务（跟hadoop mapreduce一个意思吧），这表示每个任务都是一个操作
输出结果~~：
- 可以看到 展示不是按照顺序来的，
- 当然forEach 改成forEachOrdered 结果会是顺序的，
- 如果forEachOrdered 中间有其他 如filter()的中介操作，他会试着平行化处理，最终forEachOrdered 会以原数据的顺序处理，因此forEachOrdered 处理可能会失去平行化的优势
三、 ForkJoin
- ForkJoinPool是什么
想深入研究parallelStream 我们必须先了解ForkJoin框架和ForkJonPool ,因为parallelStream 依赖于ForkJonPool ，所以这里做简单介绍
ForkJoin框架是jdk1.7的新特性，他与ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用一个无限的队列保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有构造函数中传入数据，那么就会用当前计算机可用CPU数量作为线程的默认值
2.ForkJoinPool作用
- ForkJoinPool主要使用分治法来解决问题。典型的应用比如快速排序算法。ForkjoinPool需要使用相对少的线程来处理大量的任务（比如10个线程 执行10000个任务）
– ForkJoinPool 能让其中的线程创建新的任务，并挂起当前任务，此时线程就能够在队列中选择任务执行（ThreadPoolExecutor就不行） - 也就是说使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务，比如使用10个线程来完成100万任务，但是，使用ThreadPoolExeutor是不可能完成的，因为ThreadPoolExecutor中的Thread无法选择优先执行子任务，需要完成100万个具有父子任务时，业务要100万线程，这显然不合理~
四、窃取算法
- 窃取算法是什么？
ForkJoin最核心的地方就是利用现代硬件设备多核，在一个操作的时候会有空闲的cpu，那么如何利用好这个空闲的cpu就成了提高性能的关键，而这里我们要提到工作窃取算法，工作窃取算法是整个ForkJoin的核心，工作窃取算法值某个线程从其他队列里窃取任务来执行。 - 为什么需要使用窃取算法？
假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个独立的线程来执行队列里的任务，线程队列一一对应，比如X线程负责处理X队列里的任务。但是有的线程会先把自己手里的任务完成，而其他线程对应的队列里的任务等待处理。完成任务的线程与其等着，不如去帮其他线程做任务，于是它就去其他队里里边窃取一个任务来执行，而在这时它们会同时访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常 会使用双端队列，被窃取的线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
六、ForkJion ParallelStreams
Java8给ForkJoinPool添加了一个通用的线程池，这个线程池用来处理那些没有被显式提交到任何线程池的任务。他是ForkJoinPool类型上的一个静态元素，他拥有默认线程数量。当调用Arrays类上添加的新方法时（parallelSort ），自动并行化就会发生。
对于ForkJoinPool通用线程池的线程数量，通常使用默认值就可以了，即运行时计算机的数量。
package along;import java.util.ArrayList;import java.util.List;import java.util.Set;import java.util.concurrent.CopyOnWriteArraySet;import java.util.concurrent.CountDownLatch;public class javaTest {public static void main(String[] args) throws Exception {
System.out.println("Hello World!");// 构造一个10000个元素的集合
List<Integer> list = new ArrayList<>();for (int i = 0; i < 10000; i++) {
list.add(i);}// 统计并行执行list的线程
Set<Thread> threadSet = new CopyOnWriteArraySet<>();// 并行执行
list.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();// System.out.println(thread);// 统计并行执行list的线程
threadSet.add(thread);});
System.out.println("threadSet一共有" + threadSet.size() + "个线程");
System.out.println("系统一共有" + Runtime.getRuntime().availableProcessors()+ "个cpu");
List<Integer> list1 = new ArrayList<>();
List<Integer> list2 = new ArrayList<>();for (int i = 0; i < 100000; i++) {
list1.add(i);
list2.add(i);}
Set<Thread> threadSetTwo = new CopyOnWriteArraySet<>();
CountDownLatch countDownLatch = new CountDownLatch(2);
Thread threadA = new Thread(() -> {
list1.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();
threadSetTwo.add(thread);});
countDownLatch.countDown();});
Thread threadB = new Thread(() -> {
list2.parallelStream().forEach(integer -> {
Thread thread = Thread.currentThread();
threadSetTwo.add(thread);});
countDownLatch.countDown();});
threadA.start();
threadB.start();
countDownLatch.await();
System.out.print("threadSetTwo一共有" + threadSetTwo.size() + "个线程");}}//threadSet是 4 也就是是3个线程池线程+1个forEach线程本身//threadSetTwo 是5 也就是3个线程池线程+2个forEach线程//所以如果ForkJoinPool需要4个线程 将它的值设置为 3 即可ParallelStreams 的陷阱
parallelStreams不是万能的 不可乱用!
在这个例子中，我们想同时查询多个搜索引擎并且获得第一个返回的结果。
- 标红处，有问题！！
- 查询搜索引擎是一个阻塞操作。所以在某时刻所有线程都会调用get()方法并且在那里等待结果返回
- 由于ForkJoinPool workders的存在，这样平行的等待相对于使用主线程的等待会产生的一种副作用。
- 现在ForkJoin pool 的实现是:它并不会因为产生了新的workers 而抵消掉阻塞的workers。那么在某个时间所有ForkJoinPool.common()的线程都会被用光。
- 也就是说，下一次你调用这个查询方法，就可能会在一个时间与其他的parallel stream同时运行（因为还有上边的get()没有执行完），而导致第二个任务的性能大大受损。
小结：
- 当需要处理递归分治算法时，考虑使用ForkJoinPool。
- 仔细设置不再进行任务划分的阈值，这个阈值对性能有影响。
- Java 8中的一些特性会使用到ForkJoinPool中的通用线程池。在某些场合下，需要调整该线程池的默认的线程数量。
标签：java,Thread,队列,ForkJoinPool,Collection,任务,线程,问答,窃取 来源： https://blog.51cto.com/u_12198094/2704033