1.比较三种锁的效率

为什么LongAdder的效率更高
LongAdder和Atomic效率比sync高的原因是他们采用了无锁操作
Atomic底层用的CAS
LongAdder更高的原因是底层采用了分段锁

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

public class T12 {
    static long count2 = 0L;
    static AtomicLong count1= new AtomicLong(0L);
    static LongAdder count3 = new LongAdder();

    public static void main(String[] args)throws Exception {
        Thread[] threads = new Thread[1000];
        for(int i = 0;i<threads.length;i++){
            threads[i] = new Thread(()->{
                for(int k = 0;k<100000;k++) count1.incrementAndGet();
            });
        }
        long start = System.currentTimeMillis();
        for(Thread t:threads)t.start();
        for(Thread t:threads)t.join();
        long end = System.currentTimeMillis();
        System.out.println("Atomic:"+count1.get()+" time"+(end-start));
        //-------------------------------------------------------------
        Object lock = new Object();
        for(int i = 0;i<threads.length;i++){
            threads[i] =
                    new Thread(new Runnable(){
                        @Override
                        public void run(){
                            for(int k = 0;k<100000;k++)
                                synchronized (lock){
                                count2++;
                                }
                        }
                    });
        }
        start = System.currentTimeMillis();
        for(Thread t:threads)t.start();
        for(Thread t:threads)t.join();
        end = System.currentTimeMillis();
        System.out.println("Sync:"+count1.get()+" time"+(end-start));
        //-------------------------------------------------------------
        for(int i = 0;i<threads.length;i++){
            threads[i] =
                    new Thread(()->{
                        for(int k = 0;k<100000;k++)count3.increment();
                    });
        }
        start = System.currentTimeMillis();
        for(Thread t:threads)t.start();
        for(Thread t:threads)t.join();
        end = System.currentTimeMillis();
        System.out.println("LonAdder:"+count1.get()+" time"+(end-start));
    }
}

运行结果
Atomic:100000000 time1918
Sync:100000000 time3091
LonAdder:100000000 time229


间歇性复习

  • synchronized
  • volatile
  • atomicXXX ->CAS
  • increment -> sync atomicXXX LongAdder

2.证明锁可重入

import java.util.concurrent.TimeUnit;

public class T13 {
    synchronized void m1(){
        for(int i = 0;i<10;i++){
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(i);
            if(i==2) m2();
        }
    }
    Object o = new Object();
    synchronized void m2(){
        System.out.println("m2 ...");
    }

    public static void main(String[] args) {
        T13 t = new T13();
        new Thread(t::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(t::m2).start();
    }
}

3.Lock锁

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class T14 {
    Lock lock = new ReentrantLock();

    void m1() {
        try {
            lock.lock();  //synchronized(this)
            for (int i = 0; i < 10; i++) {
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock(); //解锁  synchronized是自动解锁的
        }
    }

    void m2() {
        boolean locked = false;
        try {
            locked = lock.tryLock(5, TimeUnit.SECONDS);
//            lock.lockInterruptibly();
            System.out.println("m2 ...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if(locked)
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        T14 t = new T14();
        new Thread(t::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(t::m2).start();
    }
}

4.公平锁

import java.util.concurrent.locks.ReentrantLock;

public class T15 extends Thread{
    private static ReentrantLock lock = new ReentrantLock(true); //参数为true表示公平锁,谁排在前面谁先执行
    public  void run(){
        for(int i = 0;i<100;i++){
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName()+"获得锁");
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) {
        T15 t = new T15();
        Thread th1 = new Thread(t);
        Thread th2 = new Thread(t);
        th1.start();
        th2.start();
    }
}

运行结果
Thread-1获得锁
Thread-2获得锁
Thread-1获得锁
Thread-2获得锁
Thread-1获得锁
... ...

ReentrantLock VS synchronized

  • cas vs sync
  • trylock
  • lockinterupptibly
  • 公平和非公平的切换

5.CountDownLatch

类似门栓

import java.util.concurrent.CountDownLatch;

public class T16 {
    public static void main(String[] args) {
        usingJoin();
        usingCountDownLatch();
    }
    private static void usingCountDownLatch(){
        Thread[]threads = new Thread[100];
        CountDownLatch latch = new CountDownLatch(threads.length);
        for(int i = 0;i<threads.length;i++){
            threads[i] = new Thread(()->{
                int result = 0;
                for(int j = 0;j<10000;j++) result+=j;
                latch.countDown();
            });
        }
        for (int i = 0;i<threads.length;i++){
            threads[i].start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("end latch");
    }
    private static void usingJoin(){
        Thread []threads = new Thread[100];
        for(int i = 0;i<threads.length;i++){
            threads[i] = new Thread(()->{
                int result = 0;
                for(int j = 0;j<10000;j++) result+=j;
            });
        }
        for (int i = 0;i<threads.length;i++){
            threads[i].start();
        }
        for (int i = 0;i<threads.length;i++){
            try {
                threads[i].join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("end join");
    }
}

6.CyclicBarrier

可以限流

  • 复杂操作

    • 1.数据库
    • 2.网络
    • 3.文件
  • 并发执行

    • 线程-操作
    • 线程-操作
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class T17 {
    public static void main(String[] args) {
//        CyclicBarrier barrier = new CyclicBarrier(20);
        CyclicBarrier barrier = new CyclicBarrier(20,new Runnable(){
            public void run(){
                System.out.println("满人,发车");
            }
        });
        for(int i = 0;i<100;i++){
            new Thread(()->{
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

运行结果
满人,发车
满人,发车
满人,发车
满人,发车
满人,发车

7.Phaser

package c_000;

import java.util.Random;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;

public class T09_TestPhaser2 {
    static Random r = new Random();
    static MarriagePhaser phaser = new MarriagePhaser();


    static void milliSleep(int milli) {
        try {
            TimeUnit.MILLISECONDS.sleep(milli);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {

        //定义总线程数量
        phaser.bulkRegister(7);

        for(int i=0; i<5; i++) {

            new Thread(new Person("p" + i)).start();
        }

        new Thread(new Person("新郎")).start();
        new Thread(new Person("新娘")).start();

    }



    static class MarriagePhaser extends Phaser {
        //当所有线程都到达后自动调用
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {

            switch (phase) {
                case 0:
                    System.out.println("所有人到齐了!" + registeredParties);
                    System.out.println();
                    return false;
                case 1:
                    System.out.println("所有人吃完了!" + registeredParties);
                    System.out.println();
                    return false;
                case 2:
                    System.out.println("所有人离开了!" + registeredParties);
                    System.out.println();
                    return false;
                case 3:
                    System.out.println("婚礼结束!新郎新娘抱抱!" + registeredParties);
                    return true;
                default:
                    return true;
            }
        }
    }


    static class Person implements Runnable {
        String name;

        public Person(String name) {
            this.name = name;
        }

        public void arrive() {

            milliSleep(r.nextInt(1000));
            System.out.printf("%s 到达现场!\n", name);
            phaser.arriveAndAwaitAdvance();
        }

        public void eat() {
            milliSleep(r.nextInt(1000));
            System.out.printf("%s 吃完!\n", name);
            phaser.arriveAndAwaitAdvance();
        }

        public void leave() {
            milliSleep(r.nextInt(1000));
            System.out.printf("%s 离开!\n", name);


            phaser.arriveAndAwaitAdvance();
        }

        private void hug() {
            if(name.equals("新郎") || name.equals("新娘")) {
                milliSleep(r.nextInt(1000));
                System.out.printf("%s 洞房!\n", name);
                phaser.arriveAndAwaitAdvance();  //线程等待
            } else {
                phaser.arriveAndDeregister();    //线程注销
                //phaser.register() //增加线程数量
            }
        }

        @Override
        public void run() {
            arrive();


            eat();


            leave();


            hug();

        }
    }
}

运行结果
p4 到达现场!
新娘 到达现场!
新郎 到达现场!
p1 到达现场!
p3 到达现场!
p2 到达现场!
p0 到达现场!
所有人到齐了!7

新郎 吃完!
新娘 吃完!

7.读写锁

读写锁其实就是共享锁(读锁)和排他锁(写锁)
读锁在执行时,允许其他的读锁进行读操作,写锁在执行时,不允许其他的锁进行

package c_000;

import java.util.Random;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class T10_TestReadWriteLock {
    static Lock lock = new ReentrantLock();
    private static int value;

    static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    static Lock readLock = readWriteLock.readLock();
    static Lock writeLock = readWriteLock.writeLock();

    public static void read(Lock lock) {
        try {
            lock.lock();
            Thread.sleep(1000);
            System.out.println("read over!");
            //模拟读取操作
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public static void write(Lock lock, int v) {
        try {
            lock.lock();
            Thread.sleep(1000);
            value = v;
            System.out.println("write over!");
            //模拟写操作
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }





    public static void main(String[] args) {
        //Runnable readR = ()-> read(lock);
        Runnable readR = ()-> read(readLock);

        //Runnable writeR = ()->write(lock, new Random().nextInt());
        Runnable writeR = ()->write(writeLock, new Random().nextInt());

        for(int i=0; i<18; i++) new Thread(readR).start();
        for(int i=0; i<2; i++) new Thread(writeR).start();


    }
}

8.semaphore

作用:限流

package c_000;

import java.util.concurrent.Semaphore;

public class T11_TestSemaphore {
    public static void main(String[] args) {
        //Semaphore s = new Semaphore(2);
        Semaphore s = new Semaphore(2, true);  //允许两个线程同时执行
        //允许一个线程同时执行
        //Semaphore s = new Semaphore(1);

        new Thread(()->{
            try {
                s.acquire();

                System.out.println("T1 running...");
                Thread.sleep(200);
                System.out.println("T1 running...");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                s.release();
            }
        }).start();

        new Thread(()->{
            try {
                s.acquire();

                System.out.println("T2 running...");
                Thread.sleep(200);
                System.out.println("T2 running...");

                s.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

9.Exchanger(交换锁)

import java.util.concurrent.Exchanger;

public class T12_TestExchanger {

    static Exchanger<String> exchanger = new Exchanger<>();

    public static void main(String[] args) {
        new Thread(()->{
            String s = "T1";
            try {
                s = exchanger.exchange(s);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " " + s);

        }, "t1").start();


        new Thread(()->{
            String s = "T2";
            try {
                s = exchanger.exchange(s);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " " + s);

        }, "t2").start();


    }
}

运行结果
t2 T1
t1 T2