枫秀学习日志

package JUC.tools;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**************************************
 *      Author : zhangke
 *      Date   : 2018/4/20 20:00
 *      Desc   : Semaphore 学习
 ***************************************/
public class SemaphoreTest1 {

    public static void main(String[] args) {
        Semaphore sem = new Semaphore(10);
        ExecutorService threadPool = Executors.newFixedThreadPool(3);

        //在线程池中执行任务
        threadPool.execute(new MyThread(sem, 5));
        threadPool.execute(new MyThread(sem, 4));
        threadPool.execute(new MyThread(sem, 7));

        //关闭池
        threadPool.shutdown();
    }


    static class MyThread extends Thread {
        private Semaphore sem;   //信号量
        private int count;   //申请信号量的大小


        public MyThread(Semaphore sem, int count) {
            this.sem = sem;
            this.count = count;
        }


        @Override
        public void run() {

            try {
                //从信号量中获取count个许可
                sem.acquire(count);
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() 
                                   + " acquire count=" + count);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 释放给定数目的许可，将其返回到信号量。
                sem.release(count);
                System.out.println(Thread.currentThread().getName() 
                                   + " release " + count + "");
            }
        }
    }
}

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

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

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) 
        throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 如果线程是中断状态，则抛出异常。
    if (Thread.interrupted())
        throw new InterruptedException();
    // 否则，尝试获取“共享锁”；获取成功则直接返回，
    // 获取失败，则通过doAcquireSharedInterruptibly()获取。
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

protected int tryAcquireShared(int acquires) {
    for (;;) {
        // 判断当前线程是不是CLH队列中的第一个线程线程，
        // 若是的话，则返回-1。
        if (hasQueuedPredecessors())
            return -1;
        // 设置可以获得的信号量的许可数
        int available = getState();
        // 设置获得acquires个信号量许可之后，剩余的信号量许可数
        int remaining = available - acquires;
        // 如果剩余的信号量许可数>=0，则设置可以获得的信号量许可数为remaining。
        // 设置成功则返回remaining
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

private void doAcquireSharedInterruptibly(long arg)
    throws InterruptedException {
    // 创建当前线程的Node节点，且Node中记录的锁是共享锁类型；
    // 并将该节点添加到CLH队列末尾。
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 获取上一个节点。
            // 如果上一节点是CLH队列的表头，则”尝试获取共享锁“。
            final Node p = node.predecessor();
            if (p == head) {
                long r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 当前线程一直等待，直到获取到共享锁。
            // 如果线程在等待过程中被中断过，则再次中断该线程(还原之前的中断状态)。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

public void release() {
    sync.releaseShared(1);
}
public void release(int permits) {
    if (permits < 0) 
    throw new IllegalArgumentException();
    sync.releaseShared(permits);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        // 获取“可以获得的信号量的许可数”
        int current = getState();
        // 获取“释放releases个信号量许可之后，剩余的信号量许可数”
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        // 设置“可以获得的信号量的许可数”为next。
        if (compareAndSetState(current, next))
            return true;
    }
}

private void doReleaseShared() {
    for (;;) {
        // 获取CLH队列的头节点
        Node h = head;
        // 如果头节点不为null，并且头节点不等于tail节点。
        if (h != null && h != tail) {
            // 获取头节点对应的线程的状态
            int ws = h.waitStatus;
            // 如果头节点对应的线程是SIGNAL状态，则意味着“头节点的下一个节点所对应的线程”需要被unpark唤醒。
            if (ws == Node.SIGNAL) {
                // 设置“头节点对应的线程状态”为空状态。失败的话，则继续循环。
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒“头节点的下一个节点所对应的线程”。
                unparkSuccessor(h);
            }
            // 如果头节点对应的线程是空状态，则设置“节点对应的线程所拥有的共享锁”为其它线程获取锁的空状态。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 如果头节点发生变化，则继续循环。否则，退出循环。
        if (h == head)                   // loop if head changed
            break;
    }
}

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        // 设置“可以获得的信号量的许可数”
        int available = getState();
        // 设置“获得acquires个信号量许可之后，剩余的信号量许可数”
        int remaining = available - acquires;
        // 如果“剩余的信号量许可数>=0”，则设置“可以获得的信号量许可数”为remaining。
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

// 内部使用的读锁
private final ReentrantReadWriteLock.ReadLock readerLock;
// 内部使用的写锁
private final ReentrantReadWriteLock.WriteLock writerLock;

// 读锁和写锁共同使用的锁类型，可以是公平锁和非公平锁
final Sync sync;

public ReentrantReadWriteLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

// ReadLock 类中
public void lock() {
    // 获取共享锁
    sync.acquireShared(1);
}

//AQS 类中
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

protected final int tryAcquireShared(int unused) {

    Thread current = Thread.currentThread();
    // 获取锁的状态
    int c = getState();
    // 如果锁被独占锁获取并且获取独占锁的线程不是当前线程，
    // 直接返回-1 达标获取锁失败
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;
    // 获取读锁的共享计数
    int r = sharedCount(c);
    // 判断不需要阻塞，并且已经获取读锁的数量小于MAX_COUNT
    // 则通过CAS函数更新读锁的状态，将读锁的共享计数加1
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
        // 第一次获取读锁
        if (r == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
          
        } else if (firstReader == current) {
              // 如果当前获取锁的线程是第一个获取读锁的线程
            firstReaderHoldCount++;
        } else {
            // HoldCounter是用来统计该线程获取“读取锁”的次数。
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                cachedHoldCounter = rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
             // 将该线程获取“读取锁”的次数+1。
            rh.count++;
        }
        return 1;
    }
    // 如果获取读锁失败，则通过下面函数来进行获取读锁
    return fullTryAcquireShared(current);
}

static final int SHARED_SHIFT   = 16;
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

// 返回共享锁的数量
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
// 返回独占说的数量
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

// 计数器对象，用于记录每个线程保持读锁的数量
// 这个对象被记录在ThreadLocal中，缓存在cachedHoldCounter
static final class HoldCounter {
    int count = 0;
    final long tid = getThreadId(Thread.currentThread());
}

// 自定的ThreadLocal对象，设置初始化方法
static final class ThreadLocalHoldCounter
   			 extends ThreadLocal<HoldCounter> {
    
    public HoldCounter initialValue() {
        return new HoldCounter();
    }
}
// 记录当前线程获取读锁的数量，在构造器中初始化，当保持的读锁数量为空的时候删除
private transient ThreadLocalHoldCounter readHolds;

// 用于记录上一个线程成功获取读锁的数量
private transient HoldCounter cachedHoldCounter;

// 下面来个一个是记录第一个获取读锁的线程和获取读锁的数量
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;

// 第一次获取读锁
if (r == 0) {
    firstReader = current;
    firstReaderHoldCount = 1;

} else if (firstReader == current) {
    // 如果当前获取锁的线程是第一个获取读锁的线程
    firstReaderHoldCount++;
} else {
    // HoldCounter是用来统计该线程获取“读取锁”的次数。
    HoldCounter rh = cachedHoldCounter;
    if (rh == null || rh.tid != getThreadId(current))
        cachedHoldCounter = rh = readHolds.get();
    else if (rh.count == 0)
        readHolds.set(rh);
    // 将该线程获取“读取锁”的次数+1。
    rh.count++;
}

final int fullTryAcquireShared(Thread current) {
    // 下面这部分代码和tryAcquireShared有一部分是重复的。
    // 但是tryAcquireShared只是先尝试获取，但是如果出现竞争则获取
    // 不到共享锁，即前面那部分加快锁的获取。下面这部分通过
    // 循环尝试，保证如果可以获取读锁，则一定获取到
    HoldCounter rh = null;
    for (;;) {
        // 获取锁的状态
        int c = getState();
        // 如果是独占锁，并且获取锁的线程不是current线程；则返回-1。
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
            
        // 如果需要阻塞等待。
        // 当需要阻塞等待的线程是第1个获取锁的线程的话，则继续往下执行。
        // 当需要阻塞等待的线程获取锁的次数为0时，则返回-1。
        } else if (readerShouldBlock()) {
           // 确保不是有一次获取读锁
            if (firstReader == current) {
                //忽略
            } else {
                // 获取当前线程获取读锁的数量，如果为0，则调用
                // ThreadLocal.remove删除这个ThreadLocal
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                // 如果当前线程获取读锁的计数=0,则返回-1。
                // 表示还没有获取过读锁，不在这里进行获取，
                // 则需要阻塞获取读锁的进程
                if (rh.count == 0)
                    return -1;
            }
        }
        // 则获取读取锁的共享统计数；
        // 如果共享统计数超过MAX_COUNT，则抛出异常。
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 将线程的读取锁次数加1
        // 这一步和上面一样，就不具体解释
        // 放在这里主要是因为CAS失败，如果失败，进入下一次循环
        if (compareAndSetState(c, c + SHARED_UNIT)) {       
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

private void doAcquireShared(int arg) {
    // 创建当前线程对应的节点，并将该线程添加到CLH队列中。
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 获取前继节点
            final Node p = node.predecessor();
            // 如果当前节点是头结点，尝试获取锁
            if (p == head) {
                int r = tryAcquireShared(arg);
                // 获取成功设置节点为可传播状态，
                // 然后释放后继获取读锁的节点
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 如果当前线程不是CLH队列的表头，
            // 则通过shouldParkAfterFailedAcquire()判断是否需要等待，
            // 需要的话，则通过parkAndCheckInterrupt()进行阻塞等待。
            // 若阻塞等待过程中，线程被中断过，则设置interrupted为true。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // 上面出现异常，则取消当前节点
        if (failed)
            cancelAcquire(node);
    }
}

public void unlock() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

protected final boolean tryReleaseShared(int unused) {
    
    // 获取当前线程，即释放共享锁的线程。
    Thread current = Thread.currentThread();
    
    // 如果想要释放锁的线程(current)是第1个获取锁(firstReader)的线程，
    // 并且第1个获取锁的线程获取锁的次数=1，则设置firstReader为null；
    // 否则，将第1个获取锁的线程的获取次数-1。
    if (firstReader == current) {
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    // 获取rh对象，并更新当前线程获取锁的信息。
    } else {
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != current.getId())
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            readHolds.remove();
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        --rh.count;
    }
    for (;;) {
        // 获取锁的状态
        int c = getState();
        // 将锁的获取次数-1。
        int nextc = c - SHARED_UNIT;
        // 通过CAS更新锁的状态。通过判断锁的状态是否为0来判断锁是否可以释放。
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

private void doReleaseShared() {
    for (;;) {
        // 获取CLH队列的头节点
        Node h = head;
       
        // 如果头节点不为null，并且头节点不等于tail节点。
        if (h != null && h != tail) {
            // 获取头节点对应的线程的状态
            int ws = h.waitStatus;
            
            // 如果头节点对应的线程是SIGNAL状态，
            // 则意味着头节点的下一个节点所对应的线程需要被unpark唤醒。
            if (ws == Node.SIGNAL) {
                // 设置头节点对应的线程状态为空状态。失败的话，则继续循环。
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒头节点的下一个节点所对应的线程。
                unparkSuccessor(h);
            }
            // 如果头节点对应的线程的状态值是0，
            // 则设置头结点的状态为PROPAGATE状态，等待有线程来获取锁才会结束循环。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 如果头节点没有改变，则继续循环。否则，退出循环。
        if (h == head)                   // loop if head changed
            break;
    }
}

final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

final boolean readerShouldBlock() {
    return apparentlyFirstQueuedIsExclusive();
}
final boolean apparentlyFirstQueuedIsExclusive() {
    Node h, s;
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}

package JUC.locks;

import lombok.Data;
import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**************************************
 *      Author : zhangke
 *      Date   : 2018/4/18 22:13
 *      Desc   : 读写锁
 ***************************************/
public class ReadWriteLockTest {

    public static void main(String[] args) {
        // 创建账户
        MyCount myCount = new MyCount("4238920615242830", 10000);

        // 创建用户，并指定账户
        User user = new User("Tommy", myCount);

        // 分别启动3个“读取账户金钱”的线程 和 3个“设置账户金钱”的线程
        for (int i = 0; i < 3; i++) {
            user.getCash();
            user.setCash((i + 1) * 1000);
        }
    }


    static class User {
        private String name;            //用户名

        private MyCount myCount;        //所要操作的账户

        private ReadWriteLock myLock;   //执行操作所需的锁对象


        public User(String name, MyCount myCount) {
            this.name = name;
            this.myCount = myCount;
            this.myLock = new ReentrantReadWriteLock();

        }


        public void getCash() {
            new Thread(() -> {
                try {
                    myLock.readLock().lock();
                    System.out.println(Thread.currentThread().getName() + " getCash start");
                    myCount.getCash();
                    Thread.sleep(1);
                    System.out.println(Thread.currentThread().getName() + " getCash end");
                } catch (InterruptedException e) {

                } finally {
                    myLock.readLock().unlock();
                }
            }).start();
        }


        public void setCash(final int cash) {
            new Thread(() -> {
                try {
                    Thread.sleep(100);
                    myLock.writeLock().lock();
                    System.out.println(Thread.currentThread().getName() 
                                       + " setCash start");
                    myCount.setCash(cash);
                    System.out.println(Thread.currentThread().getName() 
                                       + " setCash end");
                } catch (InterruptedException e) {

                } finally {
                    myLock.writeLock().unlock();
                }
            }).start();
        }
    }

    static class MyCount {

        @Getter
        @Setter
        private String id; //账户id

        private int cash; //现金


        public MyCount(String id, int cash) {
            this.id = id;
            this.cash = cash;
        }


        public int getCash() {
            System.out.println(Thread.currentThread().getName() + " getCash" +
                               " cash= " + cash);
            return cash;
        }


        public void setCash(int cash) {
            System.out.println(Thread.currentThread().getName() + " setCash" +
                               " cash= " + cash);
            this.cash = cash;
        }
    }
}