一个进程往往可以包含多个线程,至少包含一个! Java默认有几个线程? 2 个: mian、GC 对于Java而言:Thread、Runnable、Callable 三种实现线程的方式。 JAVA不能开启线程,是调用本地方法,查看start方法可以看到底层是C++来开启线程的
并发编程的本质:充分利用CPU的资源
线程的6种状态
COPYpublic enum State {
// 新生
NEW,
// 运行
RUNNABLE,
// 阻塞
BLOCKED,
// 等待,死死地等
WAITING,
// 超时等待
TIMED_WAITING,
// 终止
TERMINATED;
}
wait和sleep的区别
synchronized方式上锁
COPY/**
* 真正的多线程开发,公司中的开发,降低耦合性
* 线程就是一个单独的资源类,没有任何附属的操作!
* 1、 属性、方法
*/
public class SaleTicketDemo01 {
public static void main(String[] args) {
// 并发:多线程操作同一个资源类, 把资源类丢入线程
Ticket ticket = new Ticket();
// @FunctionalInterface 函数式接口,jdk1.8 lambda表达式 (参数)->{ 代码 }
new Thread(()->{
for (int i = ; i < ; i++) {
ticket.sale();
}
},"A").start();
new Thread(()->{
for (int i = ; i < ; i++) {
ticket.sale();
}
},"B").start();
new Thread(()->{
for (int i = ; i < ; i++) {
ticket.sale();
}
},"C").start();
}
}
// 资源类 OOP
class Ticket {
// 属性、方法
private int number = ;
// 卖票的方式
// synchronized 本质: 队列,锁
public synchronized void sale(){
if (number>){
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余:"+number);
}
}
}
lock方式上锁
COPYpublic class SaleTicketDemo02 {
public static void main(String[] args) {
// 并发:多线程操作同一个资源类, 把资源类丢入线程
Ticket2 ticket = new Ticket2();
// @FunctionalInterface 函数式接口,jdk1.8 lambda表达式 (参数)->{ 代码 }
new Thread(()->{for (int i = ; i < ; i++) ticket.sale();},"A").start();
new Thread(()->{for (int i = ; i < ; i++) ticket.sale();},"B").start();
new Thread(()->{for (int i = ; i < ; i++) ticket.sale();},"C").start();
}
}
// Lock三部曲
// 1、 new ReentrantLock();
// 2、 lock.lock(); // 加锁
// 3、 finally=> lock.unlock(); // 解锁
class Ticket2 {
// 属性、方法
private int number = ;
Lock lock = new ReentrantLock();
public void sale(){
lock.lock(); // 加锁
try {
// 业务代码
if (number>){
System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余:"+number);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock(); // 解锁
}
}
}
两者的区别
await 和 notify的实现方式
COPYpublic class B {
public static void main(String[] args) {
Data2 data = new Data2();
new Thread(()->{
for (int i = ; i < ; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"A").start();
new Thread(()->{
for (int i = ; i < ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"B").start();
new Thread(()->{
for (int i = ; i < ; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"C").start();
new Thread(()->{
for (int i = ; i < ; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"D").start();
}
}
// 判断等待,业务,通知
class Data2{ // 数字 资源类
private int number = ;
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
//condition.await(); // 等待
//condition.signalAll(); // 唤醒全部
//+1
public void increment() throws InterruptedException {
lock.lock();
try {
// 业务代码
while (number!=){ //0
// 等待
condition.await();
}
number++;
System.out.println(Thread.currentThread().getName()+"=>"+number);
// 通知其他线程,我+1完毕了
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
//-1
public synchronized void decrement() throws InterruptedException {
lock.lock();
try {
while (number==){ // 1
// 等待
condition.await();
}
number--;
System.out.println(Thread.currentThread().getName()+"=>"+number);
// 通知其他线程,我-1完毕了
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
用JUC实现
COPYpublic class C {
public static void main(String[] args) {
Data3 data = new Data3();
new Thread(()->{
for (int i = ; i < ; i++) {
data.printA();
}
},"A").start();
new Thread(()->{
for (int i = ; i < ; i++) {
data.printB();
}
},"B").start();
new Thread(()->{
for (int i = ; i < ; i++) {
data.printC();
}
},"C").start();
}
}
class Data3{ // 资源类 Lock
//几个线程就用几个Condition 精准的通知和唤醒线程
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
private int number = ; // 1A 2B 3C
public void printA(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
while (number!=){
// 等待
condition1.await();
}
System.out.println(Thread.currentThread().getName()+"=>AAAAAAA");
// 唤醒,唤醒指定的人,B
number = ;
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printB(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
while (number!=){
condition2.await();
}
System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
// 唤醒,唤醒指定的人,c
number = ;
condition3.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void printC(){
lock.lock();
try {
// 业务,判断-> 执行-> 通知
// 业务,判断-> 执行-> 通知
while (number!=){
condition3.await();
}
System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
// 唤醒,唤醒指定的人,c
number = ;
condition1.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
原生的List,Map,Set的实现类方法在多线程会出现不同步的问题,所以有以下解决的方案:
COPYpublic class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// new Thread(new Runnable()).start();
// new Thread(new FutureTask<V>()).start();
// new Thread(new FutureTask<V>( Callable )).start();
new Thread().start(); // 怎么启动Callable
MyThread thread = new MyThread();
FutureTask futureTask = new FutureTask(thread); // 适配类
new Thread(futureTask,"A").start();
new Thread(futureTask,"B").start(); // 结果会被缓存,效率高
Integer o = (Integer) futureTask.get(); //这个get 方法可能会产生阻塞!把他放到最后
// 或者使用异步通信来处理!
System.out.println(o);
}
}
class MyThread implements Callable<Integer> {
@Override
public Integer call() {
System.out.println("call()"); // 会打印几个call
// 耗时的操作
return ;
}
}
countDownLatch.countDown();
// 数量-1countDownLatch.await();
// 等待计数器归零,然后再向下执行countDownLatch.await()
就会被唤醒,继续执行COPY// 计数器
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
// 总数是6,必须要执行任务的时候,再使用!
CountDownLatch countDownLatch = new CountDownLatch();
for (int i = ; i <= ; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+" Go out");
countDownLatch.countDown(); // 数量-1
},String.valueOf(i)).start();
}
countDownLatch.await(); // 等待计数器归零,然后再向下执行
System.out.println("Close Door");
}
}
只有等待线程数达到特定值才会结束。
COPYpublic class CyclicBarrierDemo {
public static void main(String[] args) {
/**
* 集齐7颗龙珠召唤神龙
*/
// 召唤龙珠的线程
CyclicBarrier cyclicBarrier = new CyclicBarrier(,()->{
System.out.println("召唤神龙成功!");
});
for (int i = ; i <= ; i++) {
final int temp = i;
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"收集"+temp+"个龙珠");
try {
cyclicBarrier.await(); // 等待
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}
规定有几个线程就最多只能有几个线程执行。 - semaphore.acquire();
获得,假设如果已经满了,等待,等待被释放为止 - semaphore.release();
释放,会将当前的信号量释放 + 1,然后唤醒等待的线程!
作用: 多个共享资源互斥的使用!并发限流,控制大的线程数!
COPYpublic class SemaphoreDemo {
public static void main(String[] args) {
// 线程数量:停车位! 限流!
Semaphore semaphore = new Semaphore();
for (int i = ; i <= ; i++) {
new Thread(()->{
// acquire() 得到
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"抢到车位");
TimeUnit.SECONDS.sleep();
System.out.println(Thread.currentThread().getName()+"离开车位");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
semaphore.release(); // release() 释放
}
},String.valueOf(i)).start();
}
}
}
ReadLock
读锁,多线程可同时占有执行。 writeLock()
写锁,多线程中只能一个线程占有执行。
COPY/**
* 独占锁(写锁) 一次只能被一个线程占有
* 共享锁(读锁) 多个线程可以同时占有
* ReadWriteLock
* 读-读 可以共存!
* 读-写 不能共存!
* 写-写 不能共存!
*/
public class ReadWriteLockDemo {
public static void main(String[] args) {
MyCache myCache = new MyCache();
// 写入
for (int i = ; i <= ; i++) {
final int temp = i;
new Thread(()->{
myCache.put(temp+"",temp+"");
},String.valueOf(i)).start();
}
// 读取
for (int i = ; i <= ; i++) {
final int temp = i;
new Thread(()->{
myCache.get(temp+"");
},String.valueOf(i)).start();
}
}
}
// 加锁的
class MyCacheLock{
private volatile Map<String,Object> map = new HashMap<>();
// 读写锁: 更加细粒度的控制
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private Lock lock = new ReentrantLock();
// 存,写入的时候,只希望同时只有一个线程写
public void put(String key,Object value){
readWriteLock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key,value);
System.out.println(Thread.currentThread().getName()+"写入OK");
} catch (Exception e) {
e.printStackTrace();
} finally {
readWriteLock.writeLock().unlock();
}
}
// 取,读,所有人都可以读!
public void get(String key){
readWriteLock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
} catch (Exception e) {
e.printStackTrace();
} finally {
readWriteLock.readLock().unlock();
}
}
}
/**
* 自定义缓存
*/
class MyCache{
private volatile Map<String,Object> map = new HashMap<>();
// 存,写
public void put(String key,Object value){
System.out.println(Thread.currentThread().getName()+"写入"+key);
map.put(key,value);
System.out.println(Thread.currentThread().getName()+"写入OK");
}
// 取,读
public void get(String key){
System.out.println(Thread.currentThread().getName()+"读取"+key);
Object o = map.get(key);
System.out.println(Thread.currentThread().getName()+"读取OK");
}
}
COPY public static void test2(){
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();
System.out.println(blockingQueue.offer("a"));
System.out.println(blockingQueue.offer("b"));
System.out.println(blockingQueue.offer("c"));
System.out.println(blockingQueue.peek());
// System.out.println(blockingQueue.offer("d")); // false 不抛出异常!
System.out.println("============================");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll()); // null 不抛出异常!
}
3. 死死的阻塞等待 put==>take
COPYpublic static void test3() throws InterruptedException {
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();
// 一直阻塞
blockingQueue.put("a");
blockingQueue.put("b");
blockingQueue.put("c");
// blockingQueue.put("d"); // 队列没有位置了,一直阻塞
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take()); // 没有这个元素,一直阻塞
}
COPY public static void test4() throws InterruptedException {
// 队列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>();
blockingQueue.offer("a");
blockingQueue.offer("b");
blockingQueue.offer("c");
// blockingQueue.offer("d",2,TimeUnit.SECONDS); // 等待超过2秒就退出
System.out.println("===============");
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
blockingQueue.poll(,TimeUnit.SECONDS); // 等待超过2秒就退出
}
同步队列
COPYpublic static void main(String[] args) {
BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); // 同步队列
new Thread(()->{
try {
System.out.println(Thread.currentThread().getName()+" put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName()+" put 2");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName()+" put 3");
blockingQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T1").start();
new Thread(()->{
try {
TimeUnit.SECONDS.sleep();
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
TimeUnit.SECONDS.sleep();
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
TimeUnit.SECONDS.sleep();
System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"T2").start();
}
线程池的好处: 1. 降低资源的消耗 2. 提高响应的速度 3. 方便管理。
线程复用、可以控制最大并发数、管理线程
ExecutorService threadPool = Executors.newSingleThreadExecutor();
// 单个线 程ExecutorService threadPool = Executors.newFixedThreadPool(10);
// 创建一 个固定的线程池的大小ExecutorService threadPool = Executors.newCachedThreadPool();
// 由执行的线程数决定大小。具体过程:创建线程池对象--->执行---->关闭线程池
COPY try {
for (int i = ; i <= ; i++) {
// 使用了线程池之后,使用线程池来创建线程
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 线程池用完,程序结束,关闭线程池
threadPool.shutdown();
}
一般开发中使用线程池最好不使用Executors去创建对象。如阿里巴巴开发手册所说的:OOM(内存溢出)
所以应该使用ThreadPoolExecutor来创建线程池 查看源码得到以下: - 当执行线程数大于核心线程锁时,其它休眠的线程位置将会启用,最多不超过最大线程数。 - 当超过最大线程锁的线程任务会在阻塞队列中进行等待。 - 当此时并发执行的线程任务超过了最大线程数+阻塞队列长度时,拒绝策略起效。
拒绝策略有四种
new ThreadPoolExecutor.AbortPolicy()
// 同时并发的线程数大于最大线程数+阻塞队列长度时,不处理超过部分的线程任务,直接抛出异常。new ThreadPoolExecutor.DiscardPolicy()
//同时并发的线程数大于最大线程数+阻塞队列长度时,直接丢掉超过部分的线程任务,不抛出异常。new ThreadPoolExecutor.CallerRunsPolicy()
// 同时并发的线程数大于最大线程数+阻塞队列长度时,超过部分的线程任务回归main主线程执行。new ThreadPoolExecutor.DiscardOldestPolicy()
//阻塞队列满了,尝试去和最先执行的线程发起竞争,也不会抛出异常!COPYpublic ThreadPoolExecutor(int corePoolSize, //核心线程数
int maximumPoolSize, //最大线程数。最多几个线程并发。
long keepAliveTime, //当线程无任务时,几秒后结束该线程
TimeUnit unit,//线程结束的时间单位
BlockingQueue<Runnable> workQueue,//阻塞队列,限制等候线程数
ThreadFactory threadFactory,//线程工厂
RejectedExecutionHandler handler)//拒绝策略
例子如下:
COPYExecutorService threadPool = new ThreadPoolExecutor(
,
Runtime.getRuntime().availableProcessors(),
,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardOldestPolicy()); //队列满了,尝试去和最早的竞争,也不会抛出异常!
如何选择线程池的大小
Runtime.getRuntime().availableProcessors()
获取当前主机的核数。