线程同步问题

22年的笔记了，JUC，线程管理存档

经典卖票问题

这似乎是非常完美的，当然由于Java的线程是抢占式的，所以输出的结果没有按照顺序来，所以结果是乱的。但是呢，现实生活中，出票也是需要时间的，所以我们就给线程增加一个睡眠时间去模拟这一情况。

lock解决

public class SellTicket {
    public static void main(String[] args) {
        //资源
        Ticket ticket = new Ticket();
        while(true){
            for(int i = 1; i <= 10; i++){
                new Thread(()->{ticket.sell();}, "窗口"+String.valueOf(i)).start();
            }
        }

    }
}

class Ticket{
    private int num = 100;
    Lock lock = new ReentrantLock();

    public void sell(){
        lock.lock();
        try{
            if(num > 0){
                //模拟出票时间
                try {
                    TimeUnit.MILLISECONDS.sleep(10L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"卖出票号为"+(num--));
            }
        }finally {
            lock.unlock();
        }
    }
}

synchronized 锁代码块，锁方法，都可以

synchronized适合锁少量的代码同步问题，Lock适合所大量的同步代码

经典同步问题

敖丙稳住了多线程翻车的现场

生产者消费者问题

相当于一个阻塞队列

lock+condition

condition.await()/signalAll()相当于 Object.wait()/notifyAll()

class Data{
   private int num = 0;
   Lock lock = new ReentrantLock();
   Condition condition = lock.newCondition();
   public void increment() throws InterruptedException {
       lock.lock();
       try{
           while(num != 0){
//               this.wait();
               condition.await();
           }
           num++;
           System.out.println(Thread.currentThread().getName()+"生产了"+num);
//           this.notifyAll();
           condition.signalAll();
       }finally{
           lock.unlock();
       }

   }
   public void decrement() throws InterruptedException {
       lock.lock();
       try{
           while(num == 0){
               condition.await();
           }
           num--;
           System.out.println(Thread.currentThread().getName()+"消费了"+num);
           condition.signalAll();
       }finally {
           lock.unlock();
       }

   }
}

8锁就是关于锁的八个问题

阻塞队列

四组API

方式	抛出异常	有返回值，不抛出异常	阻塞等待	超时等待
添加	add	offer()	put()	offer(a, 3)
移除	remove()	poll()	take()	poll(b, 3)
检查队首元素	element	peek	-	-

源码分析

阻塞队列详细介绍 JDK7提供了7个阻塞队列。 JDK提供的阻塞队列中，LinkedBlockingDeque 是一个 Deque（双向的队列），其实现的接口是 BlockingDeque；其余6个阻塞队列则是 Queue（单向队列），实现的接口是 BlockingQueue。

Java阻塞队列

实现的原理是ReentranLock + Condition 实现队列的阻塞，ReentranLock 是锁，Condition是条件状态，通过等待/通知机制，来实现线程之间的通信。

`ArrayBlockingQueue`

有返回值，不抛出异常，offer, poll


/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
            return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}

private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();//唤醒
}

private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();//唤醒
    return x;
}

阻塞等待==put，take

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();//1.
        return dequeue();
    } finally {
        lock.unlock();
    }
}
    public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();//2. 
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

LinkedBlockingDeque 双向阻塞队列，减少了一半竞争

SynchronousQueue 一个不存储元素的阻塞队列，长度为1

读者写者问题

哲学家问题

集合类不安全

List

public static void main(String[] args) {
    List<String> list = new ArrayList<>();
    for(int i = 0; i < 100; i++){
        new Thread(()->{
            list.add(UUID.randomUUID().toString().substring(0,5));
            System.out.println(list);
        }, String.valueOf(i)).start();
    }
}

问题：Exception in thread "37" java.util.ConcurrentModificationException 并发修改异常

快速失败 fail-fast ===modCount

每当迭代器使用hashNext()/next()遍历下一个元素之前，都会检测modCount变量是否为expectedmodCount值，是的话就返回遍历；否则抛出异常，终止遍历。

modCount

此列表被结构修改的次数。结构修改是指改变列表的大小，或者以某种方式扰乱列表，从而导致正在进行的迭代可能产生不正确的结果。

该字段由迭代器和listIterator方法返回的迭代器和列表迭代器实现使用。如果该字段的值意外更改，迭代器(或列表迭代器)将抛出ConcurrentModificationException异常，以响应next、remove、previous、set或add操作。这提供了快速故障行为，而不是在迭代期间面对并发修改时的非确定性行为。

子类是否使用此字段是可选的。如果一个子类希望提供快速失败迭代器(和列表迭代器)，那么它只需要在它的add(int, E)和remove(int)方法(以及它覆盖的导致列表结构修改的任何其他方法)中增加这个字段。单个调用add(int, E)或remove(int)必须向该字段添加不超过一个，否则迭代器(和列表迭代器)将抛出虚假的concurrentmodificationexception异常。如果实现不希望提供快速失败迭代器，则可以忽略此字段。

java.util包下的集合都是fail-fast的

java.util.concurrent下的集合是fail-safe（安全失败）的

解决方案：

List<String> list = Collections.synchronizedList(new ArrayList<String>());

对方法加synchronized

static class SynchronizedCollection<E> implements Collection<E>, Serializable {
        final Collection<E> c;  // Backing Collection
        final Object mutex;     // Object on which to synchronize
    
}

static class SynchronizedList<E> extends SynchronizedCollection<E> implements List<E> {
    public void add(int index, E element) {
        synchronized (mutex) {list.add(index, element);}
    }
    public E remove(int index) {
        synchronized (mutex) {return list.remove(index);}
    }
}

List<String> list = new CopyOnWriteArrayList<>();

//jdk11    
	public boolean add(E e) {
        synchronized (lock) {
            Object[] es = getArray();
            int len = es.length;
            es = Arrays.copyOf(es, len + 1);
            es[len] = e;
            setArray(es);
            return true;
        }
    }
//jdk8
//写入时复制，
    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

安装jdk 8，镜像地址

set

两种操作

1 2	Set<String> set1 = Collections.synchronizedSet(new HashSet<String>()); Set<String> set2 = new CopyOnWriteArraySet<>();

map

两种操作

1 2	Map<String, String> map1 = Collections.synchronizedMap(new HashMap<String, String>()); Map<String, String> map2 = new ConcurrentHashMap<>();

Hashtable也是对数据的操作方法加 synchronized

ConcurrentHashMap & Hashtable

synchronizedMap类

private final Map<K,V> m;     // Backing Map 普通对象map
final Object      mutex;        // Object on which to synchronize 排他锁

SynchronizedMap(Map<K,V> m) {
    this.m = Objects.requireNonNull(m);
    mutex = this;
}

ConcurrentHashMap

ConcurrentHashMap源码详解

ConcurrentHashMap源码分析(1.8)

jdk1.7和jdk1.8中做了很大的改动，从segment+reentranlock===>node+synchronized

ConcurrentHashMap在进行put操作的还是比较复杂的，大致可以分为以下步骤：

根据 key 计算出 hashcode 。

判断是否需要进行初始化。

即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。

如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。

如果都不满足，则利用 synchronized 锁写入数据。

如果数量大于 TREEIFY_THRESHOLD 则要转换为红黑树。

hashentry改为node，

因为ConcurrentHashMap已经将锁细化到了一个槽中了，冲突不会太大。
synchronized做了锁的升级机制优化，不会一下就挂起。
ReentrantLock的话，只要当前线程只要不是排在第二个等待节点就会被挂起，消耗资源。线程的挂起和唤醒会导致线程之间的切换，会频繁的导致内核态和用户态之间的切换，十分耗费资源。

//内部类，继承了可重入锁
    static class Segment<K,V> extends ReentrantLock implements Serializable {
        private static final long serialVersionUID = 2249069246763182397L;
        final float loadFactor;
        Segment(float lf) { this.loadFactor = lf; }
    }

创建线程的三种方式

Thread
Runnable
Callable

优缺点：

Thread编写相对简单，但是继承了Thread类就不能再继承其他类。

Runnable适合多个线程同时执行相同任务的情形，可以避免单继承带来的局限性；任务与线程本身是分离的提高了程序的健壮性，线程池接受Runable接口。

Callable可以继承其他的类，可以抛出异常，还可以有返回值。方法不同。

CountDownLatch 减法计数器等待线程都执行完再继续向下执行。

//count – the number of times countDown must be invoked before threads can pass through await    
	public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

	public void countDown() {
        sync.releaseShared(1);
    }
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
//使用示例
		//大爷锁门
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for(int i = 1; i <= 5; i++){
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"等待"+countDownLatch.getCount());
                countDownLatch.countDown();
            }).start();
        }
        countDownLatch.await();
        System.out.println("close");

CyclicBarrier 加法计数器

/*parties – the number of threads that must invoke await before the barrier is tripped;
barrierAction – the command to execute when the barrier is tripped, or null if there is no action*/
	public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

	public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
//使用示例
        //召唤神龙
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5, ()->{
            System.out.println("open");
        });
        for(int i = 1; i <= 5; i++){
            final int temp = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"召唤"+temp);
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }

Semaphore

信号量

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public void release() {
        sync.releaseShared(1);
    }

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
//使用示例
        //资源数量,抢车位
        Semaphore semaphore = new Semaphore(3);
        for(int i = 0; i < 5; i++){
            new Thread(()->{
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"抢到了车位");
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"离开了车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    semaphore.release();
                }
            }).start();
        }

线程池

java线程池使用最全详解

三大方法，七大参数，四种拒绝策略

降低资源消耗，提高相应速度。

阿里巴巴开发手册：线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式；

Executors返回线程池对象的弊端：

FixedThreadPool和SingleThreadPool：可能会堆积大量的请求，从而导致OOM
CachedThreadPool和ScheduledThreadPool：可能会创建大量的进程，从而导致OOM

三大方法

1
2
3

ExecutorService threadPool = Executors.newSingleThreadExecutor();
ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
ExecutorService threadPool2 = Executors.newCachedThreadPool();

七大参数

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {}

四种拒绝策略

AbortPolicy; CallerRunsPolicy; DiscardPolicy; DiscardOldestPolicy


 /**
  * A handler for rejected tasks that throws a
  * {@code RejectedExecutionException}.
  */	  
public static class AbortPolicy implements RejectedExecutionHandler {
     public AbortPolicy() { }
     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
         throw new RejectedExecutionException("Task " + r.toString() +
                                              " rejected from " +
                                              e.toString());
     }
 }    

 /**
  * A handler for rejected tasks that runs the rejected task
  * directly in the **calling thread** of the {@code execute} method,
  * unless the executor has been shut down, in which case the task
  * is discarded.
  */
 public static class CallerRunsPolicy implements RejectedExecutionHandler {
     public CallerRunsPolicy() { }
     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
         if (!e.isShutdown()) {
             r.run();
         }
     }
 }
 
 /**
  * A handler for rejected tasks that silently discards the
  * rejected task.
  */
 public static class DiscardPolicy implements RejectedExecutionHandler {
     public DiscardPolicy() { }
     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
     }
 }


 /**
  * A handler for rejected tasks that discards the oldest unhandled
  * request and then retries {@code execute}, unless the executor
  * is shut down, in which case the task is discarded.
  */
 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
     public DiscardOldestPolicy() { }
     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
         if (!e.isShutdown()) {
             e.getQueue().poll();
             e.execute(r);
         }
     }
 }

Java8新特性：函数式接口

函数式接口；lambda表达式（语法糖）；链式编程；Stream流计算。

函数式接口(Functional Interface)就是一个有且仅有一个抽象方法，但是可以有多个非抽象方法的接口。

函数式接口可以被隐式转换为 lambda 表达式。

Consumer Supplier Predicate Function

接口	描述
Supplier	无参数，返回一个结果。
Consumer	代表了接受一个输入参数并且无返回的操作
Function<T,R>	接受一个输入参数，返回一个结果。
Predicate	接受一个输入参数，返回一个布尔值结果。

JMM

线程安全的实现方法

阻塞同步

临界区，互斥量（mutex）信号量

synchronized 和lock

非阻塞同步

乐观锁。ABA问题。

AtomicInteger

整数原子类 compareAndSet() 和 getAndIncrement() 都使用了sun.misc.Unsafe类中的CAS操作。

public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

无同步方案

可重入代码

线程本地存储 ThreadLocal