多线程

Author: gitking

src/com/yale/test/math/AtomicIntegerDemo.java

@@ -0,0 +1,44 @@
+package com.yale.test.math;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/*
+ * Java的java.util.concurrent包除了提供底层锁、并发集合外，还提供了一组原子操作的封装类，它们位于java.util.concurrent.atomic包。
+ * 我们以AtomicInteger为例，它提供的主要操作有：
+ * 增加值并返回新值：int addAndGet(int delta)
+ * 加1后返回新值：int incrementAndGet()
+ * 获取当前值：int get()
+ * 用CAS方式设置：int compareAndSet(int expect, int update)
+ *Atomic类是通过无锁（lock-free）的方式实现的线程安全（thread-safe）访问。它的主要原理是利用了CAS：Compare and Set。
+ * 如果我们自己通过CAS编写incrementAndGet()，它大概长这样：
+ * 小结
+ * 使用java.util.concurrent.atomic提供的原子操作可以简化多线程编程：
+ * 原子操作实现了无锁的线程安全；
+ * 适用于计数器，累加器等。
+ */
+public class AtomicIntegerDemo {
+	
+	/*
+	 * CAS是指，在这个操作中，如果AtomicInteger的当前值是prev，那么就更新为next，返回true。如果AtomicInteger的当前值不是prev，就什么也不干，返回false。
+	 * 通过CAS操作并配合do ... while循环，即使其他线程修改了AtomicInteger的值，最终的结果也是正确的。
+	 * 通常情况下，我们并不需要直接用do ... while循环调用compareAndSet实现复杂的并发操作，而是用incrementAndGet()这样的封装好的方法，因此，使用起来非常简单。
+	 * 在高度竞争的情况下，还可以使用Java 8提供的LongAdder和LongAccumulator。
+	 */
+	public int incrementAndGet(AtomicInteger ai) {
+		int prev, next;
+		do {
+			prev = ai.get();
+			next = prev + 1;
+		} while(!ai.compareAndSet(prev, next));
+		return next;
+	}
+}
+
+class IdGenerator {
+	//我们利用AtomicLong可以编写一个多线程安全的全局唯一ID生成器：
+	AtomicLong al = new AtomicLong(0);
+	public long getNextId() {
+		return al.incrementAndGet();
+	}
+}

src/com/yale/test/math/array/ArraysTest.java

@@ -110,6 +110,15 @@ public static void main(String[] args) {
 		Arrays.sort(dataC);
 		System.out.println("排序后的数组:" + Arrays.toString(dataC));
 
+        int [] dataCopy = Arrays.copyOf(dataC, dataC.length);
+        System.out.println("复制数组dataC:" + Arrays.toString(dataCopy));
+        
+        /*
+         * Fork/Join线程池在Java标准库中就有应用。Java标准库提供的java.util.Arrays.parallelSort(array)可以进行并行排序，
+         * 它的原理就是内部通过Fork/Join对大数组分拆进行并行排序，在多核CPU上就可以大大提高排序的速度。
+         */
+        Arrays.parallelSort(dataCopy);//1.8之后才有的
+		
 		HashMap<String, String> paramMap = new HashMap<String, String>();
 		paramMap.put("a", "A");
 		paramMap.put("b", "B");

src/com/yale/test/math/array/CollectionsDemo.java

@@ -79,10 +79,12 @@ public static void main(String[] args) {
 
 
 		//Collections还提供了一组方法，可以把线程不安全的集合变为线程安全的集合：
-		//因为从Java 5开始，引入了更高效的并发集合类，所以上述这几个同步方法已经没有什么用了。
+		//因为从Java 5开始，引入了更高效的并发集合类，所以下面这几个同步方法已经没有什么用了。
+		//https://www.liaoxuefeng.com/wiki/1252599548343744/1299919855943714
 		List<String> listSafe = Collections.synchronizedList(listTest);
 		Set<String> setSafe = Collections.synchronizedSet(singleSet);
 		Map<String,String> mapSafe = Collections.synchronizedMap(singleMap);
+		//但是它实际上synchronizedMap是用一个包装类包装了非线程安全的Map，然后对所有读写方法都用synchronized加锁，这样获得的线程安全集合的性能比java.util.concurrent集合要低很多，所以不推荐使用。
 
 		List<String> strList = new ArrayList<String>();
 		Collections.addAll(strList, "A", "B", "C");//相当于调用了3次add方法

src/com/yale/test/thread/heima/zhangxiaoxiang/BlockingQueueTest.java

@@ -6,8 +6,24 @@
 
 /**
  * 先进先出就叫队列,BlockingQueue是阻塞队列的接口,
+ * 我们在前面已经通过ReentrantLock和Condition实现了一个BlockingQueue：
+ * BlockingQueue的意思就是说，当一个线程调用这个TaskQueue的getTask()方法时，该方法内部可能会让线程变成等待状态，直到队列条件满足不为空，线程被唤醒后，getTask()方法才会返回。
+ * 因为BlockingQueue非常有用，所以我们不必自己编写，可以直接使用Java标准库的java.util.concurrent包提供的线程安全的集合：ArrayBlockingQueue。
+ * 除了BlockingQueue外，针对List、Map、Set、Deque等，java.util.concurrent包也提供了对应的并发集合类。我们归纳一下：
+ * interface	non-thread-safe				thread-safe
+	List		ArrayList				CopyOnWriteArrayList
+	Map			HashMap					ConcurrentHashMap
+	Set			HashSet/TreeSet			CopyOnWriteArraySet
+	Queue		ArrayDeque/LinkedList	ArrayBlockingQueue/LinkedBlockingQueue
+	Deque		ArrayDeque/LinkedList	LinkedBlockingDeque
+	使用这些并发集合与使用非线程安全的集合类完全相同。我们以ConcurrentHashMap为例：
+	因为所有的同步和加锁的逻辑都在集合内部实现，对外部调用者来说，只需要正常按接口引用，其他代码和原来的非线程安全代码完全一样。即当我们需要多线程访问时，把：
+	Map<String, String> map = new HashMap<>();改为：Map<String, String> map = new ConcurrentHashMap<>();就可以了。
+	java.util.Collections工具类还提供了一个旧的线程安全集合转换器，可以这么用：
+	Map unsafeMap = new HashMap();
+	Map threadSafeMap = Collections.synchronizedMap(unsafeMap);
+	但是它实际上是用一个包装类包装了非线程安全的Map，然后对所有读写方法都用synchronized加锁，这样获得的线程安全集合的性能比java.util.concurrent集合要低很多，所以不推荐使用。
  * @author dell
- *
  */
 public class BlockingQueueTest {
 

src/com/yale/test/thread/heima/zhangxiaoxiang/CallableAndFuture.java

@@ -13,6 +13,26 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
 
+/*
+ * Runnable接口有个问题，它的方法没有返回值。如果任务需要一个返回结果，那么只能保存到变量，还要提供额外的方法读取，非常不便。
+ * 所以，Java标准库还提供了一个Callable接口，和Runnable接口比，它多了一个返回值：并且Callable接口是一个泛型接口，可以返回指定类型的结果。
+ * 现在的问题是，如何获得异步执行的结果？
+ * 如果仔细看ExecutorService.submit()方法，可以看到，它返回了一个Future类型，一个Future类型的实例代表一个未来能获取结果的对象：
+ * ExecutorService executor = Executors.newFixedThreadPool(4); 
+	// 定义任务:
+	Callable<String> task = new Task();
+	// 提交任务并获得Future:
+	Future<String> future = executor.submit(task);
+	// 从Future获取异步执行返回的结果:
+	String result = future.get(); // 可能阻塞
+	当我们提交一个Callable任务后，我们会同时获得一个Future对象，然后，我们在主线程某个时刻调用Future对象的get()方法，就可以获得异步执行的结果。在调用get()时，
+	如果异步任务已经完成，我们就直接获得结果。如果异步任务还没有完成，那么get()会阻塞，直到任务完成后才返回结果。
+	一个Future<V>接口表示一个未来可能会返回的结果，它定义的方法有：
+    get()：获取结果（可能会等待）
+    get(long timeout, TimeUnit unit)：获取结果，但只等待指定的时间；
+    cancel(boolean mayInterruptIfRunning)：取消当前任务；
+    isDone()：判断任务是否已完成。
+ */
 public class CallableAndFuture {
 
 	public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

src/com/yale/test/thread/heima/zhangxiaoxiang/CompletableFutureDemo.java

@@ -0,0 +1,231 @@
+package com.yale.test.thread.heima.zhangxiaoxiang;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.ThreadMXBean;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+
+/*
+ * 使用CompletableFuture
+ * 使用Future获得异步执行结果时，要么调用阻塞方法get()，要么轮询看isDone()是否为true，这两种方法都不是很好，因为主线程也会被迫等待。
+ * 从Java 8开始引入了CompletableFuture，它针对Future做了改进，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法。
+ * 我们以获取股票价格为例，看看如何使用CompletableFuture：
+ * 小结
+	CompletableFuture可以指定异步处理流程：
+    	thenAccept()处理正常结果；
+    	exceptional()处理异常结果；
+    	thenApplyAsync()用于串行化另一个CompletableFuture；
+    	anyOf()和allOf()用于并行化多个CompletableFuture。
+ */
+public class CompletableFutureDemo {
+	public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
+		/*
+		 * 创建异步执行任务
+		 * 创建一个CompletableFuture是通过CompletableFuture.supplyAsync()实现的，它需要一个实现了Supplier接口的对象：
+		 * public interface Supplier<T> {
+			    T get();
+			}
+		 * 这里我们用lambda语法简化了一下，直接传入CompletableFutureDemo::fetchPrice，因为CompletableFutureDemo.fetchPrice()静态方法的签名符合Supplier接口的定义（除了方法名外）。
+		 * 紧接着，CompletableFuture已经被提交给默认的线程池执行了，我们需要定义的是CompletableFuture完成时和异常时需要回调的实例。完成时，CompletableFuture会调用Consumer对象：
+		 * public interface Consumer<T> {
+			    void accept(T t);
+			}
+		 * 异常时，CompletableFuture会调用Function对象：
+		 * public interface Function<T, R> {
+			    R apply(T t);
+			}
+		 * 这里我们都用lambda语法简化了代码。
+		 * 可见CompletableFuture的优点是：
+			    异步任务结束时，会自动回调某个对象的方法；
+			    异步任务出错时，会自动回调某个对象的方法；
+			    主线程设置好回调后，不再关心异步任务的执行。
+		 * 
+		 */
+		CompletableFuture<Double> cf = CompletableFuture.supplyAsync(CompletableFutureDemo::fetchPrice);//supplyAsync内部会启动线程
+		
+		cf.thenAccept((result)->{//如果执行成功
+			System.out.println("price:" + result);
+		});
+		
+		cf.exceptionally((e) -> {//如果执行异常
+			e.printStackTrace();
+			return null;
+		});
+		
+		//主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭;
+		Thread.sleep(200);
+		
+		
+		/*
+		 * 如果只是实现了异步回调机制，我们还看不出CompletableFuture相比Future的优势。CompletableFuture更强大的功能是，多个CompletableFuture可以串行执行，
+		 * 例如，定义两个CompletableFuture，第一个CompletableFuture根据证券名称查询证券代码，第二个CompletableFuture根据证券代码查询证券价格，这两个CompletableFuture实现串行操作如下：
+		 */
+		CompletableFuture<String> cfQuery = CompletableFuture.supplyAsync(()->{//第一个任务
+			return queryCode("中国石油");
+		});
+		
+		CompletableFuture<Double> cfFetch = cfQuery.thenApplyAsync((code)->{//cfQuery成功后继续执行下一个任务
+			return fetchPrice(code);
+		});
+		
+		cfFetch.thenAccept((result)->{//cfFetch成功后打印结果:
+			System.out.println("price:" + result);
+		});
+		//主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭
+		Thread.sleep(2000);
+		
+		/*
+		 * 除了串行执行外，多个CompletableFuture还可以并行执行。例如，我们考虑这样的场景：
+		 * 同时从新浪和网易查询证券代码，只要任意一个返回结果，就进行下一步查询价格，查询价格也同时从新浪和网易查询，只要任意一个返回结果，就完成操作：
+		 * 下面逻辑实现的异步查询规则实际上是：
+			┌─────────────┐ ┌─────────────┐
+			│ Query Code  │ │ Query Code  │
+			│  from sina  │ │  from 163   │
+			└─────────────┘ └─────────────┘
+			       │               │
+			       └───────┬───────┘
+			               ▼
+			        ┌─────────────┐
+			        │    anyOf    │
+			        └─────────────┘
+			               │
+			       ┌───────┴────────┐
+			       ▼                ▼
+			┌─────────────┐  ┌─────────────┐
+			│ Query Price │  │ Query Price │
+			│  from sina  │  │  from 163   │
+			└─────────────┘  └─────────────┘
+			       │                │
+			       └────────┬───────┘
+			                ▼
+			         ┌─────────────┐
+			         │    anyOf    │
+			         └─────────────┘
+			                │
+			                ▼
+			         ┌─────────────┐
+			         │Display Price│
+			         └─────────────┘
+		 * 除了anyOf()可以实现“任意个CompletableFuture只要一个成功”，allOf()可以实现“所有CompletableFuture都必须成功”，这些组合操作可以实现非常复杂的异步流程控制。
+		 * 最后我们注意CompletableFuture的命名规则：
+		 * 	xxx()：表示该方法将继续在已有的线程中执行；
+		 * 	xxxAsync()：表示将异步在线程池中执行。
+		 * CompletableFuture可以指定异步处理流程：
+		 * thenAccept()处理正常结果；
+		 * exceptional()处理异常结果；
+		 * thenApplyAsync()用于串行化另一个CompletableFuture；
+		 * anyOf()和allOf()用于并行化多个CompletableFuture。
+		 */
+		CompletableFuture<String> cfQueryFromSina = CompletableFuture.supplyAsync(()->{//俩个CompletableFuture执行异步查询
+			return queryCode("中国石油", "https://finace.sina.com.cn/code/");
+		});
+		
+		CompletableFuture<String> cfQueryForm163 = CompletableFuture.supplyAsync(()->{
+			return queryCode("中国石油", "https://money.163.com/code/");
+		});
+		
+		//用anyof合并为一个新的CompletableFuture:
+		CompletableFuture<Object> cfQueryMegre = CompletableFuture.anyOf(cfQueryFromSina, cfQueryForm163);
+		
+		CompletableFuture<Double> cfFetchFromSina = cfQueryMegre.thenApplyAsync((code)->{//俩个CompletableFuture执行异步查询
+			return fetchPrice((String)code, "https://finance.sina.com.cn/price/");
+		});
+		
+		CompletableFuture<Double> cfFetchFrom163 = cfQueryMegre.thenApplyAsync((code)->{
+			return fetchPrice((String)code, "https://money.163.com/price");
+		});
+		
+		//用anyof合并为一个新的CompletableFuture:
+		CompletableFuture<Object> cfFetchSec = CompletableFuture.anyOf(cfFetchFromSina, cfFetchFrom163);
+		cfFetchSec.thenAccept((result)->{//最终结果
+			System.out.println("price:" + result);
+		});
+		//主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭
+		Thread.sleep(2000);
+		
+		/**
+		 * 咨询一下老师。我看到supplyAsync在thenAccept的前面，而且任务是在supplyAsync中启动的。那有没有可能在通过cf.thenAccept设置回调前，
+		 * 异步执行的任务已经完成了（假如fetchPrice是个空函数），这样在thenAccept中设置的回调就不会被执行了，有这种可能吗？
+		 * 答案是:可以执行
+		 */
+		// 创建异步执行任务:
+		CompletableFuture<Double> cfTime = CompletableFuture.supplyAsync(CompletableFutureDemo::fetchPriceZer);
+		// 暂停让Future完成:
+		Thread.sleep(5000);
+		// 如果执行成功:
+		cfTime.thenAccept((result) -> {
+			System.out.println("thenAccept: " + result);
+		});
+		// 如果执行异常:
+		cfTime.exceptionally((e) -> {
+			e.printStackTrace();
+			return null;
+		});
+		// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
+		Thread.sleep(2000);
+	}
+	
+	static Double fetchPriceZer() {
+		System.out.println("fetched!");
+		return 123.46;
+	}
+	
+	static Double fetchPrice() {
+		try {
+			Thread.sleep(100);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		if (Math.random() < 0.3) {
+			throw new RuntimeException("fetch price failed");
+		}
+		return 5 + Math.random() * 20;
+	}
+	
+	static String queryCode(String name) {
+		try {
+			Thread.sleep(100);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		return "601857";
+	}
+	
+	static Double fetchPrice(String code) {
+		try {
+			Thread.sleep(100);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		return 5 + Math.random() * 20;
+	}
+	
+	static String queryCode(String name, String url) {
+		System.out.println("query Code " + name + " from " + url + ".....");
+		try {
+			Thread.sleep((long)(Math.random() * 100));
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		return "601857";
+	}
+	
+	static Double fetchPrice(String code, String url) {
+		System.out.println("query price" + code + " from " + url);
+		try {
+			Thread.sleep((long)(Math.random() * 100));
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+		return 5 + Math.random() * 20;
+	}
+}