java多线程那点事儿

栏目: Java · 发布时间: 6年前

内容简介：前段时间应隔壁部门大佬的邀约，简单地帮他们部门的童靴梳理了下多线程相关的内容，客串了一把讲师【因为部门内有不少是c#转java的童鞋，所以讲的稍微浅显了些】ok，按照个人习惯先来大纲知识点：

前段时间应隔壁部门大佬的邀约，简单地帮他们部门的童靴梳理了下多线程相关的内容，客串了一把讲师【因为部门内有不少是c#转 java 的童鞋，所以讲的稍微浅显了些】

ok，按照个人习惯先来大纲

知识点：

1）进程多线程的相关概念涉及到CPU调度稍微谈下JVM内存模型程序计数器

2）多线程的三种实现手段及其代码实现这边可以谈下futurtask的底层源码

3）常用锁概念及实现说明隐式锁显式锁乐观锁悲观锁 CAS 可重入锁不可重入锁读写锁线程安全引申的谈下分布式锁及分布式锁的原理，常用的三种实现手段 volatile关键字及其底层源码

4）线程池的概念，线程池的使用扒一扒线程池的源码缓存队列核心线程池线程池创建任务的过程线程池的生命周期等

多线程：多线程是什么？多线程是一个程序（进程）运行时产生了不止一个线程。

进程和线程区别一个正在执行的程序，进程是控制程序的执行顺序。这个顺序又被称为一个控制单元。

并行和并发的概念：并行：多个CPU实例或者多台机器同时执行一段处理逻辑并发：通过CPU调度算法，让用户看上去是同时执行的，在CPU层面不是同时。

这边衍生的可以谈下JVM中内存模型的程序计数器就是记录java执行字节码的行号指示器。

jvm内存模型：线程私有：

程序计数器：记录程序执行过程中的字节码的行号指示器

java虚拟机栈: 主要是是被调用的java方法代表的是一个个栈帧局部变量表操作数栈动态链接方法出口等等 java.lang.StackOverflowError

本地方法栈：主要是是被调用的native方法代表的是一个个栈帧

线程公有

堆 : 对象实例

方法区：最重要的就是运行时常量池 gc

为什么要是用多线程：

1）充分的利用CPU资源，如果只有一个线程的话，第二个任务必须等第一个任务完成之后才能进行。

2）进程之间无法共享数据，但是线程可以

3）创建进程需要为这个进程分配系统资源，创建线程的代价小

多线程的实现手段【3种手段】

1）Thread

package com.Allen.test;

import java.util.concurrent.TimeUnit;

public class testThread extends Thread{
	public static void main(String[] args) {
		testThread t1=new testThread();
		testThread t2=new testThread();
		t1.start();
		t2.start();
	}
	
	
	 public void run(){
		 System.out.println("start");
		 try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		 System.out.println("end");
	}
}
复制代码

2）runnable

package com.Allen.test;

import java.util.concurrent.TimeUnit;

public class testRunnable {
	public static void main(String[] args) {
		testAllen th1=new testAllen();
		for(int i=0;i<5;i++){
			Thread t1=new Thread(th1);
			t1.start();
		}
	}
}

class testAllen implements Runnable{

	@Override
	public void run() {
		 System.out.println("start");
		 try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		 System.out.println("end");
	}	
}
复制代码

3）future callable

package com.Allen.test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

public class testFuture {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		futureTask task=new futureTask();
//		//线程池  单线程的线程池
//		ExecutorService service=Executors.newCachedThreadPool();
//		Future<Integer> future=service.submit(task);
//		//说到下面这个方法就要说起线程池的状态  四个种状态
//		service.shutdown();
		
		FutureTask<Integer>future=new FutureTask<>(task);
		Thread t1=new Thread(future);
		t1.start();
		System.out.println("run task ....");
		TimeUnit.SECONDS.sleep(1);
		System.out.println("result : "+future.get());
	}
}

class futureTask implements Callable<Integer>{

	@Override
	public Integer call() throws Exception {
		TimeUnit.SECONDS.sleep(2);
		int result=0;
		//模拟一个庞大的计算
		for(int i=0;i<100;i++){
			for(int j=0;j<i;j++){
				result+=j;
			}
		}
		return result;
	}
	
}
复制代码

扒一扒futureTask的源码

首先我们看run方法

1）run方法中state不是new或者线程再次给他设置启动出错直接让他return掉

2）完后方法后调用set方法把结果赋值给outcome

这边用了CAS方式来设置stat状态

拿到栈顶的元素一个个去唤醒

报错的话返回异常信息

接下来我们谈下多线程操作中比较核心的东西

多线程操作共享变量的问题

锁机制

最常用的锁机制 synchronized及lock

隐式锁 synchronized 谈下底层原理

显示锁 lock

重入锁

我设计了一个实例来直观的观察多线程操作共享变量的线程安全问题

package com.Allen.test;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class TestSynchronized {
	public static void main(String[] args) throws InterruptedException {
		for (int s = 0; s < 10; s++) {
			Person person = new Person();
			person.setAge(0);
			ReentrantLock lock = new ReentrantLock();
			for (int i = 0; i < 100; i++) {
				Thread t1 = new Thread(new test111(person,lock));
				t1.start();
			}
			TimeUnit.SECONDS.sleep(3);
			System.out.println(person.getAge());
		}
	}
}

class test111 implements Runnable {
	Person person;
	ReentrantLock lock;

	public test111(Person person,ReentrantLock lock) {
		this.person = person;
		this.lock=lock;
	}

	public void run() {
//		synchronized (person) {
//			person.setAge(person.getAge() + 1);
//		}
		
		lock.lock();
		person.setAge(person.getAge() + 1);
		lock.unlock();
		 //person.setAge(person.getAge()+1);
		 
	}
}
复制代码

悲观锁乐观锁 CAS

一悲观锁

在关系型数据库管理系统中，悲观并发控制（悲观锁）是一种并发控制的方法。简单而言，就是它“悲观”地默认每次拿数据都认为别人会修改，所以在每次拿之前去上锁，这样别人想拿这个数据就会block直到它拿到锁。传统悲观锁实现机制大多利用数据库提供的锁机制（也只有数据库层面提供的锁机制才能真正保证了数据访问的排他性）。

悲观锁流程如下：

1 对任意记录进行修改之前，尝试给它加排它锁。

2 若是加锁失败，说明该记录正在修改，那么当前需要等待或者抛出异常。

3 如果成功加锁，那么就可以对记录进行修改，事务完成之后解锁。

4 期间其他人需要对该记录进行修改或者加排查锁操作，就不许等待我们解锁或者直接抛出异常。

以 mysql 为例

使用悲观锁，先关闭mysql的自动提交属性。

set autocommit=0

begin；

select status from t_goods where id=1 for update;

insert into t_orders(id,goods_id)values(null,1);

update t_goods set status=2;

commit;

发起事务操作提交事务

select for update 开启排它锁方式实现悲观锁，mysql InnoDB默认行级锁【ps：注意一点，行级锁都是基于索引，如果用不到索引会使用表级锁吧整张表锁住】

优点与缺点：

悲观并发控制实现上是“先取锁再访问”的保守策略，为数据安全提供保障，但是牺牲了效率，处理加锁会让数据库产生额外的开销，还增加了死锁的机会，降低了并行性，如果一个实物锁定了某行数据，其他事物必须等待改事务处理完才能处理那一行。适用于状态修改非常高，冲突非常严重的系统。

二乐观锁

假设了多用户并发事务处理下不会彼此影响，各事务在不产生锁的情况下处理各自的那部分数据，

每次去拿数据都认为别人不会修改，所以不会上锁，只会在更新的时候判断一下此期间别人有没有更新这个数据。如果有其他事务更新的话，正在提交的事务会回滚。

一般来说乐观锁不会使用数据库提供的机制，我们通常采用记录数据版本来实现乐观锁记录的方式有版本号或者时间戳。

在数据初始化的时候指定一个版本号，每次对数据更新在对版本号做+1操作，并判断当前的版本号是不是该数据的最新版本。

优点与缺点；

乐观并发控制事务之间数据竞争概率较小，可以一直做下去知道提交才会锁定，有点类似于svn

当俩个事务都读到某一行修改回传数据库就会遇到问题。

三 CAS 无锁化编程

java.util.concurrent包中借助CAS实现了区别于synchronouse同步锁的一种乐观锁。无锁化编程

CAS有三个操作数，内存值V，旧的预期值A，修改的新值B。当且仅当预期值A与内存值V相同时，将内存值V修改为B，否则什么都不做。

优点与缺点：可以用CAS在无锁的情况下实现原子操作，但要明确应用场合，非常简单的操作且又不想引入锁可以考虑使用CAS操作，当想要非阻塞地完成某一操作也可以考虑CAS。 CAS虽然很高效的解决原子操作，但是CAS仍然存在三大问题。ABA问题，循环时间长开销大和只能保证一个共享变量的原子操作

ABA问题。因为CAS需要在操作值的时候检查下值有没有发生变化，如果没有发生变化则更新，但是如果一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。ABA问题的解决思路就是使用版本号。在变量前面追加上版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A-2B－3A。从Java1.5开始JDK的atomic包里提供了一个类AtomicStampedReference来解决ABA问题。这个类的compareAndSet方法作用是首先检查当前引用是否等于预期引用，并且当前标志是否等于预期标志，如果全部相等，则以原子方式将该引用和该标志的值设置为给定的更新值。关于ABA问题参考文档: blog.hesey.net/2011/09/res…
循环时间长开销大。自旋CAS如果长时间不成功，会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升，pause指令有两个作用，第一它可以延迟流水线执行指令（de-pipeline）,使CPU不会消耗过多的执行资源，延迟的时间取决于具体实现的版本，在一些处理器上延迟时间是零。第二它可以避免在退出循环的时候因内存顺序冲突（memory order violation）而引起CPU流水线被清空（CPU pipeline flush），从而提高CPU的执行效率。
只能保证一个共享变量的原子操作。当对一个共享变量执行操作时，我们可以使用循环CAS的方式来保证原子操作，但是对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁，或者有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量i＝2,j=a，合并一下ij=2a，然后用CAS来操作ij。从Java1.5开始JDK提供了AtomicReference类来保证引用对象之间的原子性，你可以把多个变量放在一个对象里来进行CAS操作。

compareAndSet有点类似于如下

if (this == expect) {

this = update

return true;

} else {

return false;

}

如下附上CAS实例

volatile关键字但是我们要思考下并发编程下的俩个关键问题？

1 线程之间是如何通信

1.1 共享内存

隐式通信

1.2 消息传递

显式通信

2 线程之间是如何同步

在共享内存的并发模型中，同步是显示做的，synchronized

在消息传递的并发模型，由于消息的发送必须要在消息的接受之前，所以同步是隐式的。

2 定位内存可见性问题：

什么对象是内存共享的，什么不是。

主内存：共享变量

私有本地内存：存储共享变量的副本

可见性原理

1 volatile声明的变量进行写操作的时候，jvm会向处理器发送一条lock前缀的指令，会把这个变量所在缓存行的数据回到系统内存。

2 在多处理器的情况下，保证各个处理器缓存一致性的特点，就会实现缓存一致性协议。

synchronized ：可重入锁，互斥性，可见性。

volatile：原子性，可见性。不能做到复合操作的原子性。性能开销更小。

synchronized 线程A释放锁之后会把本地内存的变量同步到主内存

线程B获取锁的时候会把主内存的共享变量同步到本地内存中。

调用monitorenter monitorexit

对象同步方法调用的时候必须要获取到它的监视器，用完之后会释放。

获取不到监听器会放在队列中。

多进程下访问共享变量

分布式锁包括三种常用的使用手段

1 ）通过数据库的方式

create table lock(

Method_Name 唯一约束

)

每次去操作文件的是否，都去插入表，获取锁是否才能去操作文件，

等锁释放【删除这个记录】才能insert

缺点：

删除失败【等待程序不可用】

重入锁【可以对进程进行编号，来判断重入】

2）使用zookeeper

临时有序节点

谁先写到节点上，获取锁

有个watch机制，会判断节点是否失效，失效之后会读取下一个节点

3） redis

setnx

谁先set这个值，谁就现获取锁

后续set失败的就是没有获取锁

等待锁释放之后你才能set这个值。

线程池

线程池种类及区别：

运行executors类给我们提供静态方法调用不同的线程池

newsingleThreadExecutor：

返回一个单线程的executor，将多个任务交给这个Executor时，这个线程处理完一个任务之后接着处理下一个任务，若该线程出现异常，将会有一个新的线程替代。

newFixedThreadPool

返回一个包含指定数目线程的线程池，任务数量多于线程数目，则没有执行的任务必须等待，直到任务完成。

newCahedThreadPool

根据用户的任务数创建相应的线程数来处理，该线程池不会对线程数加以限制，完全依赖于JVM能创建的线程数量，可能引起内存不足。

用法如下：

package com.Allen.TestPoolSize;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Test {
	public static void main(String[] args) {
		//单线程化的线程池
		ExecutorService service1=Executors.newSingleThreadExecutor();
		service1.execute(new Runnable() {
			@Override
			public void run() {
				System.out.println("aaa");
			}
		});
		//可缓存线程池
		ExecutorService service2=Executors.newCachedThreadPool();
		service2.execute(new Runnable() {
			public void run() {
				System.out.println("bbb");
			}
		});
		//定长线程池
		ExecutorService service3=Executors.newFixedThreadPool(3);
		service3.execute(new Runnable() {
			public void run() {
				System.out.println("ccc");
			}
		});
		//定长线程池支持定时和周期任务
		ScheduledExecutorService service4=Executors.newScheduledThreadPool(5);
		service4.schedule(new Runnable() {
			@Override
			public void run() {
				System.out.println("ddd");	
			}
		}, 3, TimeUnit.SECONDS);	
	}
}
复制代码

过程原理

扒一扒源码

首先我们写一个test类

package com.Allen.studyThread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class test {
	public static void main(String[] args) {
		ExecutorService threadpool=Executors.newFixedThreadPool(3);
		for(int i=0;i<10;i++){
			threadpool.submit(new testrunnable("allen_"+i));
		}
		threadpool.shutdown();
	}
}

class testrunnable implements Runnable{
	
	private String name;
	
	public testrunnable(String name){
		this.name=name;
	}
	
	public void run() {
		System.out.println(name+" start...");
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println(name+" end...");
	} 
}
复制代码

看线程池源码，我们入口从newFixedThreadPool入口入，我们看看他初始化做了什么进入到Executors类的newFixedThreadPool方法传入一个参数