期货程序化交易ctp接口-深入理解AQSjdk层面管程实现

????优质资源分享????

???? Python实战微信订餐小程序 ????	进阶级	本课程是python flask+微信小程序的完美结合，从项目搭建到腾讯云部署上线，打造一个全栈订餐系统。
????Python量化交易实战????	入门级	手把手带你打造一个易扩展、更安全、效率更高的量化交易系统

什么是AQS

java.util.concurrent包中的大多数同步器实现都是围绕着共同的基础行为，比如等待队列、条件队列、独占获取、共享获取等，而这些行为的抽象就是基于AbstractQueuedSynchronizer实现的，AQS是一个抽象同步框架，可以用来实现一个依赖状态的同步器。

JDK中提供的大多数的同步器如Lock,Latch,Barrier等，都是基于AQS框架来实现的

一般是通过一个内部类Sync继承AQS

将同步器所有调用都映射到Sync对应的方法

AQS具备的特性：

阻塞等待队列，共享/独占，公平/非公平，可重入，允许中断

AQS定义两种资源共享方式

Exclusive-独占，只有一个线程能执行，如ReentrantLock

Share-共享，多个线程可以同时执行，如Semaphore/CountDownLatch

AQS定义两种队列

同步等待队列【主要用于维护获取锁失败时入队的线程】

AQS当中的同步等待队列也称CLH队列，CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列，是FIFO先进先出线程等待队列，Java中的CLH队列是原CLH队列的一个变种,线程由原自旋机制改为阻塞机制。

AQS依赖CLH同步队列来完成同步状态的管理：

当前线程如果获取同步状态失败时，AQS则会将当前线程已经等待状态等信息构造成一个节点并将其加入到CLH同步队列，同时会阻塞当前线程

当同步状态释放时，会把首节点唤醒，使其再次尝试获取同步状态。

通过signal或signalAll将条件队列中的节点转移到同步队列。

示：

条件等待队列【调用await的时候会释放锁，然后线程会加入到条件队列，调用signal唤醒的时候会把条件队列中的线程节点移动到同步队列中，等待再次获得锁】

AQS中条件队列是使用单向列表保存的，用nextWaiter来连接:

调用await方法阻塞线程；

当前线程存在于同步队列的头结点，调用await方法进行阻塞

AQS定义了5个队列中节点状态：

值为0，初始化状态，表示当前节点在sync队列中，等待着获取锁。

CANCELLED，值为表示当前的线程被取消；

SIGNAL，值为-表示当前节点的后继节点包含的线程需要运行，也就是unpark；

CONDITION，值为-表示当前节点在等待condition，也就是在condition队列中；

PROPAGATE，值为-表示当前场景下后续的acquireShared能够得以执行；

源码详解

属性值解析

//用链表来表示队列
private transient volatile Node head;
private transient volatile Node tail;

private volatile int state;  //可以表示锁的加锁状态【独占锁只为1，共享锁可以大于1】，又可以表示锁的重入次数，0为没有加锁

方法解析

//定义了主体的大体逻辑，如入队，如尝试加锁
private Node addWaiter(Node mode) {
 Node node = new Node(Thread.currentThread(), mode);
 // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
 if (pred != null) {
 node.prev = pred;
 if (compareAndSetTail(pred, node)) {
 pred.next = node;
 return node;
 }
 }
 enq(node);
 return node;
}

protected final boolean compareAndSetState(int expect, int update) {
 // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

//模板方法的处理，如果子类没有实现，则子类中调用的话会报错
//提供给子类去实现的公平与非公平的逻辑
protected boolean tryAcquire(int arg) {
 throw new UnsupportedOperationException();
}
//释放锁的逻辑
protected boolean tryRelease(int arg) {
 throw new UnsupportedOperationException();
}

Node类详解

代码展示

static final class Node {

 static final Node SHARED = new Node();  // 共享模式标记
    static final Node EXCLUSIVE = null;     // 独占模式标记

    static final int CANCELLED =  1;
 static final int SIGNAL    = -1;
 static final int CONDITION = -2;
 static final int PROPAGATE = -3;

 //值为0，初始化状态，表示当前节点在sync队列中，等待着获取锁。
 //CANCELLED，值为1，表示当前的线程被取消；
 //SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark；
 //CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中；
 //PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行；
    volatile int waitStatus;

 
 volatile Node prev;//前驱结点
    volatile Node next;//后继结点
    volatile Thread thread; //与节点绑定的线程
    Node nextWaiter; // 存储condition队列中的后继节点

    final boolean isShared() {
 return nextWaiter == SHARED;
 }

 final Node predecessor() throws NullPointerException {
 Node p = prev;
 if (p == null)
 throw new NullPointerException();
 else
            return p;
 }

 Node() {}

 Node(Thread thread, Node mode) { // Used by addWaiter
        this.nextWaiter = mode;
 this.thread = thread;
 }

 Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
 this.thread = thread;
 }
}

Condition接口详解

代码展示

//Condition用来替代synchronized锁的监视器的功能，而且更加灵活
//一个Condition实例需要与一个lock进行绑定
public interface Condition {
 //调用此方法的线程将加入等待队列，阻塞直到被通知或者线程中断
    void await() throws InterruptedException;

 //调用此方法的线程将加入等待队列，阻塞直到被通知（线程中断忽略）
    void awaitUninterruptibly();

 //调用此方法的线程将加入等待队列，阻塞直到被通知或者线程中断或等待超时
    long awaitNanos(long nanosTimeout) throws InterruptedException;

 //调用此方法的线程将加入等待队列，阻塞直到被通知或者线程中断或等待超时
    boolean await(long time, TimeUnit unit) throws InterruptedException;

 //调用此方法的线程将加入等待队列，阻塞直到被通知或者线程中断或超出指定日期
    boolean awaitUntil(Date deadline) throws InterruptedException;

 //唤醒一个等待中的线程
    void signal();

 //唤醒所以等待中的线程
    void signalAll();
}

发现说明

在Condition中，用await替换wait，用signal替换notify，用signalAll替换notifyAll，传统线程的通信方式，Condition都可以实现，这里注意，Condition是被绑定到Lock上的，要创建一个Lock的Condition必须用newCondition方法。Condition的强大之处在于，对于一个锁，我们可以为多个线程间建立不同的Condition。如果采用Object类中的wait,notify,notifyAll实现的话，当写入数据之后需要唤醒读线程时，不可能通过notify或notifyAll明确的指定唤醒读线程，而只能通过notifyAll唤醒所有线程，但是notifyAll无法区分唤醒的线程是读线程，还是写线程。通过Condition能够更加精细的控制多线程的休眠与唤醒。

但，condition的使用必须依赖于lock对象，通过lock对象的newCondition方法初始化一个condition对象。

ConditionObject类详解【Condition接口的实现类】

属性值解析

//由头尾两个节点指针形成的链表来达到队列的效果
private transient Node firstWaiter;
private transient Node lastWaiter;

方法解析

核心await方法

public final void await() throws InterruptedException {
 //如果线程中断，直接抛出异常 
    if (Thread.interrupted())
 throw new InterruptedException();

 //进入等待队列中
    Node node = addConditionWaiter();
 //释放当前线程持有的锁，并获取当前同步器状态
    int savedState = fullyRelease(node);
 int interruptMode = 0;
 //如果不在同步队列中，那么直接阻塞当前线程；直到被唤醒时，加入到同步队列中
    while (!isOnSyncQueue(node)) {
 LockSupport.park(this);
 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
 break;
 }
 //此时已经被唤醒，那么尝试获取锁
    if (acquireQueued(node, savedState) && interruptMode != THROW\_IE)
 interruptMode = REINTERRUPT;
 //如果节点中断取消，那么清除节点
    if (node.nextWaiter != null) // clean up if cancelled
 unlinkCancelledWaiters();
 if (interruptMode != 0)
 reportInterruptAfterWait(interruptMode);
}

//addConditionWaiter将一个节点添加到condition队列中。在入队时，判断当前尾节点是不是CONDITION。如果不是则判断当前尾节点已经被取消，将当前节点出队。那么也就是说在队列中的节点状态，要么是CONDITION,要么是CANCELLED
private Node addConditionWaiter() {
 Node t = lastWaiter;
 // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
 unlinkCancelledWaiters();
 t = lastWaiter;
 }
 Node node = new Node(Thread.currentThread(), Node.CONDITION);
 if (t == null)
 firstWaiter = node;
 else
 t.nextWaiter = node;
 lastWaiter = node;
 return node;
}

//方法的作用是移除取消的节点。方法本身只有在持有锁的时候会被调用。方法会遍历当前condition队列，将所有非Condition状态的节点移除。
private void unlinkCancelledWaiters() {
 Node t = firstWaiter;
 Node trail = null;
 while (t != null) {
 Node next = t.nextWaiter;
 if (t.waitStatus != Node.CONDITION) {
 t.nextWaiter = null;
 if (trail == null)
 firstWaiter = next;
 else
 trail.nextWaiter = next;
 if (next == null)
 lastWaiter = trail;
 }
 else
 trail = t;
 t = next;
 }
}

核心signal方法与signalAll方法

public final void signal() {
 if (!isHeldExclusively())
 throw new IllegalMonitorStateException();
 Node first = firstWaiter;
 if (first != null)
 doSignal(first);
}

public final void signalAll() {
 if (!isHeldExclusively())
 throw new IllegalMonitorStateException();
 Node first = firstWaiter;
 if (first != null)
 doSignalAll(first);
}

private void doSignal(Node first) {
 do {
 if ( (firstWaiter = first.nextWaiter) == null)
 lastWaiter = null;
 first.nextWaiter = null;
 } while (!transferForSignal(first) &&
 (first = firstWaiter) != null);
}

private void doSignalAll(Node first) {
 lastWaiter = firstWaiter = null;
 do {
 Node next = first.nextWaiter;
 first.nextWaiter = null;
 transferForSignal(first);
 first = next;
 } while (first != null);
}

final boolean transferForSignal(Node node) {
 //如果不能更改waitStatus，则表示该节点已被取消
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
 return false;

 Node p = enq(node);
 int ws = p.waitStatus;
 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
 LockSupport.unpark(node.thread);
 return true;
}

文章为作者独立观点，不代表股票程序化软件自动交易接口观点