多线程：7-线程协作

阅读量

7.线程协作

线程通信

应用场景：生产者消费者问题

  • 假设仓库中只能存放一件产品，生产者将生产出来的商品放入仓库，消费者取走仓库中的产品。
  • 如果仓库中没有产品，生产者将产品放入仓库，否则停止生产并等待，直到仓库中产品被消费者取走。
  • 如果仓库中放有产品，消费者可以将产品取走，否则消费者停止消费并等待，直到仓库中有产品为止。
    加载图片！

这是一个线程同步问题，生产者和消费者共享同一资源，并且生产者和消费者之间相互依赖，互为条件。

  • 对于生产者，没有生产 产品之前，要通知消费者等待。生产了之后，又要通知消费者 消费。
  • 对于消费者，消费之后，要通知生产者生产新的产品。
  • 在生产者消费者问题中，仅有synchronized是不够的。
    • synchronized 可阻止并发更新同一共享资源，实现了同步。
    • synchronized 不能用来实现不同线程之间的通信。

Java提供了几个方法，用来解决线程间通信问题：

方法 作用
wait() 表示线程一直等待，直到接到其他线程通知。
与sleep() 不同，会释放锁。
wait(long timeout) 指定等待的毫秒数。
notify() 唤醒一个处于等待状态的线程。
notifyAll() 唤醒同一个对象上所有调用wait() 方法的线程。
优先级高的线程 优先调度。

注意： 这些均是Object类中的方法，都只能在同步方法 或 同步代码块中使用，否则会抛出异常lllegalMonitorStateException.

解决方法1：管程法

  • 生产者：负责生产数据的模块（可能是方法、对象、线程、进程）
  • 消费者：负责处理数据的模块（可能是方法、对象、线程、进程）
  • 缓冲区：消费者不能直接使用生产者的数据，他们之间有个缓存区。

生产者将生产的数据放入缓冲区，消费者从缓冲区拿数据。
加载图片！

解决方法2：信号灯法

  • 用一个标志位。

1.管程法

1
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
// 管程法：需要生产者、消费者、产品、缓冲区
public class TestPC {
    public static void main(String[] args) {

        // 缓冲区 负责连接生产者和消费者
        SynContainer container = new SynContainer();

        new Productor(container).start();    //启动生产者
        new Consumer(container).start();     //启动消费者
    }
}

// 生产者
class Productor extends Thread{

    // 缓冲区 负责连接生产者和消费者
    SynContainer container;

    public Productor(SynContainer container) {
        this.container = container;
    }

    // 生产产品 100只鸡
    public void run() {
        for(int i=0; i<100; i++) {
            System.out.println("生产第-->" + i + "只鸡");
            container.push(new Chicken(i));  // 缓冲区加入产品
        }
    }
}

// 消费者
class Consumer extends Thread{

    // 缓冲区 负责连接生产者和消费者
    SynContainer container;

    public Consumer(SynContainer container) {
        this.container = container;
    }

    // 消费者 消费1000只鸡
    public void run() {
        for(int i=0; i<1000; i++) {
            System.out.println("消费第-->" + container.pop().id + "只鸡");
        }
    }
}

// 产品
class Chicken{
    int id; // 产品id

    public Chicken(int id) {
        this.id = id;
    }
}

// 缓冲区
class SynContainer{

    // 容器大小（缓冲区大小）
    Chicken[] chickens = new Chicken[10];

    // 容器计数器
    int count = 0;

    // 生产者 放入产品 （需要同步，因为涉及到并发）
    public synchronized void push(Chicken chicken){
        // 容器若满了，通知并等待 消费者消费
        if(count == chickens.length){
            try {
                this.wait();    //线程阻塞，消费者通知生产时解除阻塞
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 容器未满，放入产品
        chickens[count] = chicken;
        count++; // 存一个,计数器就加一

        // 生产后, 容器中有数据了, 可以通知消费
        this.notifyAll();
    }

    // 消费者 消费产品 (拿的时候，从最后一个位置开始拿数据)
    public synchronized Chicken pop(){
        // 如果缓冲区为空 就等待生产者生产
        if(count == 0){
            try {
                this.wait();  //线程阻塞，生产者通知消费时解除阻塞
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 缓冲区不为空，则可以消费
        count--;  // 根据存储的代码  来理解  为什么要先自减
        Chicken chicken = chickens[count];

        // 消费后，缓冲区有空间了，通知生产者生产
        this.notifyAll();

        return chicken;
    }
}

2.信号灯法

1
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
/**
 * 信号灯法
 * 借助标志位
 * 
 * 例：看直播 演员说什么 我们听什么
 */
public class CoTest02 {
	public static void main(String[] args) {

		Tv tv = new Tv();

		new Player(tv).start();  // 开启演员线程
		new Watcher(tv).start(); // 开启观众线程
	}
}

// 生产者 演员
class Player extends Thread{

	Tv tv;

	public Player(Tv tv) {
		this.tv = tv;
	}
	
	public void run() {
		for(int i=0; i<20; i++) {  // 循环20次 奇数偶数 说不同的话
			if(i % 2 == 0) {
				this.tv.paly("直播胡扯");
			}else {
				this.tv.paly("直播广告1234 ");
			}
		}
	}
}

// 消费者 观众
class Watcher extends Thread{

	Tv tv;

	public Watcher(Tv tv) {
		this.tv = tv;
	}
	
	public void run() {
		for(int i=0; i<20; i++) {
			tv.watch();
		}
	}
}

// 同一份资源 电视
class Tv{

	String voice;  // 听的声音
	
	// 加入信号灯  通过信号灯进行切换
	// true  表示演员表演 观众等待
	// false 表示观众观看 演员等待
	boolean flag = true;

	// 演员表演 观众等待 （涉及到并发）
	public synchronized void paly(String voice) {

		// 演员等待 观众观看完
		if(flag == false) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		// 表演时刻
		System.out.println("表演了->" + voice);
		this.voice = voice;
		
		// 唤醒 -- 演员表演 就可以 唤醒观众了
		this.notifyAll();

		// 切换标志 标志位取反
		this.flag = !this.flag; 
	}

	// 观众观看 演员等待 （涉及到并发）
	public synchronized void watch() {

		// 观众等待 演员表演完
		if(flag == true) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		// 观看
		System.out.println("听到了<-" + voice);
		
		// 唤醒 -- 观众听完了 就可以 唤醒演员继续说
		this.notifyAll();

		// 切换标志 标志位取反
		this.flag = !this.flag; 
	}
}

结果：

1
2
3
4
5
6
7
8
9
10
11
12
表演了->直播胡扯
听到了<-直播胡扯
表演了->直播广告1234 
听到了<-直播广告1234 
表演了->直播胡扯
听到了<-直播胡扯
表演了->直播广告1234 
听到了<-直播广告1234 
表演了->直播胡扯
听到了<-直播胡扯
表演了->直播广告1234 
听到了<-直播广告1234

使用线程池

  • 背景：经常创建和销毁、使用容量特别大的资源。在并发的情况下的线程，对性能影响非常大。
  • 思路：提前创建好多个线程，放入线程池中。使用时直接获取，使用完再放回池中。可避免频繁创建销毁，实现了重复利用。
  • 好处：
    • 提高响应速度（减少了创建新线程的时间）
    • 降低资源消耗（重复利用线程池中的线程，不需要每次都创建）
    • 便于线程管理
      • corePoolSize：线程池大小
      • maximumPoolSize：最大线程数
      • keepAliveTime：线程没有任务时，最多保持多长时间后 终止

ExecutorService 与 Executors：

  • ExecutorService：真正的线程池接口。常用子类ThreadPoolExecutor
    • void execute(Runnable command) ：执行任务/命令，一般用来执行Runnable
    • <T>Future<T> submit(Callable<T> task)：执行任务，一般用来执行Callable
    • void shutdown()：关闭线程池
  • Executors：工具类、线程池的工厂类，用于创建并返回不同类型的线程池。

测试：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class TestPool {
    public static void main(String[] args) {
        // 创建线程池（大小为10）
        ExecutorService es = Executors.newFixedThreadPool(10);

        // 把线程丢入池中，线程自动启动
        es.execute(new MyThread());
        es.execute(new MyThread());

        // 关闭连接
        es.shutdown();
    }
}

class MyThread implements Runnable{
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName());
    }
}

结果：

1
2
pool-1-thread-1
pool-1-thread-2
赏

失败是成功之母

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

JAVA<br><br>学习永不止步<br>简简单单