多线程初阶3(单例模式，阻塞式队列，线程池和计时器)·多线程案例

122 阅读 0 评论 81 点赞

我是靠谱客的博主大方面包，这篇文章主要介绍多线程初阶3(单例模式，阻塞式队列，线程池和计时器)·多线程案例，现在分享给大家，希望可以做个参考。

·多线程案例

在开发中，为了提高对象的复用性，大多采用单例模式的写法以达到重复利用对象的目的

一.单例模式

1.饿汉模式
可以保证多个线程下的唯一实例，类加载的时候对象就产生了（根据字面意思记，因为总是感觉到很饿，所以每次都先把对象实例化出来）

复制代码

1
2
3
4
5
6
7
8
class Singleton {
private static Singleton instance=new Singleton();
private Singleton(){}
public static Singleton getInstance(){
return instance;
}
}

2.懒汉模式——单线程版
在多线程环境下不能保证单例的唯一性（可以理解为因为懒，所以先不进行对象实例化，等到用的时候再实例化）

复制代码

1
2
3
4
5
6
7
8
9
10
11
class Singleton{
private static Singleton instance=null;
private Singleton(){}
public static Singleton getInstance(){
if(instance==null){
instance=new Singleton();
}
return instance;
}
}

3.懒汉模式——多线程版-性能低

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
class Singleton{
private static Singleton instance=null;
private Singleton(){}
public synchronized
static Singleton getInstance(){
if(instance==null){
instance=new Singleton();
}
return instance;
}
}

4.懒汉模式——多线程版-双重校验-性能高

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Singleton{
//volatile禁止重排序操作
private static volatile Singleton instance=null;
private Singleton(){}
public static Singleton getInstance(){
if(instance==null){
synchronized (Singleton.class){
if(instance==null){
instance=new Singleton();
}
}
}
return instance;
}
}

二.阻塞式队列

生产消费者模型
生产者消费者模型就是通过一个容器来解决生产者和消费者的强耦合问题，生产者和消费者直接不直接通讯，而通过阻塞式队列来进行通信，所以生产者生产完数据后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞式队列取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

代码示例：

复制代码

//1.实现阻塞式队列，满足生产者消费者模型
//2.在生产消费达到上限或者下限的时候，不能生产或消费
public class MessageQueue<E> {
private Object[] items;
//存放商品的数组
private int takeIndex;
//弹出索引
private int putIndex;
//存放添加商品的索引
private int size;
//有效元素个数
public MessageQueue(int capacity){
items=new Object[capacity];
}
//做添加操作
public synchronized void put(E e) throws InterruptedException{
while(size==items.length){
//达到上限，需要等待
wait();
}
items[putIndex]=e;
putIndex=(putIndex+1)%items.length;
size++;
notifyAll();
}
//做弹出操作
public synchronized E take() throws InterruptedException{
while(size==0){
//达到下限，需要等待
wait();
}
size--;
notifyAll();
E e=(E) items[takeIndex];
takeIndex=(takeIndex+1)%items.length;
return e;
}
public static void main(String[] args) throws InterruptedException{
MessageQueue<Integer> queue=new MessageQueue<>(100);
for(int i=0;i<5;i++){
final int k=i;
new Thread(new Runnable() {
@Override
public void run() {
try {
for(int j=0;j<100;j++){
queue.put(k*100+j);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
while(true){
int num=queue.take();
System.out.println(num);
}
}
}

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
//1.实现阻塞式队列，满足生产者消费者模型
//2.在生产消费达到上限或者下限的时候，不能生产或消费
public class MessageQueue<E> {
private Object[] items;
//存放商品的数组
private int takeIndex;
//弹出索引
private int putIndex;
//存放添加商品的索引
private int size;
//有效元素个数
public MessageQueue(int capacity){
items=new Object[capacity];
}
//做添加操作
public synchronized void put(E e) throws InterruptedException{
while(size==items.length){
//达到上限，需要等待
wait();
}
items[putIndex]=e;
putIndex=(putIndex+1)%items.length;
size++;
notifyAll();
}
//做弹出操作
public synchronized E take() throws InterruptedException{
while(size==0){
//达到下限，需要等待
wait();
}
size--;
notifyAll();
E e=(E) items[takeIndex];
takeIndex=(takeIndex+1)%items.length;
return e;
}
public static void main(String[] args) throws InterruptedException{
MessageQueue<Integer> queue=new MessageQueue<>(100);
for(int i=0;i<5;i++){
final int k=i;
new Thread(new Runnable() {
@Override
public void run() {
try {
for(int j=0;j<100;j++){
queue.put(k*100+j);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
while(true){
int num=queue.take();
System.out.println(num);
}
}
}

三.线程池

线程池最大的好处就是减少每次启动，销毁线程的损耗。

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
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPool {
public static void main(String[] args) throws InterruptedException{
//创建线程池,初始化4个线程，但是此时还没有任务，这4个线程就处于阻塞状态，等待接收任务
//线程池可以接收大于参数的任务(>4),如果当前所有线程都处于忙碌运行状态，多出来的任务就放到仓库(线程池内部的一个属性，阻塞队列)
//等待空闲的线程来执行，如果仓库中没有任务了，线程就等待，一直等到有任务再去执行
ExecutorService pool= Executors.newFixedThreadPool(4);
pool.execute(new Runnable() {
@Override
public void run() {
//任务代码，线程池里的线程就可以直接运行任务代码
System.out.println("送快递");
}
});
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("倒杯水");
}
});
System.out.println("正在忙");
}
}

2.复杂版线程池：

复制代码

import java.util.concurrent.*;
public class ThreadPool {
public static void main(String[] args) throws InterruptedException{
//创建线程池，有7个参数
ExecutorService pool= new ThreadPoolExecutor(
3,
//1.corePoolSize核心线程数(正式线程数)
5,
//2.maximumPoolSize最大线程数(正式工+临时工来完成任务：临时工是任务比价多，正式工忙碌不过来才会去雇佣临时工)
30,
//3.keepAliveTime时间数量(若任务比较少，临时工距离上次工作之后处于空闲的时间达到给定的时间范围，就将临时工‘解雇掉’)
TimeUnit.SECONDS,
//4.时间单位
//5.阻塞式队列，存放任务的结构体(如果阻塞式队列任务已经放满，还要往线程里放任务，线程池提供多种策略)
new ArrayBlockingQueue<>(1000),
//6.线程池创建Thread线程的工厂类，没有提供，就使用线程池内部默认的创建线程的方法
new ThreadFactory(){
@Override
public Thread newThread(Runnable r){
return null;
}
},
//7.拒绝策略
//CallerRunsPolicy()：线程池已经满了，已经不接受任务，任务由自己完成
//AbortPolicy()：线程池已满，再调用就直接抛出异常
//DiscardPolicy()：从阻塞式队列抛弃最新的任务（队尾）
//DiscardOldestPolicy():从阻塞式队列丢弃最旧的任务(队头)
new ThreadPoolExecutor.CallerRunsPolicy()
);
pool.execute(new Runnable() {
@Override
public void run() {
//任务代码，线程池里的线程就可以直接运行任务代码
System.out.println("送快递");
}
});
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("倒杯水");
}
});
System.out.println("正在忙");
}
}

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
import java.util.concurrent.*;
public class ThreadPool {
public static void main(String[] args) throws InterruptedException{
//创建线程池，有7个参数
ExecutorService pool= new ThreadPoolExecutor(
3,
//1.corePoolSize核心线程数(正式线程数)
5,
//2.maximumPoolSize最大线程数(正式工+临时工来完成任务：临时工是任务比价多，正式工忙碌不过来才会去雇佣临时工)
30,
//3.keepAliveTime时间数量(若任务比较少，临时工距离上次工作之后处于空闲的时间达到给定的时间范围，就将临时工‘解雇掉’)
TimeUnit.SECONDS,
//4.时间单位
//5.阻塞式队列，存放任务的结构体(如果阻塞式队列任务已经放满，还要往线程里放任务，线程池提供多种策略)
new ArrayBlockingQueue<>(1000),
//6.线程池创建Thread线程的工厂类，没有提供，就使用线程池内部默认的创建线程的方法
new ThreadFactory(){
@Override
public Thread newThread(Runnable r){
return null;
}
},
//7.拒绝策略
//CallerRunsPolicy()：线程池已经满了，已经不接受任务，任务由自己完成
//AbortPolicy()：线程池已满，再调用就直接抛出异常
//DiscardPolicy()：从阻塞式队列抛弃最新的任务（队尾）
//DiscardOldestPolicy():从阻塞式队列丢弃最旧的任务(队头)
new ThreadPoolExecutor.CallerRunsPolicy()
);
pool.execute(new Runnable() {
@Override
public void run() {
//任务代码，线程池里的线程就可以直接运行任务代码
System.out.println("送快递");
}
});
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("倒杯水");
}
});
System.out.println("正在忙");
}
}

3.自己实现一个简单的线程池：

复制代码

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
public class MyThreadPool {
private MessageQueue<Runnable> queue;
public MyThreadPool(int size,int capacity){
queue=new MessageQueue<>(capacity);
//在构造方法里直接创建好核心线程数让去阻塞式队列里取任务
for(int i=0;i<size;i++){
new Thread(new Runnable() {
@Override
public void run() {
try {
//核心线程数一直在运行
while(true) {
Runnable task = queue.take();
//直接调用实例方法去执行任务
task.run();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
public void execute(Runnable task){
try {
queue.put(task);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
MyThreadPool pool=new MyThreadPool(5,100);
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("A");
}
});
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("B");
}
});
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("C");
}
});
}
}

四.计时器

前置知识：JDK的基础的时间操作

1.Date

复制代码

1
2
3
Date date1=new Date();
Date date2=new Date(99999999);

无参构造：运行这段代码时所在主机系统当前时间的获取
一个参数：以格林威治时间1970-01-01开始，经过给定参数时间数量的毫秒

2.DateFormat（日期格式化）

复制代码

1
2
3
4
 DateFormat df=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println(df.format(date1));
System.out.println(df.format(date2));

3.System时间获取

复制代码

1
2
3
 //从1970-01-01开始到当前时间点经过的毫秒数
long current=System.currentTimeMillis();

自己实现简单的计时器：

复制代码

import java.util.concurrent.*;
/**
* 定时器
* 1.在约定好的时间点上执行某个任务
* 2.设置好间隔时间，间隔时间到后再一次通知
*/
public class MyTimer {
private BlockingQueue<MyTimerTask> queue=new PriorityBlockingQueue();
public MyTimer(int count){
for(int i=0;i<count;i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
while (true) {
MyTimerTask task = queue.take();
synchronized (queue) {
long current = System.currentTimeMillis();
if (task.next > current) {
queue.wait( task.next-current );
queue.put(task);
} else {
task.task.run();
if (task.period > 0) {
task.next = task.next + task.period;
queue.put(task);
}
}
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
public void schedule(Runnable task,long delay,long period){
try {
queue.put(new MyTimerTask(task,System.currentTimeMillis()+delay,period));
synchronized (queue){
queue.notifyAll();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static class MyTimerTask implements Comparable<MyTimerTask>{
private Runnable task;
private long next;
private long period;
public MyTimerTask(Runnable task,long next,long period){
this.task=task;
this.next=next;
this.period=period;
}
@Override
public int compareTo(MyTimerTask o){
return Long.compare(next,o.next);
}
}
public static void main(String[] args) {
new MyTimer(4).schedule(new Runnable() {
@Override
public void run() {
System.out.println("起床啦");
}
},3000,1000);
}
}

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
import java.util.concurrent.*;
/**
* 定时器
* 1.在约定好的时间点上执行某个任务
* 2.设置好间隔时间，间隔时间到后再一次通知
*/
public class MyTimer {
private BlockingQueue<MyTimerTask> queue=new PriorityBlockingQueue();
public MyTimer(int count){
for(int i=0;i<count;i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
while (true) {
MyTimerTask task = queue.take();
synchronized (queue) {
long current = System.currentTimeMillis();
if (task.next > current) {
queue.wait( task.next-current );
queue.put(task);
} else {
task.task.run();
if (task.period > 0) {
task.next = task.next + task.period;
queue.put(task);
}
}
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
public void schedule(Runnable task,long delay,long period){
try {
queue.put(new MyTimerTask(task,System.currentTimeMillis()+delay,period));
synchronized (queue){
queue.notifyAll();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static class MyTimerTask implements Comparable<MyTimerTask>{
private Runnable task;
private long next;
private long period;
public MyTimerTask(Runnable task,long next,long period){
this.task=task;
this.next=next;
this.period=period;
}
@Override
public int compareTo(MyTimerTask o){
return Long.compare(next,o.next);
}
}
public static void main(String[] args) {
new MyTimer(4).schedule(new Runnable() {
@Override
public void run() {
System.out.println("起床啦");
}
},3000,1000);
}
}