1 消息队列

消息队列：生产消费者模型

1. 队列
2. 生产者
3. 消费者

2 需求

1. 生产者生产的数据，消费者依次消费
2. 生产者当队列满的嘶吼，这个时候，所有的生产着将不再生产
3. 消费者当队列空的时候，这个时候，所有的消费者将进入等待
4. 消息中间件

2.1 实例1

1.package com.test;  
2.  
3.public class Customer implements Runnable {  
4.     private final Queue<String> queue;  
5.     public Customer(final Queue<String> queue){  
6.          this.queue =queue;  
7.     }      
8.     @Override  
9.     public  void  run() {  
10.         int i=0;  
11.         while(true) {  
12.         try {  
13.            Thread.sleep(1000);  
14.        } catch (InterruptedException e) {  
15.            // TODO Auto-generated catch block  
16.            e.printStackTrace();  
17.        }  
18.         System.out.println("消费者消费了"+this.queue.get());  
19.  
20.         }  
21.     }  
22.}  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

2.2 实例2

23.package com.test;  
24.  
25.public class Producer implements Runnable {  
26.     private final Queue<String> queue;  
27.     public Producer(final Queue<String> queue){  
28.          this.queue =queue;  
29.     }  
30.     @Override  
31.     public  void  run(){  
32.          int i=0;  
33.          while(true) {  
34.              try {  
35.                Thread.sleep(500);  
36.            } catch (InterruptedException e) {  
37.                // TODO Auto-generated catch block  
38.                e.printStackTrace();  
39.            }  
40.              this.queue.add("第"+(++i)+"消息");  
41.              System.out.println("生产者生产了第"+i+"个消息");  
42.          }  
43.     }  
44.}  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

2.3 实例3

45.package com.test;  
46.  
47.import java.util.LinkedList;  
48.  
49.public class Queue<T> {  
50.       //设计的是队列FIFO，还是栈FILO  
51.       private LinkedList<T> queue=new LinkedList<T>();   
52.       private int min;  
53.       private int max;  
54.       private  final static  int QUEUE_SIZE_DEFAULT_MAX=10;  
55.       public Queue() {  
56.            this(QUEUE_SIZE_DEFAULT_MAX);  
57.       }     
58.       public Queue(int maxSize) {  
59.            this.max=maxSize;  
60.       }  
61.       //往队列当中放数据  
62.       public synchronized  void add(T t) {  
63.           //当消息满的时候，进入等待  
64.           while(this.queue.size() ==this.max) {  
65.                try {  
66.                     System.out.println("消息已满，开始等待");  
67.                    this.wait();  //等待是释放锁的  
68.                } catch (InterruptedException e) {  
69.                    // TODO Auto-generated catch block  
70.                    e.printStackTrace();  
71.                }  
72.            }  
73.           this.queue.addLast(t);  
74.           this.notifyAll(); // 队列不再空的时候，唤醒所有线程  
75.       }  
76.       // 往队列当中取数据     
77.       public synchronized  T get() {  
78.           //当消息为空的时候，进入等待  
79.           while(this.queue.size() ==0) { //改成while，保证被唤醒后，再来判断一次  
80.                try {  
81.                    System.out.println("消息为空，开始等待");  
82.                    this.wait();  //等待是释放锁的,被唤醒后，重新持锁，会从该从继续执行  
83.                } catch (InterruptedException e) {  
84.                    // TODO Auto-generated catch block  
85.                    e.printStackTrace();  
86.                }  
87.           }  
88.           T t=this.queue.getFirst();  
89.           // 取用后的消息，不应当在存在与队列当中  
90.           this.queue.removeLast();  
91.           this.notifyAll(); //队列不再满的时候，唤醒所有线程  
92.           return t;  
93.      }  
94.}  
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

2.4 实例4

95.package com.test;  
96.  
97.public class Test {  
98.    public static  void  main(String[] args) {  
99.         Queue<String> q=new Queue<String>();  
100.         Thread c1=new Thread(new Customer(q));  
101.         c1.start();  
102.         Thread c2=new Thread(new Customer(q));  
103.         c2.start();  
104.         Thread p1=new Thread(new Producer(q));  
105.         p1.start();  
106.    }  
107.} 
1
2
3
4
5
6
7
8
9
10
11
12
13

3 池

3.1 注意事项

将一些对象放入“池”中，然后交给池托管这些对象的执行方式，作为使用者，不需要再去关心如何管理这些对象

3.2 线程池管理

线程池管理的是线程

3.2.1 固定长度线程池

1. 线程池中，线程用完后是要全部回收的，再用重新分配
2. 完成了对线程对象的重用，避免了重复创建线程对象造成的资源浪费
3. 关闭线程池后，不可以再添加新的线程到池当中，再加就会报错

3.2.2 单任务线程

类似join,是让线程串行执行

1. 所有线程用完归还给池
2. 依次执行

3.2.3 可变尺寸线程池

可变尺寸线程池的执行，是全部并发

3.2.4 延迟的线程

是在其被加入到池当中以后，立即开始计时

1. 当前线程池资源被占用，延迟线程延时排队执行
2. 如果资源没有被占用，则加入到批次中
3. 延迟线程一样需要竞争资源

3.2.5 自定义线程池

1. 核心线程满了后，加入队列，是同时可以执行的线程数
2. 队列线程满了以后，如果最大线程池的数量没满，则队列多余的线程立即执行，这些线程是不经过线程池管理的线程，他们没有做到对象的重用
3. 如果队列满了，最大线程池数量也满了，则不再接受新的线程执行
4. 队列内的线程会被核心线程不断的重复取用

3.2.6 有返回值的线程

1. 实现了Callable接口
2. 任何线程必须join到Callable之后才会执行
3. 应用场景：当当前线程需要前一个线程的执行结果，才能继续执行时，需要get()结果
4. 如果不需要返回值，则其和普通线程没有任何区别