# 一、通过synchronize 中的 wait 和 notify 实现
【1】我们可以将生产者和消费者需要的方法写在公共类中
package com.yintong.concurrent;
import java.util.LinkedList;
public class Concurrentcomm {
//常量
private static int MAX_VALUE = 10;
//可以理解为缓存
LinkedList<String> linkedList = new LinkedList<>();
Object object = new Object();
/*
* 生产者方法
*/
public void product() throws Exception {
synchronized(linkedList) {
while(MAX_VALUE == linkedList.size()) {
System.out.println("仓库已满,【生产者】: 暂时不能执行生产任务!");
linkedList.wait();
}
linkedList.push(" 李四 ");
System.out.println("【生产者】:生产了一个产品\t【现仓储量为】:" + linkedList.size());
linkedList.notifyAll();
}
}
/*
* 消费者方法
*/
public void customer() throws Exception {
/*
* 根据jdk的void notifyAll()的描述,“解除那些在该对象上调用wait()方法的线程的阻塞状态。该方法只能在同步方法或同步块内部调用。
* 如果当前线程不是对象所得持有者,
* 该方法抛出一个java.lang.IllegalMonitorStateException 异常”
* so我们使用同一把锁
*/
synchronized (linkedList) {
//多线程判断中使用 while 不要使用 if 否则会出现虚假唤醒问题
while(linkedList.size() == 0) {
System.out.println("仓库无货,【消费者】: 暂时不能执行消费任务!");
linkedList.wait();
}
linkedList.pop();
System.out.println("【消费者】:消费了一个产品\t【现仓储量为】:" + linkedList.size());
linkedList.notifyAll();
}
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
【2】在 main
函数中调用生产者和消费者方法,并加限制即可
/**
* @author zzx
* @desc 生产者与消费者
*
*/
public class Concurrent {
//常量
private static int MAX_VALUE = 100;
public static void main(String[] args) {
Concurrentcomm con = new Concurrentcomm();
new Thread(new Runnable() {
@Override
public void run() {
try {
for (int i = 0; i < MAX_VALUE; i++) {
Thread.sleep(0);
con.product();
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}).start();
// 消费者
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10);
for (int i = 0; i < MAX_VALUE; i++) {
con.customer();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
【3】简单的生产者与消费者模式就完成了,可以看下运行的结果
【生产者】:生产一个产品 【现仓储量为】:10
仓库已满【生产者】:暂时不能执行生产任务! # 生产者 wait()
【消费者】:消费一个产品 【现仓储量为】:9
【生产者】:生产一个产品 【现仓储量为】:10
【消费者】:消费一个产品 【现仓储量为】:9
【消费者】:消费一个产品 【现仓储量为】:8
【消费者】:消费一个产品 【现仓储量为】:7
【消费者】:消费一个产品 【现仓储量为】:6
【消费者】:消费一个产品 【现仓储量为】:5
【消费者】:消费一个产品 【现仓储量为】:4
【消费者】:消费一个产品 【现仓储量为】:3
【生产者】:生产一个产品 【现仓储量为】:4 # 两个线程交叉执行
【消费者】:消费一个产品 【现仓储量为】:3
【消费者】:消费一个产品 【现仓储量为】:2
【消费者】:消费一个产品 【现仓储量为】:1
【消费者】:消费一个产品 【现仓储量为】:0
仓库无货,【消费者】:暂时不能执行任务! # 消费者 wait()
【生产者】:生产一个产品 【现仓储量为】:1
【消费者】:消费一个产品 【现仓储量为】:0
仓库无货,【消费者】:暂时不能执行任务!
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 二、通过 Lock
中的 await
与 signalAll
实现
【1】我们将公共的属性和方法放在 Resouce
类中,在资源类中使用 Lock
中的 lock()
进行加锁,控制并发操作。使用 await()
方法阻塞线程。使用 signalAll()
唤醒线程。
/**
* 通过 Lock 实现生产者与消费者
* 资源类:将公共的资源放在一个单独的类中,可以将其看做一个产品,自身就就有生产和消费的能力(方法)
*/
public class ProductAndConsumer {
public static void main(String[] args) {
Resouce resouce = new Resouce();
//生产者
new Thread(()->{
for (int i=1;i<=5;i++) {
resouce.product();
}
},String.valueOf("生产者")) .start();
//消费者
new Thread(()->{
for (int i=1;i<=5;i++){
resouce.consumer();
}
},String.valueOf("消费者")).start();
}
}
//资源类
class Resouce {
private int MAX_VALUE = 3;
private int MIN_VALUE = 0;
private int number = 0;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
//生产者
public void product(){
try {
lock.lock();
//如果生产的数量大于最大值则阻塞
while(number >= MAX_VALUE){
condition.await();
}
number++;
System.out.println("【生产者】:生产了一个产品\t【现仓储量为】:" + number);
condition.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
//消费者
public void consumer(){
try {
lock.lock();
//如果消费的值=0则阻塞
while(number <= MIN_VALUE){
condition.await();
}
number--;
System.out.println("【消费者】:消费了一个产品\t【现仓储量为】:" + number);
condition.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
【2】输出结果展示
【生产者】:生产一个产品 【现仓储量为】:1
【生产者】:生产一个产品 【现仓储量为】:2
【生产者】:生产一个产品 【现仓储量为】:3
【消费者】:消费一个产品 【现仓储量为】:2
【消费者】:消费一个产品 【现仓储量为】:1
【消费者】:消费一个产品 【现仓储量为】:0
【生产者】:生产一个产品 【现仓储量为】:1
【生产者】:生产一个产品 【现仓储量为】:2
【消费者】:消费一个产品 【现仓储量为】:1
【消费者】:消费一个产品 【现仓储量为】:0
2
3
4
5
6
7
8
9
10
# 三、synchronized 和 Lock 的区别
【1】原始构成:synchronized
是关键字属于 JVM 层面。底层通过 monitorenter
(进入)monitorexit
(退出)实现。底层是通过 monitor
对象完成,其实 wait/notify
等方法也依赖于 monitor
对象,只有在同步块或方法中才能调用 wait/notify
等方法。Lock
是具体类(java.util.concurrent.locks.Lock
)是 API
层面的锁。
【2】使用方法:synchronized
不需要用户手动释放锁,当 synchronized
代码执行完后,系统会自动释放锁。ReentrantLock
则需要用户手动释放锁,若未主动释放锁,就可能导致出现死锁的现象。
【3】等待是否中断:synchronized
不可中断,除非抛出异常或者正常运行完成。ReentrantLock
可中断。
1)、设置超时时间 tryLock
(long timeout
,TimeUnit unit
)
2)、lockInterruptibly()
放在代码块中,调用 interrupt()
方法可中断。
【4】加锁是否公平:synchronized
非公平锁。ReentrantLock
两者都可以,默认是非公平锁,构造方法可以传入 boolean
值,true
为公平锁,false
为非公平锁。
【5】锁绑定多个条件 Condition
:synchronized
没有。ReentrantLock
用来实现分组唤醒需要唤醒的线程们,可以精确唤醒,而不是像 synchronized
要么随机唤醒一个线程要么唤醒全部线程。
# 四、通过阻塞队列实现生产者与消费者
【1】通过blockQueue
中的 put/take
方法实现生产者与消费者,具体实现如下:当生产者使用put
生产到指定的队列大小3时,就会阻塞当前线程。这是消费者线程会通过 take
方法消费队列中的消息。当队列中没有消息时,会阻塞,直到有消息消费。
public class BlockProductConsumer {
public static void main(String[] args) {
MyResouce resouce = new MyResouce(new ArrayBlockingQueue(3));
//生产者线程
new Thread(()->{
for(int i=1;i<=10;i++){
resouce.product();
}
},"生产者").start();
//消费者线程
new Thread(()->{
for(int i=1;i<=10;i++){
try {
resouce.consumer();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"消费者").start();
try {
TimeUnit.SECONDS.sleep(1);
resouce.stop();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 公共资源类
*/
class MyResouce{
//标记 while 无限循环
private volatile boolean FLAG = true;
//队列中存入的数值
private AtomicInteger atomicInteger = new AtomicInteger();
//组合一个阻塞队列,通过构造器传入
private BlockingQueue blockingQueue;
public MyResouce(BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
}
//生产者
public void product(){
try {
while (FLAG){
blockingQueue.put(String.valueOf(atomicInteger.incrementAndGet()));
System.out.println("生产者生产第"+blockingQueue.size()+"个产品");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//消费者
public void consumer() throws InterruptedException {
while (FLAG){
blockingQueue.take();
System.out.println("消费者消费第"+(blockingQueue.size()+1)+"个产品");
}
}
public void stop(){
FLAG = false;
System.out.println("========================");
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
【2】效果展示:
生产者生产第1个产品
生产者生产第2个产品
生产者生产第3个产品
消费者消费第3个产品
消费者消费第2个产品
消费者消费第1个产品
生产者生产第1个产品
消费者消费第1个产品
消费者消费第1个产品
生产者生产第1个产品
生产者生产第1个产品
生产者生产第1个产品
消费者消费第1个产品
=========================
2
3
4
5
6
7
8
9
10
11
12
13
14