Java多线程与并发库

同步方式

import javax.xml.stream.events.StartDocument;public class TestSynchronized {public static void main(String[] args) {// TODO Auto-generated method stubTestSynchronized test = new TestSynchronized();test.init();}void init() {final Outputer outputer = new Outputer();new Thread(new Runnable(){@Overridepublic void run() {// TODO Auto-generated method stubwhile(true){try {Thread.sleep(10);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}outputer.Output("wangyuyuyuyuyuyuyuyu");}}}){}.start();new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubwhile(true){try {Thread.sleep(10);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}outputer.Output("zhouzhanzhaozhaozhaozhao");}}}){}.start();}class Outputer{final static String lockKey = "lock";//定义输出函数//第一种方式，提供某个锁public void Output(String name){int len = name.length();synchronized (lockKey) {for (int i = 0; i < len; i++){System.out.print(name.charAt(i));}System.out.println();}}//第二种方式，锁住自己public void Output1(String name){int len = name.length();synchronized (this) { //也可以用Outputer.classfor (int i = 0; i < len; i++){System.out.print(name.charAt(i));}System.out.println();}}//第三种方式，函数前面加关键字synchronizedpublic synchronized void Output2(String name){int len = name.length();for (int i = 0; i < len; i++){System.out.print(name.charAt(i));}System.out.println();}}}

定时器Timer

import java.util.Timer;
import java.util.TimerTask;public class TimerTest {public static void main(String[] args){new Timer().schedule(new TimerTask(){@Overridepublic void run() {// TODO Auto-generated method stubSystem.out.println("boom");}}, 3000);}
}

HashMap存储线程

import java.util.HashMap;
import java.util.Map;
import java.util.Random;public class TestThreadMap {private static HashMap<Thread, Integer> map = new HashMap<Thread, Integer>();public static void main(String[] args) {//TestThreadMap testThreadMap = new TestThreadMap();for (int i = 0; i < 2; i++) {new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubint data = new Random().nextInt();map.put(Thread.currentThread(), data);new A().Get();new B().Get();}}){}.start();}}static class A{public void Get(){int data = map.get(Thread.currentThread());System.out.println("A from " + Thread.currentThread().getName() + " get data " + data);}}static class B{public void Get(){int data = map.get(Thread.currentThread());System.out.println("B from " + Thread.currentThread().getName() + " get data " + data);}}
}

ThreadLocal类似HashMap存储线程

import java.util.HashMap;
import java.util.Map;
import java.util.Random;public class TestThreadMap {private static HashMap<Thread, Integer> map = new HashMap<Thread, Integer>();public static void main(String[] args) {//TestThreadMap testThreadMap = new TestThreadMap();for (int i = 0; i < 2; i++) {new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubint data = new Random().nextInt();map.put(Thread.currentThread(), data);new A().Get();new B().Get();}}){}.start();}}static class A{public void Get(){int data = map.get(Thread.currentThread());System.out.println("A from " + Thread.currentThread().getName() + " get data " + data);}}static class B{public void Get(){int data = map.get(Thread.currentThread());System.out.println("B from " + Thread.currentThread().getName() + " get data " + data);}}
}

import java.util.Random;public class TestThreadlocal_2 {public static void main(String[] args) {// TODO Auto-generated method stubfor (int i = 0; i < 2; i++){new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubint data = new Random().nextInt();MyData.getInstance().setName("myData" + data);MyData.getInstance().setAge(data);new A().Get();new B().Get();}}){}.start();}}static class A{public void Get(){MyData data = MyData.getInstance();System.out.println("A from " + Thread.currentThread().getName() + " get data " + data.getName() + ", " + data.getAge());}}static class B{public void Get(){MyData data = MyData.getInstance();System.out.println("B from " + Thread.currentThread().getName() + " get data " + data.getName() + ", " + data.getAge());}}}class MyData{String name;int age;public static /*synchronized*/ MyData getInstance(){MyData instance = threadLocal.get();if (instance == null){  //不存在就创建一个与本线程有关的实例对象instance = new MyData();threadLocal.set(instance);}return instance;}//private static MyData instance = null;private static ThreadLocal<MyData> threadLocal = new ThreadLocal<MyData>();public String getName() {return name;}public void setName(String name) {this.name = name;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}}

线程池

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class TestThreadPool {public static void main(String[] args){//固定大小的线程池ExecutorService threadPool = Executors.newFixedThreadPool(3);//缓存线程池，当线程不够用时会自动增加，多了会自动减少//ExecutorService threadPool = Executors.newCachedThreadPool();//单一线程池，线程死了可以重新启动//ExecutorService threadPool = Executors.newSingleThreadExecutor();for (int i = 1; i <= 10; i++){final int taskid = i;threadPool.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubfor (int j = 1; j <= 10; j++) {System.out.println(Thread.currentThread().getName() + " is loop of " + j + " for task of " + taskid);}}});}System.out.println("all have finished");threadPool.shutdown(); //线程池里没有任务了，线程池才关闭，等10个任务都完成后才关闭//threadPool.shutdownNow(); //一个线程完成之后立马关闭，此时只完成了3个任务/*//定时器线程池Executors.newScheduledThreadPool(3).schedule(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubSystem.out.println("booming");}}, 6,TimeUnit.SECONDS); //多长时间后执行任务
*/        Executors.newScheduledThreadPool(3).scheduleAtFixedRate( //以固定频率执行任务new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubSystem.out.println("booming");}}, 6,  //初始时间2,  //间隔时间TimeUnit.SECONDS);}
}

锁Lock

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;import javax.xml.stream.events.StartDocument;public class TestLock {public static void main(String[] args) {// TODO Auto-generated method stubTestLock test = new TestLock();test.init();}void init() {final Outputer outputer = new Outputer();new Thread(new Runnable(){@Overridepublic void run() {// TODO Auto-generated method stubwhile(true){try {Thread.sleep(10);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}outputer.Output("wangyuyuyuyuyuyuyuyu");}}}){}.start();new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubwhile(true){try {Thread.sleep(10);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}outputer.Output("zhouzhanzhaozhaozhaozhao");}}}){}.start();}class Outputer{Lock lock = new ReentrantLock(); //锁public void Output(String name){int len = name.length();lock.lock();try {for (int i = 0; i < len; i++){System.out.print(name.charAt(i));}System.out.println("");} finally{// TODO: handle exceptionlock.unlock();}}}}

读写锁ReadWriterLock

import java.util.HashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;public class TestReadWriterLock {HashMap<String, Object> mp = new HashMap<String, Object>();private ReadWriteLock rwl = new ReentrantReadWriteLock();public Object Get(String key){rwl.readLock().lock();Object value = null;try {value = mp.get(key);if (value == null) {rwl.readLock().unlock();rwl.writeLock().lock();try {if (value == null){value = "aaa";}} finally {// TODO: handle finally clauserwl.writeLock().unlock();}rwl.readLock().lock();}} finally {rwl.readLock().unlock();}return value;}
}

信号灯Semaphere(控制当前运行的线程个数)

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.function.IntToDoubleFunction;public class TestSemaphere {public static void main(String[] args) {// TODO Auto-generated method stub/*1.创建线程池*2.创建信号灯，大小为3 *3.循环10次，Runnable里设置信号灯acqure*/ExecutorService es = Executors.newCachedThreadPool();Semaphore semaphore = new Semaphore(3);for(int i = 0; i < 10; i++){es.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubtry {semaphore.acquire();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}System.out.println("线程" + Thread.currentThread().getName() + "进入");try {Thread.sleep((int)Math.random() * 10000);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}System.out.println("线程" + Thread.currentThread().getName() + "即将结束");semaphore.release();System.out.println("线程" + Thread.currentThread().getName() + "已结束");es.shutdown();}});}}}

数据交换Exchanger(当两个数据都到达后才能进行交换，否则阻塞)

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class TestExchanger {public static void main(String[] args) {// TODO Auto-generated method stubExecutorService es = Executors.newCachedThreadPool();Exchanger<String> exchanger = new Exchanger<String>();es.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubtry {String data1 = "x";System.out.println("线程" + Thread.currentThread().getName() + "正在把数据" + data1 + "发送出去");Thread.sleep((int)(Math.random() * 5000));String getData = exchanger.exchange(data1);System.out.println("线程" + Thread.currentThread().getName() + "接受到的数据为: " + getData);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}});es.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubtry {String data1 = "y";System.out.println("线程" + Thread.currentThread().getName() + "正在把数据" + data1 + "发送出去");Thread.sleep((int)(Math.random() * 5000));String getData = exchanger.exchange(data1);System.out.println("线程" + Thread.currentThread().getName() + "接受到的数据为: " + getData);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}});}}

同步屏障CyclicBarrier(多个线程彼此等待，集合后再往后运行)

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class TestCyclicBarrier {//多个线程彼此等待，集合后再运行public static void main(String[] args) {// TODO Auto-generated method stub//创建线程池和CyclicBarrier,同时运行多个线程，调用awaitExecutorService es = Executors.newCachedThreadPool();final CyclicBarrier cb = new CyclicBarrier(3);for (int i = 0; i < 3; i++){es.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubtry {Thread.sleep((int)(Math.random() * 1000));System.out.println("线程" + Thread.currentThread().getName() + "到达集合点1");System.out.println("当前有" + (cb.getNumberWaiting()+1) + "人在等待");cb.await();Thread.sleep((int)(Math.random() * 1000));System.out.println("线程" + Thread.currentThread().getName() + "到达集合点2");System.out.println("当前有" + (cb.getNumberWaiting()+1) + "人在等待");cb.await();Thread.sleep((int)(Math.random() * 1000));System.out.println("线程" + Thread.currentThread().getName() + "到达集合点3");System.out.println("当前有" + (cb.getNumberWaiting()+1) + "人在等待");cb.await();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}});}es.shutdown();}}

CountDownLatch(类似倒计时计数器，当计数减为0时，所有等待者才开始执行)

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class TestCountDownLatch {//类似倒计时计数器，当计数减为0的时候，所有等待者才开始执行public static void main(String[] args) {// TODO Auto-generated method stub//创建两个计数器，一个为1，一个为3，代表一个裁判员，三个运动员，裁判员在未下达命令前运动员等待，下达命令后才执行//等三个运动员都到达终点后，裁判员才公布成绩ExecutorService es = Executors.newCachedThreadPool();CountDownLatch cdOrder = new CountDownLatch(1);CountDownLatch cdAnswer = new CountDownLatch(3);for (int i = 0; i < 3; i++) {final int id = i;es.execute(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubtry {System.out.println("运动员" + id + "正准备接受命令");cdOrder.await();System.out.println("运动员"+ id + "接受到命令");Thread.sleep((int)(Math.random() * 5000));System.out.println("运动员" + id + "到达终点");cdAnswer.countDown();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}});}try {Thread.sleep(1000);System.out.println("裁判员发出指令");cdOrder.countDown();System.out.println("裁判员等待所有运动员到达终点");cdAnswer.await();System.out.println("裁判员公布成绩");} catch (Exception e) {// TODO: handle exception}}}

Callable和Future

import java.util.Random;
import java.util.concurrent.Callable;
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.TimeUnit;
import java.util.concurrent.TimeoutException;public class TestCallableAndFuture {public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException{ExecutorService threadPool = Executors.newSingleThreadExecutor();Future future =threadPool.submit(new Callable<String>() {@Overridepublic String call() throws Exception {// TODO Auto-generated method stubThread.sleep(2000);return "hello";}});System.out.println("得到结果: " + future.get()); //callable完成任务返回结果，由future去拿，需要等待一段时间//System.out.println("得到结果: " + future.get(1, TimeUnit.SECONDS)); //要在规定的时间内得到结果，如果得不到就抛出异常//CompletionService 用于提交一组callable任务，并用take方法得到一个已完成任务的future对象ExecutorService threadPool2 = Executors.newFixedThreadPool(10);CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(threadPool2);for (int i = 1; i <= 10; i++){final int taskid = i;completionService.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {// TODO Auto-generated method stubtry {Thread.sleep(new Random().nextInt(5000));} catch (Exception e) {// TODO: handle exception}return taskid;}});}for (int i = 1; i <= 10; i++){System.out.println(completionService.take().get());}}}

阻塞队列ArrayBlockingQueue

import java.util.concurrent.ArrayBlockingQueue;public class TestArrayBlockingQueue {//阻塞队列当队列为空时take会阻塞，当队列满时put会阻塞//用两个阻塞队列模拟两个线程交替运行//两个阻塞队列大小均设置为1，其中一个放一个数据public static void main(String[] args) {// TODO Auto-generated method stubBusiness business = new Business();new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubfor (int j = 1; j <= 10; j++) {business.Sub(j);}}}){}.start();for (int j = 1; j <= 10; j++) {business.Main(j);}}static class Business{ArrayBlockingQueue abq1 = new ArrayBlockingQueue(1);ArrayBlockingQueue abq2 = new ArrayBlockingQueue(1);public Business() {try {abq2.put(1);} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}public void Sub(int j){try {abq1.put(1);for (int i = 1; i <= 10; i++){System.out.println("sub thread " + i + " of loop " + j);}abq2.take();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}public void Main(int j){try {abq2.put(1);for (int i = 1; i <= 10; i++){System.out.println("main thread " + i + " of loop " + j);}abq1.take();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}}}

练习: 子线程运行10次，主线程运行20次，交替运行

public class prictice_1 {//子线程运行10次，主线程运行20次public static void main(String[] args) {// TODO Auto-generated method stubBusiness business = new Business();new Thread(new Runnable() {@Overridepublic void run() {// TODO Auto-generated method stubfor (int j = 1; j <= 10; j++) {business.Sub(j);}}}){}.start();for (int j = 1; j <= 10; j++) {business.Main(j);}}}
class Business{private boolean isSub = true;public synchronized void Sub(int j){while (!isSub) {try {this.wait();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}for (int i = 1; i <= 10; i++){System.out.println("sub thread " + i + " of loop " + j);}isSub = false;this.notify();}public synchronized void Main(int j){while (isSub){try {this.wait();} catch (Exception e) {// TODO: handle exceptione.printStackTrace();}}for (int i = 1; i <= 20; i++){System.out.println("main thread " + i + " of loop " + j);}isSub = true;this.notify();}
}