Java基础-多线程
多个线程一起做同一件事情,缩短时间,提升效率
提高资源利用率
加快程序响应，提升用户体验
创建线程
1. 继承Thread类
-
步骤
-
继承Thread类，重写run方法
-
调用的时候，直接new一个对象，然后调start()方法启动线程
-
-
特点
-
由于是继承方式，所以不建议使用，因为Java是单继承的，不够灵活
-
Thread类本质也是实现Runnable接口（public class Thread implements Runnable）
-
2. 实现Runnable接口
-
步骤
-
实现Runnable接口，重写run()方法
-
创建Runnable实现类的实例，并用这个实例作为Thread的target来创建Thread对象
-
调用Thread类实例对象的start()方法启动线程
-
-
特点
-
只是实现，保留了继承的其他类的能力
-
如果需要访问当前线程，必须使用Thread.currentThread()方法
-
3. 实现 Callable接口
-
步骤
-
实现Callable接口，重写call()方法
-
创建Callable实现类的实例，使用FutureTask类来包装Callable对象
-
并用FutureTask实例作为Thread的target来创建Thread对象
-
调用Thread类实例对象的start()方法启动线程
-
调用FutureTask类实例对象的get()方法获取异步返回值
-
-
特点
-
call方法可以抛出异常
-
只是实现，保留了继承的其他类的能力
-
如果需要访问当前线程，必须使用Thread.currentThread()方法
-
通过FutureTask对象可以了解任务执行情况，可取消任务的执行，还可获取执行结果
-
4. 匿名内部类实现
-
说明
-
本质还是前面的方法，只是使用了匿名内部类来实现，简化代码
-
Callable接口之所以把FutureTask类的实例化写出来，是因为需要通过task对象获取返回值
-
参数传递
1. 通过构造方法传递数据
通过前面的学习，可以看到，不管何种创建对象的方式，都需要新建立实例，所以我们可以通过构造函数传入参数，并将传入的数据使用类变量保存起来
-
特点
-
在线程运行之前，数据就已经传入了
-
使用构造参数，当参数较多时，使用不方便
-
不同参数条件需要不同的构造方法，使得构造方法较多
-
2. 通过变量和方法传递数据
在线程类里面定义一些列的变量，然后定义set方法，在新建实例之后，调用set方法传递参数
-
特点
- 在参数较多时使用方便，按需传递参数
3. 通过回调函数传递数据
使用线程方法自己产生的变量值作为参数，去调取外部的方法，获取返回数据的方式
-
特点
- 拥有获取数据的主动权
线程同步
要跨线程维护正确的可见性，只要在几个线程之间共享非 final 变量，就必须使用线程同步
1. ThreadLocal
ThreadLocal利用空间换时间，通过为每个线程提供一个独立的变量副本，避免了资源等待，解决了变量并发访问的冲突问题，提高了并发量。实现了线程间的数据隔离，但是线程间无法共享同一个资源
public class StudyThread {
public static void main(String[] args) throws ExecutionException, InterruptedException {
SyncTest syncTest = new SyncTest();
ConcurrentHashMap<String, String> testConMap = new ConcurrentHashMap<>();
for (int i = 0; i < 10; i++) {
ThreadTest2 threadTest2 = new ThreadTest2();
threadTest2.setSyncTest(syncTest);
Thread threadTest = new Thread(threadTest2);
threadTest.start();
}
}
}
//实现Runnable
class ThreadTest2 implements Runnable {
private SyncTest syncTest;
public void setSyncTest(SyncTest syncTest) {
this.syncTest = syncTest;
}
@Override
public void run() {
syncTest.threadLocalTest(Thread.currentThread().getName());
}
}
class SyncTest {
private static ThreadLocal<String> threadLocal = new ThreadLocal<String>();
public void threadLocalTest(String name) {
try {
System.out.println(name + "进入了threadLocal方法!");
threadLocal.set(name);
Thread.currentThread().sleep(100);
System.out.println(threadLocal.get() + "离开了threadLocal方法!");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2. synchronized
不管synchronized是用来修饰方法，还是修饰代码块，其本质都是锁定某一个对象。修饰方法时，锁上的是调用这个方法的对象，即this；修饰代码块时，锁上的是括号里的那个对象。每一个Java对象都有一个内置锁，访问synchronized代码块或synchronized方法的时候，线程都需要首先获取到对象关联的内置锁，对于static方法，线程获取的是类对象的内置锁。
-
特点
- 锁的对象越小越好
public class StudyThread {
public static void main(String[] args) throws ExecutionException, InterruptedException {
SyncTest syncTest = new SyncTest();
ConcurrentHashMap<String, String> testConMap = new ConcurrentHashMap<>();
for (int i = 0; i < 10; i++) {
ThreadTest2 threadTest2 = new ThreadTest2();
threadTest2.setSyncTest(syncTest);
threadTest2.setTestConMap(testConMap);
Thread threadTest = new Thread(threadTest2);
threadTest.start();
}
}
}
//实现Runnable
class ThreadTest2 implements Runnable {
private ConcurrentHashMap<String, String> testConMap;
private SyncTest syncTest;
public void setTestConMap(ConcurrentHashMap<String, String> testConMap) {
this.testConMap = testConMap;
}
public void setSyncTest(SyncTest syncTest) {
this.syncTest = syncTest;
}
@Override
public void run() {
//三个方法需要单独测试，因为testConMap会相互影响
//测试同步方法，锁住的对象是syncTest
//syncTest.testSyncMethod(testConMap,Thread.currentThread().getName());
//测试同步代码块，锁住的对象是testConMap
//syncTest.testSyncObject(testConMap, Thread.currentThread().getName());
//测试没有锁时执行请求是多么的混乱！！！
//syncTest.testNoneSyncObject(testConMap, Thread.currentThread().getName());
}
}
//同步测试方法类
class SyncTest {
public synchronized void testSyncMethod(ConcurrentHashMap<String, String> testConMap, String name) {
try {
System.out.println(name + "进入了同步方法！");
testConMap.put("name", name);
Thread.currentThread().sleep(10);
System.out.println(testConMap.get("name") + "离开了同步方法！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void testSyncObject(ConcurrentHashMap<String, String> testConMap, String name) {
synchronized (testConMap) {
try {
System.out.println(name + "进入了同步代码块！");
testConMap.put("name", name);
Thread.currentThread().sleep(10);
System.out.println(testConMap.get("name") + "离开了同步代码块！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void testNoneSyncObject(ConcurrentHashMap<String, String> testConMap, String name) {
try {
System.out.println(name + "进入了无人管辖区域！");
testConMap.put("name", name);
Thread.currentThread().sleep(10);
System.out.println(testConMap.get("name") + "离开了无人管辖区域！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
3. volatile
-
特点
-
保证可见性，有序性，不保证原子性
-
它会强制将对缓存的修改操作立即写入主存
-
volatile不适合复合操作(对变量的写操作不依赖于当前值)，否则需要保证只有单一线程能够修改变量的值
-
使用volatile关键字，可以禁止指令重排序(单例双重检查锁)
-
public class StudyThread {
static int v = 1;//volatile能够保证变量的可见性
public static void main(String[] args) {
//改动线程
new Thread(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
v++;//确保只有一个线程修改变量值
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
//检测线程
new Thread(new Runnable() {
@Override
public void run() {
int old = 0;
while (old < 11) {
if (old != v) {
old = v;
System.out.println("检测线程：v的值变动为" + old);
}
}
}
}).start();
}
}
4. ReentrantLock
- 说明
-
目前ReentrantLock和synchronized性能上没有什么差别
-
ReentrantLock需要手动加锁和解锁,且解锁的操作尽量要放在finally代码块中,保证线程正确释放锁
-
ReentrantLock可以实现公平锁，在锁上等待时间最长的线程将获得锁的使用权，性能没有非公平锁性能好
-
ReentrantLock提供了一个可以响应中断的获取锁的方法
lockInterruptibly(),可以用来解决死锁问题 -
ReentrantLock还提供了获取锁限时等待的方法
tryLock(),使用该方法配合失败重试机制来更好的解决死锁问题 -
ReentrantLock结合Condition接口可以实现等待通知机制
-
public class StudyThread {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
//三段代码逐一测试
//测试tryLock
Thread threadTest00 = new Thread(new Runnable() {
@Override
public void run() {
try {
while(!lock.tryLock()){
System.out.println(Thread.currentThread().getName() + "没有拿到锁，继续等待！");
Thread.sleep(50);
}
System.out.println(Thread.currentThread().getName() + "进入了lock代码块！");
Thread.currentThread().sleep(300);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest00.start();
Thread threadTest01 = new Thread(new Runnable() {
@Override
public void run() {
try {
while(!lock.tryLock()){
System.out.println(Thread.currentThread().getName() + "没有拿到锁，继续等待！");
Thread.sleep(50);
}
System.out.println(Thread.currentThread().getName() + "进入了lock代码块！");
Thread.currentThread().sleep(300);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest01.start();
//测试中断锁
Thread threadTest02 = new Thread(new Runnable() {
@Override
public void run() {
try {
lock.lockInterruptibly();
System.out.println(Thread.currentThread().getName() + "进入了lock代码块！");
Thread.currentThread().sleep(2000);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest02.start();
Thread threadTest03 = new Thread(new Runnable() {
@Override
public void run() {
try {
lock.lockInterruptibly();
System.out.println(Thread.currentThread().getName() + "进入了lock代码块！");
Thread.currentThread().sleep(2000);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest03.start();
Thread.currentThread().sleep(20);
threadTest02.interrupt();
//测试condition
Thread threadTest04 = new Thread(new Runnable() {
@Override
public void run() {
try {
lock.lock();
System.out.println(Thread.currentThread().getName() + "进入了lock代码块，等待通知！");
condition.await();
System.out.println(Thread.currentThread().getName() + "收到通知，继续执行！");
Thread.currentThread().sleep(1000);
} catch (Exception e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest04.start();
Thread threadTest05 = new Thread(new Runnable() {
@Override
public void run() {
try {
lock.lock();
System.out.println(Thread.currentThread().getName() + "进入了lock代码块！");
Thread.currentThread().sleep(1000);
condition.signal();
System.out.println(Thread.currentThread().getName() + "发出通知！");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "准备释放锁，并离开lock代码块！");
lock.unlock();
}
}
});
threadTest05.start();
}
}
5. 线程安全的类
Java中很多类说的线程安全指的是，它的每个方法单独调用（即原子操作）都是线程安全的，但是代码总体的互斥性并不受控制
-
线程安全的类有以下几类
-
Concurrentxxx
-
ThreadPoolExecutor
-
BlockingQueue和BlockingDeque
-
原子类Atomicxxx—包装类的线程安全类
-
CopyOnWriteArrayList和CopyOnWriteArraySet
-
通过synchronized 关键字给方法加上内置锁来实现线程安全：Timer，TimerTask，Vector，Stack，HashTable，StringBuffer
-
Collections中的synchronizedCollection(Collection c)方法可将一个集合变为线程安全：
Map m=Collections.synchronizedMap(new HashMap());
-
线程池
线程池只能放入实现Runable或callable类线程，不能直接放入继承Thread的类
1. Executors线程池的实现
-
要点
-
可能导致资源耗尽，OOM问题出现
-
线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式(阿里巴巴java开发)
-
public class StudyThread {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//创建一个线程池，该线程池重用固定数量的从共享无界队列中运行的线程
//ExecutorService threadPool = Executors.newFixedThreadPool(20);
//创建一个维护足够的线程以支持给定的并行级别的线程池,线程的实际数量可以动态增长和收缩,工作窃取池不保证执行提交的任务的顺序
//ExecutorService threadPool = Executors.newWorkStealingPool(8);
//创建一个使用从无界队列运行的单个工作线程的执行程序。
//ExecutorService threadPool = Executors.newSingleThreadExecutor();
//创建一个根据需要创建新线程的线程池,但在可用时将重新使用以前构造的线程。如果没有可用的线程，将创建一个新的线程并将其添加到该池中。未使用六十秒的线程将被终止并从缓存中删除
ExecutorService threadPool = Executors.newCachedThreadPool();
//放入Runnable类线程
for (int i = 0; i < 10; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println("线程名：" + Thread.currentThread().getName());
}
});
}
//Thread.currentThread().sleep(1000);
//放入Callable类线程
List<Future<String>> futures = new ArrayList<>();
for (int i = 0; i < 10; i++) {
Future<String> future = threadPool.submit(new Callable<String>() {
@Override
public String call() throws Exception {
System.out.println("线程名：" + Thread.currentThread().getName());
return Thread.currentThread().getName();
}
});
futures.add(future);
}
threadPool.shutdown();
for (Future future : futures) {
System.out.println(future.get());
}
}
}
2. ThreadPoolExecutor创建线程池
-
要点
-
线程池空闲大小和最大线程数根据实际情况确定
-
keepAliveTime一般设置为0
-
unit一般设置为TimeUnit.SECONDS(其他的也行，反正是0)
-
任务队列需要指定大小，不要使用无界队列，容易造成OOM-> new ArrayBlockingQueue<>(512)
-
ThreadFactory threadFactory使用系统默认的
-
拒绝策略：
-
AbortPolicy：抛出RejectedExecutionException（该异常是非受检异常，要记得捕获）
-
DiscardPolicy：什么也不做，直接忽略
-
DiscardOldestPolicy：丢弃执行队列中最老的任务，尝试为当前提交的任务腾出位置
-
CallerRunsPolicy：直接由提交任务者执行这个任务
-
-
public class StudyThread {
public static void main(String[] args) throws ExecutionException, InterruptedException {
int poolSize = Runtime.getRuntime().availableProcessors() * 2;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(512);
RejectedExecutionHandler policy = new ThreadPoolExecutor.DiscardPolicy();
ExecutorService executorService = new ThreadPoolExecutor(poolSize, poolSize,
0, TimeUnit.SECONDS,
queue,
policy);
//放入Runnable类线程
for (int i = 0; i < 10; i++) {
executorService.execute(new Runnable() {
@Override
public void run() {
System.out.println("线程名：" + Thread.currentThread().getName());
}
});
}
//放入Callable类线程
List<Future<String>> futures = new ArrayList<>();
for (int i = 0; i < 10; i++) {
Future<String> future = executorService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
System.out.println("线程名：" + Thread.currentThread().getName());
return Thread.currentThread().getName();
}
});
futures.add(future);
}
for (Future future:futures) {
System.out.println(future.get());
}
//放入Callable类线程
//使用CompletionService简化获取结果的操作，执行完一个任务，获取一个结果，结果顺序和执行顺序相同
CompletionService<String> ecs = new ExecutorCompletionService<String>(executorService);
for (int i = 0; i < 10; i++) {
Future<String> future = ecs.submit(new Callable<String>() {
@Override
public String call() throws Exception {
System.out.println("线程名：" + Thread.currentThread().getName());
return Thread.currentThread().getName();
}
});
}
for (int i = 0; i < 10; i++) {
System.out.println(ecs.take().get());
}
}
}