Java多线程＜二＞多线程经典场景

leetcode 多线程刷题

上锁上一次，还是上多次？
同步的顺序。

1. 交替打印字符

使用sychronize同步锁
使用lock锁
使用concurrent的默认机制
使用volitale关键字 + Thread.sleep() / Thread.yield机制
使用automic原子类

方式1 ：使用互斥访问state + Number中控制当前state进行

实现1：使用synchornized上锁，wait让出cpu
实现2：使用semophore上锁, sleep或者yield让出cpu
实现3：使用原子Integer进行访问 + yield或者sleep让出cpu
实现4：使用Lock进行访问 + condition让出cpu
实现5: 使用blockingQueue放入state,如果不是自己的state,在放进去，然后让出cpu。

方式2：使用互斥访问全局cur进行，cur代表当前数字是多少，如果cur >= n，就直接return让线程终止。

其中cur代表的是当前的数字是多少。
互斥的访问方式仍然是上面的那些种。

方式3：使用同步的通知模式

上面的方式，四个线程都是一直处于活跃状态，也就是Runnable的状态。(使用wait的除外)。另外判断是否可以运行都需要while进行判断。

但实际上，四个线程在同一时间，只需要一个线程可以运行。其他线程都必须进行阻塞。所以可以使用同步通知的方式进行，在其他线程运行的时候，阻塞另外的三个线程，并且运行完成一个线程后，可以实现精准通知另一个线程启动。

2. 打印0和奇偶数字

使用锁 sychornized和Lock
使用并发工具
- barrier
- semopher
使用cas + Thread.sleep/volatile + ThreadSleep
使用blocking que进行实现

经典模型

1. 生产者消费者的几种实现方式

操作系统课本上的经典信号量机制。
- 锁使用synchornized关键字
- 加上while(true)死循环

package cn.itedus.lottery.test;import lombok.SneakyThrows;import java.util.Stack;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantLock;/*** @author: Zekun Fu* @date: 2023/11/13 11:28* @Description:*/public class test434 {static Stack<String> que = new Stack<>();static Object full = new ReentrantLock();static Object empty = new ReentrantLock();static ReentrantLock lock = new ReentrantLock();static int n = 0;static final  int st = 10;static class Consumer {void consume() {while (true) {lock.lock();if (n > 0) {lock.unlock();System.out.println("消费者消费..." + que.pop());n--;synchronized (full) {full.notifyAll();}} else {lock.unlock();synchronized (empty) {try {empty.wait();} catch (InterruptedException e) {e.printStackTrace();}}}try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}}static class Producer {void produce() {while (true) {lock.lock();if (n < st) {lock.unlock();String id = "" + (int)(Math.random() * 100);System.out.println("生产者生产..." + id);que.add(id);n++;synchronized (empty) {empty.notifyAll();}} else {lock.unlock();synchronized (full) {try {full.wait();} catch (InterruptedException e) {e.printStackTrace();}}}try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}}public static void main(String[] args) {new Thread(new Runnable() {@Overridepublic void run() {Producer p = new Producer();p.produce();}}).start();new Thread(new Runnable() {@Overridepublic void run() {new Consumer().consume();}}).start();}
}

使用阻塞队列进行实现。 --> Java实现好的生产者消费者
手动加锁进行实现。 -->

2. 哲学家进餐

3. 读者写者

4. 并行的统计

第一个例子是课本上的匹配问题。
对于每一个文件夹可以使用线程池进行一个新的线程创建
最后对Future进行统计

package threadBase.threadPool;/*
*
*
*   java核心技术卷上面的线程池
* 使用线程池统计文件中含有关键字的文件个数
*  默认一个文件夹开启一个线程进行处理*/import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.*;public class Test1 {public static void main(String[] args) {String dir = "D:\\projects\\java\\javaBase\\threads\\data";System.out.println("文件夹的绝对路径是: " + dir);ExecutorService pool = Executors.newCachedThreadPool();String keyWord = "this";System.out.println("关键词是: " + keyWord);MatchCounter counter = new MatchCounter(pool, keyWord, new File(dir));Future<Integer> result = pool.submit(counter);try {System.out.println("含有关键词的文件个数为:" + result.get());} catch (Exception e) {e.printStackTrace();}int largestPoolSize = ((ThreadPoolExecutor)pool).getLargestPoolSize();System.out.println("线程池的最大数量是:" + largestPoolSize);pool.shutdown();                // 别忘了关闭线程池}
}class MatchCounter implements Callable<Integer> {private ExecutorService pool;private String keyWord;private File dir;public MatchCounter(ExecutorService pool, String keyWord, File dir) {this.pool = pool;this.keyWord = keyWord;this.dir = dir;}@Overridepublic Integer call() throws Exception {int cnt = 0;try {File[] files = dir.listFiles();List<Future<Integer>> ress = new ArrayList<>();for (File f: files) {           // 分治if (f.isDirectory()) {      // 开启新线程,从线程池中MatchCounter mc = new MatchCounter(pool, keyWord, f);Future<Integer>res = pool.submit(mc);ress.add(res);}else {                      // 如果是文件直接计算if (search(f)) cnt++;}}for (Future<Integer>res : ress) {cnt += res.get();}}catch (Exception e) {e.printStackTrace();}return cnt;}public boolean search(File file) {try {try (Scanner sc = new Scanner(file, "UTF-8")){boolean flag = false;while(!flag && sc.hasNextLine()) {String line = sc.nextLine();if (line.contains(keyWord)) flag = true;}return flag;}} catch (Exception e) {e.printStackTrace();}return false;}
}

5.并行的搜索

bfs的每一个结点开启一个新的线程进行搜索，使用并发的Map作为vis数组，使用并发的queue存入结点，同时使用并发的List放入结点。
适用于请求子节点会需要大量的时间的情况，这种适合一个异步的操作。在请求的时候，对以前请求到的结点进行一个过滤和统计。

package leetcode;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;/*
*
*   使用线程池 + future进行爬取
*
*
* */
public class Crawl4 {HashMap<String, List<String>> G = new HashMap<>();private class HtmlParser {List<String>getUrls(String start) {if (G.containsKey(start)) {List<String>ans = G.get(start);System.out.println("start = " + start + ", sz = "+ ans.size());return ans;}return new ArrayList<>();}}String hostName;private ConcurrentHashMap<String, Boolean> totalUrls = new ConcurrentHashMap<>();public List<String> crawl(String startUrl, HtmlParser htmlParser) {// bfs开始hostName = extractHostName(startUrl);ExecutorService pool = Executors.newCachedThreadPool();Future<List<String>>taskRes = pool.submit(new Chore(this, htmlParser, startUrl, pool));List<String>ans = new ArrayList<>();try {ans = taskRes.get();}catch (Exception e) {e.printStackTrace();}pool.shutdown();// System.out.println("最大的线程数量:" + ((ThreadPoolExecutor)pool).getLargestPoolSize());return ans;}private class Chore implements Callable<List<String>> {private Crawl4 solution;private HtmlParser htmlParser;private String urlToCrawl;private ExecutorService pool;public Chore(Crawl4 solution, HtmlParser htmlParser, String urlToCrawl, ExecutorService pool) {this.solution = solution;this.htmlParser = htmlParser;this.pool = pool;this.urlToCrawl = urlToCrawl;}@Overridepublic List<String> call() throws Exception {//            System.out.println("url = " + urlToCrawl);// 此处不需要使用并发的，因为统计只有主线程进行List<String>ans = new ArrayList<>();ans.add(urlToCrawl);this.solution.totalUrls.put(urlToCrawl, true);List<String> urls = htmlParser.getUrls(urlToCrawl);List<Future<List<String>>> ress = new ArrayList<>();for (String url : urls) {       // 每一个结点开启一个新的线程进行计算if (this.solution.totalUrls.containsKey(url)) continue;this.solution.totalUrls.put(url, true);String hostName = this.solution.extractHostName(url);if (!hostName.equals(this.solution.hostName)) continue;Chore c = new Chore(solution, htmlParser, url, pool);Future<List<String>> res = pool.submit(c);ress.add(res);}// 计算完成所有的任务，直接进行返回就行了for (Future<List<String>>f:ress) {ans.addAll(f.get());}return ans;}}private String extractHostName(String url) {String processedUrl = url.substring(7);int index = processedUrl.indexOf("/");if (index == -1) {return processedUrl;} else {return processedUrl.substring(0, index);}}public void build(int[][] edges) {String[] s = {"http://news.yahoo.com","http://news.yahoo.com/news","http://news.yahoo.com/news/topics/","http://news.google.com"};for (int[] e : edges) {String u = s[e[0]];String v = s[e[1]];if (G.containsKey(u)) {G.get(u).add(v);} else {List<String> l = new ArrayList<>();l.add(v);G.put(u, l);}}
//        for (String t : G.get("http://news.yahoo.com/news/topics/")) {
//            System.out.println(t);
//        }}public static void main(String[] args) {Crawl4 c = new Crawl4();String input = " [[0,2],[2,1],[3,2],[3,1],[3,0],[2,0]]";input = input.replace("[", "{");input = input.replace("]", "}");System.out.println(input);int[][] edges =   {{0,2},{2,1},{3,2},{3,1},{3,0},{2,0}};c.build(edges);List<String> ans = c.crawl("http://news.yahoo.com/news/topics/", c.new HtmlParser());for (String s: ans) {System.out.println(s);}}}

线程对效率的影响

1. 实现分治计算数组和

task放在循环外面

一个小实验，用来测试线程对性能的提升

计算数组中每一个数字乘以100的和。
使用双重循环计算，不用数学公式，这样计算时间长一点，容易做对比。
总共实现了四种不同的方式
- 使用单线程
- 使用4个手动写的线程
- 使用分治，先拷贝数组分成t份，之后进行合并
- 使用for循环生成4个手写的线程。

最后可以看到手动实现4个线程进行分治可以把效率提升到4倍左右。

详细代码请看下面代码1。
在这里插入图片描述

分析1：

由于我的计算机是8核16线程的，所以最多可以实现16个线程的并行计算。所以4个线程最多可以把效率提升到4倍。但是最多的效率提升就到16倍了。
使用分治，由于由大量的数组拷贝，所以计算的效率会低很多。
使用for循环创建线程，由于task的get()是阻塞的，会导致for循环没法执行，从而使得下面的线程没法执行。解决办法是
- 保存生成的task, 在for循环外面调用get方法。
- 之后的效率如下。

详细代码请看下面代码2。

在这里插入图片描述

分析2：提升线程个数带来的影响

可以看到最终的效率提升了15倍左右。
这是由于16个逻辑处理器并行工作的原因。
这么一看电脑设计的还不错。16个逻辑处理器能并行提升15倍的性能。这还是在上下文切换的情况下。在我代码垃圾的情况下。hhhh

详细代码请看下面代码3。

在这里插入图片描述

代码1

package threadBase.baseKey;import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.Random;
import java.util.Stack;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicInteger;/*
*
*
*   线程启动的三种方式
* 1. implement
* 2. extends
* 3. FutureT
*
* 4.
* */
public class StartThread {private static final int MAXN = 100000000;private static int[] nums = new int[MAXN];     // 计算数组中100个数字的和private AtomicInteger cur = new AtomicInteger();static {Arrays.fill(nums, 1);}private static long testSingle() throws Exception {long startTime = System.currentTimeMillis();long sum = 0;for (int i = 0; i < MAXN; i++) {for (int j = 0; j < 100; j++) {sum += nums[i];}}long endTime = System.currentTimeMillis();System.out.println("单线程: ");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}private static long test1() throws Exception{long startTime = System.currentTimeMillis();FutureTask<Long> task1 = new FutureTask<Long>(() -> {long tsum = 0;for (int i = 0; i < 25000000; i++) {for (int j = 0; j < 100; j++) {tsum += nums[i];}}return tsum;});FutureTask<Long> task2 = new FutureTask<Long>(() -> {long tsum = 0;for (int i = 25000000; i < 50000000; i++) {for (int j = 0; j < 100; j++) {tsum += nums[i];}}return tsum;});FutureTask<Long> task3 = new FutureTask<Long>(() -> {long tsum = 0;for (int i = 50000000; i < 75000000; i++) {for (int j = 0; j < 100; j++) {tsum += nums[i];}}return tsum;});FutureTask<Long> task4 = new FutureTask<Long>(() -> {long tsum = 0;for (int i = 75000000; i < 100000000; i++) {for (int j = 0; j < 100; j++) {tsum += nums[i];}}return tsum;});new Thread(task1).start();new Thread(task2).start();new Thread(task3).start();new Thread(task4).start();long sum = task1.get() + task2.get() + task3.get() + task4.get();long endTime = System.currentTimeMillis();System.out.println("4线程:");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}private static long test2() throws Exception{/***   首先需要一个线程进行任务的划分。*  然后由这个划分线程生成划分任务的线程数目。* 在有这个线程进程组装。* 在这使用主线程进行划分。* */int t = 5;                  // 划分线程的数量int len = MAXN / t;long sum = 0;long startTime = System.currentTimeMillis();for (int i = 0; i < t; i++) {           // 进行任务划分int[] numt = new int[len];for (int j = i * len; j < (i + 1) * len; j++) {numt[j - (i * len)] = nums[j];}// 线程执行FutureTask<Long>task = new FutureTask<Long>(()->{long ans = 0;for (int x: numt) {for (int j = 0; j < 100; j++) {ans += x;}}return ans;});new Thread(task).start();sum += task.get();}long endTime = System.currentTimeMillis();System.out.println("使用分治进行" + t + "线程划分执行:");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}private static long test3() throws Exception {StartThread startThread = new StartThread();int cnt = 4;                    // 控制线程的个数int sz = MAXN / cnt;            // 每一个线程计算的数量是多少long sum = 0;                        // 计算和是多少long startTime = System.currentTimeMillis();for (int i = 0; i < cnt; i++) {FutureTask<Long> task = new FutureTask<Long>(()->{long ans = 0L;int bg = startThread.cur.getAndIncrement();for (int j = bg * sz; j < (bg + 1) * sz; j++) {for (int k = 0; k <100; k++) {ans += nums[j];}}return ans;});new Thread(task).start();sum += task.get();}long endTime = System.currentTimeMillis();System.out.println(cnt + "个线程:");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}// 可以从第三遍开始，统计一个平均的时间public static void main(String[] args)throws Exception {long t1 = 0, t2 = 0, t3 = 0, t4 = 0;for (int i = 0; i < 11; i++) {      // 后面的8轮次进行统计System.out.println("-----------第" + i + "轮----------");if (i >= 3) {t1 += testSingle();t2 += test1();t3 += test2();t4 += test3();} else {testSingle();test1();test2();test3();}}System.out.println("平均时间:");System.out.println("单线程:" + t1 / 8 + "ms");System.out.println("4个手动多线程:" + t2 / 8 + "ms");System.out.println("4个分治多线程:" + t3 / 8 + "ms");System.out.println("for循环多线程:" + t4 / 8 + "ms");}
}

代码2

主要就是修改了test2和test3的方法。

把task.get()放在循环外面计算。
使用数组保存生成的task。

    private static long test2() throws Exception{/***   首先需要一个线程进行任务的划分。*  然后由这个划分线程生成划分任务的线程数目。* 在有这个线程进程组装。* 在这使用主线程进行划分。* */int t = 5;                  // 划分线程的数量int len = MAXN / t;long sum = 0;long startTime = System.currentTimeMillis();FutureTask<Long>[]tasks = new FutureTask[t];for (int i = 0; i < t; i++) {           // 进行任务划分int[] numt = new int[len];for (int j = i * len; j < (i + 1) * len; j++) {numt[j - (i * len)] = nums[j];}// 线程执行FutureTask<Long>task = new FutureTask<Long>(()->{long ans = 0;for (int x: numt) {for (int j = 0; j < 100; j++) {ans += x;}}return ans;});new Thread(task).start();tasks[i] = task;
//            sum += task.get();            // 这个会阻塞线程，所以会慢}for (int i = 0; i < 4; i++) {sum += tasks[i].get();}long endTime = System.currentTimeMillis();System.out.println("使用分治进行" + t + "线程划分执行:");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}public static long test4() throws Exception{StartThread startThread = new StartThread();int cnt = 4;                    // 控制线程的个数int sz = MAXN / cnt;            // 每一个线程计算的数量是多少long sum = 0;                        // 计算和是多少long startTime = System.currentTimeMillis();FutureTask<Long>[]tasks = new FutureTask[cnt];for (int i = 0; i < cnt; i++) {FutureTask<Long> task = new FutureTask<Long>(()->{long ans = 0L;int bg = startThread.cur.getAndIncrement();for (int j = bg * sz; j < (bg + 1) * sz; j++) {for (int k = 0; k <100; k++) {ans += nums[j];}}return ans;});new Thread(task).start();tasks[i] = task;}for (int i = 0; i < cnt; i++) {sum += tasks[i].get();}long endTime = System.currentTimeMillis();System.out.println(cnt + "个线程:");System.out.println("sum = " + sum + " t = " + (endTime - startTime) + "ms");return endTime - startTime;}

代码3

测试多少个线程对性能提升最大
结论是逻辑处理器的个数个。

    public static void testNumOfThread() throws Exception{int cnt = 32;long []times =  new long[cnt];for (int i = 1; i < cnt; i++) {times[i] = test4(i);}for (int i = 1; i < cnt; i++) {System.out.println(i + "个线程:" + times[i] + "ms");}}// 可以从第三遍开始，统计一个平均的时间public static void main(String[] args)throws Exception {testNumOfThread();}