Java并发编程

创建线程

1
2
3
4
5
6
7
Thread thread = new Thread("thread-1"){
    @Override
    public void run() {
        logger.info("hello world");
    }
};
thread.start();
1
2
3
4
5
6
7
Runnable runnable = new Runnable() {
    public void run() {
        logger.info("hello world");
    }
};
Thread thread = new Thread(runnable,"thread");
thread.start();
1
2
3
4
5
Runnable runnable = ()->{
    logger.info("hello world");
};
Thread thread = new Thread(runnable,"thread");
thread.start();
1
2
3
4
5
6
7
8
9
10
11
12
13
FutureTask<String> task = new FutureTask<String>(new Callable<String>() {
    @Override
    public String call() throws Exception {
        logger.info("running start...");
        Thread.sleep(1000);
        logger.info("running end...");
        return "ok";
    }
});
Thread thread = new Thread(task,"thread");
thread.start();

logger.info(task.get());//等待线程返回的结果

Thread常见方法

  • start():启动新的线程，线程处于就绪状态
  • run():线程启动调用的方法

    直接调用run方法时则由主线程进行执行

  • join():等待线程运行结束(调用线程加入当前线程执行）
  • setPriority():设置线程优先级
  • getState():获取线程状态
  • sleep(n):线程休眠n毫秒，结束后处于就绪状态
  • yield():让出CPU控制权
  • interrupt():线程中断状态设为true

    sleep/wait/join状态下，线程抛出InterruptedException，sleep清除打断标记

  • interrupted()：返回当前线程的中断状态；将当前线程的中断状态设为false，即清除打断标记
  • isInterrupted()：返回当前线程的中断状态，不清除打断标记
1
2
3
4
5
6
7
8
9
10
11
12
13
Thread thread = new Thread( new Runnable(){
    public void run(){
        // 若未发生中断，就正常执行任务
        while(!Thread.currentThread.isInterrupted()){
            // 正常任务代码……
        }

        // 中断的处理代码……
        doSomething();
    }
} ).start();
doSomething();
thread.interrupt();
  • setDaemon():设置为守护线程(非守护线程执行完毕，守护线程即停止)

synchronized

1
2
3
synchronized (对象){
	//临界区            
}

线程进入临界区需要获得锁，临界区的代码执行完才会释放锁，并唤醒阻塞线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
static int count = 0;
static Object lock = new Object();
public static void main(String[] args){
	Thread increment = new Thread(()->{
	    synchronized (lock){
	        count++;
	    }
	});
	
	Thread decrement = new Thread(()->{
	    synchronized (lock){
	        count--;
	    }
	});
	
	increment.start();
	decrement.start();
	increment.join();
	decrement.join();
	
	logger.info("count:"+count);//count始终为0
}
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
class Utils{
    private int count = 0;
    public void increment(){
        synchronized (this){
            count++;
        }
    }
    public void decrement(){
        synchronized (this){
            count--;
        }
    }
    public int getCount(){
        return this.count;
    }
}

public static void main(String[] args) {
    Utils utils = new Utils();
    Thread inc = new Thread(()->{
        utils.increment();
    });
    Thread dec = new Thread(()->{
        utils.decrement();
    });
    inc.start();
    dec.start();
    inc.join();
    dec.join();

    logger.info("count:"+utils.getCount());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public synchronized void increment(){
	//doSomething
}
public void increment(){
	synchronized (this){
	    //doSomething
	}
}
//以上等价

public static synchronized void increment(){
	//doSomething
}
public void increment(){
	synchronized (Utils.class){
	    //doSomething
	}
}
//以上等价

name.Class和this不是同一个对象，类对象只有一个，this为实例，可以有多个

Monitor(重量级锁)


加synchronized对象会和一个对应的Monitor对象关联(对象的MarkWord置为Monitor对象的指针)，当线程成为Monitor的owner时才能执行临界区的代码，即获得锁，否则阻塞，添加到EntryList中，当owner为空时竞争锁。
WaitSet中的线程为已经获得过锁，但条件不满足放弃锁，进入waiting状态。owner线程调用notify()时唤醒WaitSet中的线程。

轻量级锁

线程访问临界区的时间错开，置锁对象的MarkWord为栈帧中琐记录值来上锁

锁重入：同一线程对同一对象再加锁

偏向锁：用threadId替换锁对象的markWord，若有锁重入，即再次加锁时检测锁对象的MarkWord是否等于threadId，一致则仅在栈帧中添加一条琐记录

1
2
3
4
5
6
7
8
9
10
11
public void increment(){
    synchronized (this){
				count++;
        decrement(); 
    }
}
public void decrement(){
    synchronized (this){
        count--;
    }
}

锁膨胀

Thread-1尝试进行轻量级加锁，但Thread-0已加锁，则Thread-1为Object对象申请Monitor锁对象，进行重量级加锁，Thread-1进入Monitor的EntryList

Thread-0进入轻量级锁流程，由于此时为重量级加锁，失败

执行重量级解锁流程，即置Monitor对象的owner为空，并唤醒EntryList中的阻塞线程

wait/notify

只有获得锁的线程才能调用wait/notify方法,即需在synchronized代码块中调用(此时为重量级锁)

wait()会释放锁，sleep()不会释放锁

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
static Logger logger = Logger.getLogger("log");
static Object lock = new Object();
public static void main(String[] args) throws ExecutionException, InterruptedException {

    Thread thread = new Thread(()->{
        synchronized (lock){
            logger.info("执行...");
            try {
                lock.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            logger.info("被唤醒，执行...");
        }
    });
    thread.start();
    Thread.sleep(1000);
    synchronized (lock){
        logger.info("主线程执行...");
        logger.info("唤醒子线程...");
        lock.notify();
    }
}

十二月 04, 2020 3:51:24 下午 Main lambda$main$0
信息: 执行...
十二月 04, 2020 3:51:25 下午 Main main
信息: 主线程执行...
十二月 04, 2020 3:51:25 下午 Main main
信息: 唤醒子线程...
十二月 04, 2020 3:51:25 下午 Main lambda$main$0
信息: 被唤醒，执行...

同步-保护性暂停

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
class GuardObject {
    private String response;
    public String get() {
        synchronized (this){
            while (response == null)
                try {
                    this.wait();
                }catch (Exception e){
                    e.printStackTrace();
                }
            return response;
        }
    }
    //等待超时方法
    public String get(long timeout) {
        synchronized (this){
            long begin = System.currentTimeMillis();
            long passedTime = 0;

            while (response == null){
                long waitTime = timeout - passedTime;
                if(waitTime <= 0) break;
                try {
                    this.wait(waitTime);
                }catch (Exception e){
                    e.printStackTrace();
                }
                passedTime = System.currentTimeMillis() - begin;
            }
            return response;
        }
    }
    
    public void set(String response) {
        synchronized (this){
           this.response = response;
           this.notifyAll();
        }
    }
}

public static void main(String[] args){
    GuardObject guardObject = new GuardObject();
    new Thread(()->{
        String res = guardObject.get();
        //String res = guardObject.get(1000);
        logger.info("response:"+res.toString());
    }).start();

    new Thread(()->{
		    //Thread.sleep(2000);
        guardObject.set("I set something.");
    }).start();
}
信息: response:I set something.

生产者-消费者

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
class MessageQueue {
    Logger logger = Logger.getLogger("log");
    private LinkedList<Message> queue = new LinkedList<>();
    private int capcity;
    public  MessageQueue(int capcity){
        this.capcity = capcity;
    }
    public Message take(){
        synchronized (queue){
            while (queue.isEmpty()){
                try {
                    logger.info("queue is empty,waiting...");
                    queue.wait();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }

            Message message = queue.removeFirst();
            logger.info("consume message,"+message);
            queue.notifyAll();
            return  message;
        }
    }

    public void put(Message message){
        synchronized (queue){
            while (queue.size() == capcity){
                try {
                    logger.info("queue is full,waiting...");
                    queue.wait();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            queue.addLast(message);
            logger.info("add message,"+message);
            queue.notifyAll();
        }
    }

    @Override
    public String toString() {
        return "MessageQueue{" +
                "queue=" + queue +
                ", capcity=" + capcity +
                '}';
    }
}

final class Message{
    private int id;
    private String value;
    public Message(int id,String value){
        this.id = id;
        this.value = value;
    }

    public int getId() {
        return id;
    }

    public String getValue() {
        return value;
    }
}

static Logger logger = Logger.getLogger("log");
public static void main(String[] args){
    final MessageQueue messageQueue = new MessageQueue(5);
    //三个生产者线程
    for (int i = 0; i < 3; i++) {
        int id = i;
        new Thread(()->{
            messageQueue.put(new Message(id,"value:"+id));
        },"生产者"+i).start();
    }

    //消费者线程
    new Thread(()->{
        while (true){
            Message message = messageQueue.take();
            logger.info(message.toString());
        }
    }).start();
}

十二月 10, 2020 3:46:38 下午 MessageQueue put
信息: add message,Message@483b79f5
十二月 10, 2020 3:46:38 下午 MessageQueue take
信息: consume message,Message@483b79f5
十二月 10, 2020 3:46:38 下午 Main lambda$main$1
信息: Message@483b79f5
十二月 10, 2020 3:46:38 下午 MessageQueue put
信息: add message,Message@6a6daa45
十二月 10, 2020 3:46:38 下午 MessageQueue put
信息: add message,Message@3555b188
十二月 10, 2020 3:46:38 下午 MessageQueue take
信息: consume message,Message@6a6daa45
十二月 10, 2020 3:46:38 下午 Main lambda$main$1
信息: Message@6a6daa45
十二月 10, 2020 3:46:38 下午 MessageQueue take
信息: consume message,Message@3555b188
十二月 10, 2020 3:46:38 下午 Main lambda$main$1
信息: Message@3555b188
十二月 10, 2020 3:46:38 下午 MessageQueue take
信息: queue is empty,waiting...

Park/Unpark

1
2
LockSupport.park();//暂停当前线程
LockSupport.unpark(Thread thread);//恢复某个线程的运行
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
static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(()->{
        logger.info("start...");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("park...");
        LockSupport.park();
        logger.info("resume...");
    });

    thread.start();
    Thread.sleep(2000);
    logger.info("unpark...");
    LockSupport.unpark(thread);
}

十二月 10, 2020 4:55:48 下午 Main lambda$main$0
信息: start...
十二月 10, 2020 4:55:49 下午 Main lambda$main$0
信息: park...
十二月 10, 2020 4:55:50 下午 Main main
信息: unpark...
十二月 10, 2020 4:55:50 下午 Main lambda$main$0
信息: resume...

死锁

产生死锁的必要条件：互斥条件、请求和保持条件、不剥夺条件、环路等待条件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#哲学家进餐问题
semaphore chopstick[5]={1,1,1,1,1};//五根筷子
semaphore mutex =  1;//设置取筷子的信号量
//i号哲学家进程
Pi(){
    do{
        wait(mutex);
        wait(chopstick[i]);//取左边筷子
        wait(chopstick[(i+1]%5);//取右边筷子
        signal(mutex);
        eat;
        signal(chopstick[i]);
        signal(chopstick[(i+1]%5);
        think;
    }
}

ReentrantLock

1
2
3
4
5
6
7
ReentrantLock reentrantLock = new ReentrantLock();
reentrantLock.lock();//获得锁
try {
    //临界区
}finally {
    reentrantLock.unlock();//释放锁
}

ReentrantLock 可重入

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
static Logger logger = Logger.getLogger("log");
private static ReentrantLock reentrantLock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {

    reentrantLock.lock();
    try {
        logger.info("start");
        enter();
    }finally {
        reentrantLock.unlock();
    }
}

public static void enter(){
    reentrantLock.lock();
    try {
        logger.info("enter");
        exit();
    }finally {
        reentrantLock.unlock();
    }
}

public static void exit(){
    reentrantLock.lock();
    try {
        logger.info("exit");
    }finally {
        reentrantLock.unlock();
    }
}

十二月 11, 2020 3:29:48 下午 Main main
信息: start
十二月 11, 2020 3:29:48 下午 Main enter
信息: enter
十二月 11, 2020 3:29:48 下午 Main exit
信息: exit

ReentrantLock 可中断

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
 static Logger logger = Logger.getLogger("log");
private static ReentrantLock reentrantLock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(()->{
        try {
            logger.info("尝试获得锁...");
            reentrantLock.lockInterruptibly();//可打断
        } catch (InterruptedException e) {
            logger.warning("被打断.。.");
            e.printStackTrace();
            return;
        }

        try {
            logger.info("获得锁...");
        }finally {
            reentrantLock.unlock();
        }
    },"thread");

    
    thread.start();
    thread.interrupt();
}

十二月 11, 2020 3:28:52 下午 Main lambda$main$0
信息: 尝试获得锁...
十二月 11, 2020 3:28:52 下午 Main lambda$main$0
警告: 被打断.。.
java.lang.InterruptedException
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1220)
	at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
	at Main.lambda$main$0(Main.java:21)
	at java.lang.Thread.run(Thread.java:748)

ReentrantLock 锁超时

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
static Logger logger = Logger.getLogger("log");
private static ReentrantLock reentrantLock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(()->{
        logger.info("尝试获得锁...");
        if(!reentrantLock.tryLock()){
            logger.info("获得锁失败");
            return;
        }

        try {
            logger.info("获得锁...");
        }finally {
            reentrantLock.unlock();
        }
    },"thread");

    reentrantLock.lock();//主线程先获得锁
    thread.start();
}
十二月 11, 2020 3:36:11 下午 Main lambda$main$0
信息: 尝试获得锁...
十二月 11, 2020 3:36:11 下午 Main lambda$main$0
信息: 获得锁失败

static Logger logger = Logger.getLogger("log");
private static ReentrantLock reentrantLock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
    Thread thread = new Thread(()->{
        logger.info("尝试获得锁...");
        try {
            if(!reentrantLock.tryLock(1, TimeUnit.SECONDS)){
                logger.info("获得锁失败");
                return;
            }
        } catch (InterruptedException e) {
            logger.info("获得锁失败");
            e.printStackTrace();
            return;
        }

        try {
            logger.info("获得锁...");
        }finally {
            reentrantLock.unlock();
        }
    },"thread");

    reentrantLock.lock();//主线程先获得锁
    thread.start();
}

十二月 11, 2020 3:38:11 下午 Main lambda$main$0
信息: 尝试获得锁...
十二月 11, 2020 3:38:12 下午 Main lambda$main$0
信息: 获得锁失败

ReentrantLock 条件变量

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
static Logger logger = Logger.getLogger("log");
static ReentrantLock reentrantLock = new ReentrantLock();
static Condition condition_s = reentrantLock.newCondition();
static Condition condition_g = reentrantLock.newCondition();
static boolean has_s = false;
static boolean has_g = false;
public static void main(String[] args) throws InterruptedException {
    new Thread(()->{
        reentrantLock.lock();
        try {
            logger.info("is there a s ?");
            while (!has_s){
                condition_s.await();//等待s条件
            }
            if(has_s){
                logger.info("get s");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }).start();

    new Thread(()->{
        reentrantLock.lock();
        try {
            logger.info("is there a g ?");
            while (!has_s){
                condition_g.await();//等待g条件
            }
            if(has_s){
                logger.info("get g");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }).start();

    Thread.sleep(2000);

    new Thread(()->{
        reentrantLock.lock();
        try {
            logger.info("add s");
            has_s = true;
            condition_s.signal();
        }finally {
            reentrantLock.unlock();
        }
    }).start();

    Thread.sleep(1000);
    new Thread(()->{
        reentrantLock.lock();
        try {
            logger.info("add g");
            has_g = true;
            condition_g.signal();
        }finally {
            reentrantLock.unlock();
        }
    }).start();

}
十二月 11, 2020 4:30:39 下午 Main lambda$main$0
信息: is there a s ?
十二月 11, 2020 4:30:39 下午 Main lambda$main$1
信息: is there a g ?
十二月 11, 2020 4:30:41 下午 Main lambda$main$2
信息: add s
十二月 11, 2020 4:30:41 下午 Main lambda$main$0
信息: get s
十二月 11, 2020 4:30:42 下午 Main lambda$main$3
信息: add g
十二月 11, 2020 4:30:42 下午 Main lambda$main$1
信息: get g

volatile 可见性

可见性：某个线程修改了共享变量，其他线程能够实时的知晓其改变

1.适用于一个写线程、多个读线程；禁止volatile共享变量前指令的重排序

2.共享变量易变，多个线程每次从主存中而非缓存中读取(读屏障)，并同步对共享变量的改动至主存(写屏障)

3.保证多个线程对其的可见性，防止某线程改变了其值，其他线程仍从缓存中读取旧值

4.synchronized只能保证完全交由synchronized保护的共享变量的原子性、有序性(单例模式-双重锁机制：instance共享变量需设为volatile，阻止指令重排序)

1
volatile static boolean flag = false;

happens-before原则

加锁
volatile变量
线程start()方法调用前的所有操作对新创建的线程可见
调用join()方法线程的操作对因其阻塞的线程可见
线程操作对其被调用interrupt的线程可见

无锁模式

适合线程数小且多核cpu场景

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class AccountCas{
    private AtomicInteger balance;

    public AccountCas(int balance){
        this.balance = new AtomicInteger(balance);
    }
    public int getBalance() {
        return balance.get();
    }

    public void withdraw(int amount){
        while (true){
            int prev = getBalance();
            int next = prev - amount;
            //prev和balance比较,查看是否有其他线程修改其值，被修改则进入下一次循环
            if(balance.compareAndSet(prev,next)){
                break;
            }
            //balance.getAndAdd(-1*amount);等价以上
        }
    }
}

原子整数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//AtomicInteger
AtomicInteger i = new AtomicInteger(0);
System.out.println("i:"+i.get());
System.out.println("++i:"+i.incrementAndGet());
System.out.println("--i:"+i.decrementAndGet());
System.out.println("i++:"+i.getAndIncrement());
System.out.println("i--:"+i.getAndDecrement());
System.out.println("i:"+i.get());

System.out.println("i+(2):"+i.addAndGet(2));

System.out.println("i * 10:"+i.updateAndGet(value->value * 10));

i:0
++i:1
--i:0
i++:0
i--:1
i:0
i+(2):2
i * 10:20

原子引用

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
class AccountCas{
    private AtomicReference<BigDecimal> balance;
    public AccountCas(BigDecimal balance){
        this.balance = new AtomicReference<BigDecimal>(balance);
    }
    public BigDecimal getBalance() {
        return balance.get();
    }

    public void withdraw(BigDecimal amount){
        while (true){
            BigDecimal prev = getBalance();
            BigDecimal next = prev.subtract(amount);
            if(balance.compareAndSet(prev,next)){
                break;
            }
        }
    }
}

//AtomicStampedReference
static Logger logger = Logger.getLogger("log");
static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A",0);
public static void main(String[] args) throws InterruptedException {
    logger.info("main start");
    String prev = ref.getReference();
    new Thread(()->{
        int stamp = ref.getStamp();
        logger.info("change A -> B");
        ref.compareAndSet(ref.getReference(),"B",stamp,stamp++);
        logger.info("ref.getReference():"+ref.getReference());
    }).start();
    Thread.sleep(1000);
    new Thread(()->{
        int stamp = ref.getStamp();
        logger.info("change B -> A");
        ref.compareAndSet(ref.getReference(),"A",stamp,stamp++);
        logger.info("ref.getReference():"+ref.getReference());
    }).start();

    Thread.sleep(1000);
    int stamp = ref.getStamp();
    ref.compareAndSet(prev,"C",stamp,stamp++);

    logger.info("ref.getReference():"+ref.getReference());
}

十二月 17, 2020 5:08:24 下午 Main main
信息: main start
十二月 17, 2020 5:08:24 下午 Main lambda$main$0
信息: change A -> B
十二月 17, 2020 5:08:24 下午 Main lambda$main$0
信息: ref.getReference():B
十二月 17, 2020 5:08:25 下午 Main lambda$main$1
信息: change B -> A
十二月 17, 2020 5:08:25 下午 Main lambda$main$1
信息: ref.getReference():A
十二月 17, 2020 5:08:26 下午 Main main
信息: ref.getReference():C

自定义线程池

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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
//拒绝策略
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockingQueue<T> queue,T task);
}

class ThreadPool{
    Logger logger = Logger.getLogger("log");
    private BlockingQueue<Runnable> taskQueue;//任务队列
    private HashSet<Worker> workers = new HashSet<>();
    private int coreSize;//线程数量
    private long timeout;//获取任务超时时间
    private TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize,long timeout,TimeUnit timeUnit,int queueCapacity,RejectPolicy<Runnable> rejectPolicy){
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }
    //执行任务
    public void execute(Runnable task){
        synchronized (workers){ //保证works集合的线程安全
            if(workers.size() < coreSize){
                Worker worker = new Worker(task);
                logger.info("创建worker:"+worker);
                workers.add(worker);
                worker.start();
            }
            else {
                logger.info("任务加入队列:"+task);
                //taskQueue.put(task);替换为自定义处理方式
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }
    class Worker extends Thread{
        private  Runnable task;
        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            //while (task != null || (task = taskQueue.take()) != null){
            while (task != null || (task = taskQueue.take(timeout,timeUnit)) != null){
                try {
                    logger.info("执行任务:"+task);
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            synchronized (workers){
                logger.info("worker被移除:"+this);
                workers.remove(this);
            }
        }
    }
}

//任务阻塞队列
class BlockingQueue<T>{
    private Deque<T> queue = new ArrayDeque<>(); //任务队列
    private ReentrantLock lock = new ReentrantLock();//队列任务锁
    private Condition fullWaitSet = lock.newCondition();//队列满
    private Condition emptyWaitSet = lock.newCondition();//队列空
    private int capacity;//队列容量

    public BlockingQueue(int queueCapacity) {
        this.capacity = queueCapacity;
    }

    public T take(){
        lock.lock();
        try{
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            T task = queue.removeFirst();
            fullWaitSet.signal();
            return task;
        } finally {
            lock.unlock();
        }
    };
    public T take(long timeout, TimeUnit timeUnit){
        lock.lock();
        try{
            long nanos = timeUnit.toNanos(timeout);
            while (queue.isEmpty()){
                try {
                    if(nanos <= 0) return null;
                    nanos = emptyWaitSet.awaitNanos(nanos);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            T task = queue.removeFirst();
            fullWaitSet.signal();
            return task;
        } finally {
            lock.unlock();
        }
    };

    public void put(T element){
        lock.lock();
        try {
            while (queue.size() == capacity){
                try{
                    fullWaitSet.await();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }
    //超时
    public boolean offer(T element,long timeout,TimeUnit timeUnit){
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capacity){
                try{
                    if(nanos <= 0) return false;
                    nanos = emptyWaitSet.awaitNanos(nanos);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
            return  true;
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            if (queue.size() == capacity){
                rejectPolicy.reject(this,task);
            }
            else {
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }


    public int size(){
        lock.lock();
        try {
            return capacity;
        }finally {
            lock.unlock();
        }

    }
}


static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    ThreadPool threadPool = new ThreadPool(2,1000,TimeUnit.MILLISECONDS,10,
            ((queue, task)->{
                //queue.put(task);//死等
                //queue.offer(task,5,TimeUnit.SECONDS);//超时等待
                //throw new RuntimeException("任务执行失败.");//抛出异常
                task.run(); //主线程执行任务
            }));
    for (int i = 0; i < 5; i++){
        int index = i;
        threadPool.execute(()->{
            logger.info(""+index);
        });
    }
}

java.util.concurrent

自定义锁

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
//自定义锁(不可重入锁)
class MyLock implements Lock{
    //同步器类 独占锁 AQS
    class MySync extends AbstractQueuedSynchronizer{
        @Override //尝试获得锁
        protected boolean tryAcquire(int arg) {
            if(compareAndSetState(0,1)){
                setExclusiveOwnerThread(Thread.currentThread());//设置锁的持有者线程
                return true;
            }
            return false;
        }

        @Override //尝试释放锁
        protected boolean tryRelease(int arg) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override //是否持有独占锁
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        public  Condition newCondition(){
            return new ConditionObject();
        }
    }

    private MySync sync = new MySync();

    @Override //加锁 不成功进入等待队列等待
    public void lock() {
        sync.acquire(1);
    }

    @Override //可打断的加锁
    public void lockInterruptibly() throws InterruptedException {
        sync.tryAcquire(1);
    }

    @Override //尝试加锁 仅尝试一次
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1,unit.toNanos(time));
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }
}

static Logger logger = Logger.getLogger("log");

public static void main(String[] args) throws InterruptedException {
	MyLock lock = new MyLock();
	new Thread(()->{
		lock.lock();
		try{
			logger.info("locking");
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			logger.info("unlocking");
			lock.unlock();
		}
	}).start();
	
	new Thread(()->{
		lock.lock();
		try{
			logger.info("locking");
		}
		finally {
			logger.info("unlocking");
			lock.unlock();
		}
	}).start();
}

一月 05, 2021 4:05:27 下午 Main lambda$main$0
信息: locking
一月 05, 2021 4:05:28 下午 Main lambda$main$0
信息: unlocking
一月 05, 2021 4:05:28 下午 Main lambda$main$1
信息: locking
一月 05, 2021 4:05:28 下午 Main lambda$main$1
信息: unlocking

读写锁

并发的读操作不互斥，并发读操作和写操作互斥。

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
class DataContainer{
    static Logger logger = Logger.getLogger("log");
    private Object data;
    private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    private ReentrantReadWriteLock.ReadLock readLock= rw.readLock();
    private ReentrantReadWriteLock.WriteLock writeLock= rw.writeLock();

    public Object read(){
        logger.info("get readlock...");
        readLock.lock();
        try {
            logger.info("reading...");
            return data;
        }finally {
            logger.info("release readlock...");
            readLock.unlock();
        }
    }

    public void write(){
        logger.info("get writelock...");
        writeLock.lock();
        try {
            logger.info("writing...");
        }finally {
            logger.info("release writelock...");
            writeLock.unlock();
        }
    }
}

static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    DataContainer dataContainer = new DataContainer();
    /*
    new Thread(()->{
        dataContainer.read();
    }).start();

    new Thread(()->{
        dataContainer.read();
    }).start();
    */

    new Thread(()->{
        dataContainer.write();
    }).start();

    new Thread(()->{
        dataContainer.read();
    }).start();
}

一月 06, 2021 11:29:39 上午 DataContainer write
信息: get writelock...
一月 06, 2021 11:29:39 上午 DataContainer read
信息: get readlock...
一月 06, 2021 11:29:40 上午 DataContainer write
信息: writing...
一月 06, 2021 11:29:40 上午 DataContainer write
信息: release writelock...
一月 06, 2021 11:29:40 上午 DataContainer read
信息: reading...
一月 06, 2021 11:29:40 上午 DataContainer read
信息: release readlock...

StampedLock

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
108
109
110
111
112
113
114
115
116
117
118
119
class DataContainer{

    static Logger logger = Logger.getLogger("log");
    private int data;
    private final StampedLock lock = new StampedLock();

    DataContainer(int data){
        this.data = data;
    }

    public int read(int readTime) throws InterruptedException {
        logger.info("tryOptimisticRead...");
        long stamp = lock.tryOptimisticRead();
        logger.info("stamp:"+stamp);
        Thread.sleep(readTime);
        if(lock.validate(stamp)){ //戳是否未被更新
            logger.info("read finish(OptimisticRead)...");
            return data;
        }
        logger.info("get readlock...");
        try {
            stamp = lock.readLock();
            logger.info("stamp:"+stamp);
            logger.info("reading...");
            Thread.sleep(readTime);
            logger.info("read finish...");
            return data;
        }finally {
            logger.info("release readlock...");
            lock.unlockRead(stamp);
        }
    }

    public void write(int data){
        logger.info("get writelock...");
        long stamp = lock.writeLock();
        logger.info("stamp:"+stamp);
        try {
            logger.info("writing...");
            this.data = data;
        }finally {
            logger.info("release writelock...");
            lock.unlockWrite(stamp);
        }
    }
}

static Logger logger = Logger.getLogger("log");

public static void main(String[] args) throws InterruptedException {
    DataContainer dataContainer = new DataContainer(0);
    new Thread(()->{
        try {
            dataContainer.read(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();
    Thread.sleep(1000);
    new Thread(()->{
        try {
            dataContainer.read(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();
}

一月 08, 2021 4:08:01 下午 DataContainer read
信息: tryOptimisticRead...
一月 08, 2021 4:08:01 下午 DataContainer read
信息: stamp:256
一月 08, 2021 4:08:02 下午 DataContainer read
信息: tryOptimisticRead...
一月 08, 2021 4:08:02 下午 DataContainer read
信息: stamp:256
一月 08, 2021 4:08:02 下午 DataContainer read
信息: read finish(OptimisticRead)...
一月 08, 2021 4:08:04 下午 DataContainer read
信息: read finish(OptimisticRead)...

static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    DataContainer dataContainer = new DataContainer(0);
    new Thread(()->{
        try {
            dataContainer.read(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }).start();

    new Thread(()->{
        dataContainer.write(10);
    }).start();
}

一月 08, 2021 5:22:44 下午 DataContainer read
信息: tryOptimisticRead...
一月 08, 2021 5:22:44 下午 DataContainer write
信息: get writelock...
一月 08, 2021 5:22:44 下午 DataContainer write
信息: stamp:384
一月 08, 2021 5:22:44 下午 DataContainer read
信息: stamp:256
一月 08, 2021 5:22:44 下午 DataContainer write
信息: writing...
一月 08, 2021 5:22:44 下午 DataContainer write
信息: release writelock...
一月 08, 2021 5:22:46 下午 DataContainer read
信息: get readlock...
一月 08, 2021 5:22:46 下午 DataContainer read
信息: stamp:513
一月 08, 2021 5:22:46 下午 DataContainer read
信息: reading...
一月 08, 2021 5:22:48 下午 DataContainer read
信息: read finish...
一月 08, 2021 5:22:48 下午 DataContainer read
信息: release readlock...

Semaphore 信号量

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
static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    Semaphore semaphore = new Semaphore(3);
    for (int i = 0; i < 10; i++) {
        new Thread(()->{
            try {
                semaphore.acquire();
                logger.info("running..");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                logger.info("end...");
                semaphore.release();
            }

        }).start();
    }
}

一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: running..
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: running..
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: end...
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: end...
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: running..
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: running..
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: end...
一月 11, 2021 9:44:32 上午 Main lambda$main$0
信息: end...

CountDownLatch

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
static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch =  new CountDownLatch(2);
    new Thread(()->{
        logger.info("running");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        countDownLatch.countDown();
        logger.info("end");
    }).start();
    new Thread(()->{
        logger.info("running");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        countDownLatch.countDown();
        logger.info("end");
    }).start();
    logger.info("main thread waiting");
    countDownLatch.await();//主线程等待其他所有线程执行完毕再恢复运行
    logger.info("main thread restart");
}

一月 11, 2021 4:29:53 下午 Main lambda$main$0
信息: running
一月 11, 2021 4:29:53 下午 Main lambda$main$1
信息: running
一月 11, 2021 4:29:53 下午 Main main
信息: main thread waiting
一月 11, 2021 4:29:54 下午 Main lambda$main$0
信息: end
一月 11, 2021 4:29:55 下午 Main lambda$main$1
信息: end
一月 11, 2021 4:29:55 下午 Main main
信息: main thread restart

CyclicBarrier

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
static Logger logger = Logger.getLogger("log");
public static void main(String[] args) throws InterruptedException {
    ExecutorService service = Executors.newFixedThreadPool(2);
    CyclicBarrier cyclicBarrier = new CyclicBarrier(2,()->{
        logger.info("all tasks finished");
    });

    service.submit(()->{
        logger.info("task1 running");
        try {
            Thread.sleep(1000);
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    });

    service.submit(()->{
        logger.info("task2 running");
        try {
            Thread.sleep(1000);
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    });
    service.shutdown();
}

一月 11, 2021 4:44:32 下午 Main lambda$main$1
信息: task1 running
一月 11, 2021 4:44:32 下午 Main lambda$main$2
信息: task2 running
一月 11, 2021 4:44:33 下午 Main lambda$main$0
信息: all tasks finished
Java
#并发
Java并发编程
https://jiaopaner.github.io/2020/12/02/Java并发编程/
作者
JiaoPan
发布于
2020年12月2日
许可协议