typescript复制代码FutureTask<String> task = new FutureTask<String>(new Runnable() {
    @Override
 public void run() {
         //处理逻辑
 }
},"");

typescript复制代码FutureTask<String> callableTask = new FutureTask<String>(new Callable<String>() {

    @Override
 public String call() throws Exception {
        return "callable";
 }
});

arduino复制代码FutureTask<String> functionTask = new FutureTask<String>(()->{
    //处理逻辑
 return "demo";
});

java复制代码全局状态，用来判断当前任务处于那个阶段
    private volatile int state;
	处于新建状态
    private static final int NEW          = 0;
	任务处理完成中间状态，马上就可以获得结果
    private static final int COMPLETING   = 1;
	任务完成时的状态
    private static final int NORMAL       = 2;
	任务异常后的状态
    private static final int EXCEPTIONAL  = 3;
	任务被取消状态
    private static final int CANCELLED    = 4;
	中断时的中间状态
    private static final int INTERRUPTING = 5;
    中断状态
    private static final int INTERRUPTED  = 6;

    （具体执行任务的 接口带返回值的Callable接口
    private Callable<V> callable;
    任务结果：可能是正常结果，也可能是异常信息
    private Object outcome; // non-volatile, protected by state reads/writes
    执行callable的线程
    private volatile Thread runner;
    在等待的线程，是一个单链表
    private volatile WaitNode waiters;

    callable的构造器函数
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
	
	//基于runnable接口的构造函数，runnable没有返回值，所以需要传入一个结果类型。
    public FutureTask(Runnable runnable, V result) {
	// 调用适配器类Executors去返回一个callable接口的子类。
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }
	
	//Executors.callable,这里用到了适配器模式
	   public static <T> Callable<T> callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
			//创建了一个适配器类返回
        return new RunnableAdapter<T>(task, result); //适配器只需要实现callable接口即可
    }

java复制代码/**
     * 获取结果，如果任务没完成则调用awaitDone()方法等待任务完成,返回一个状态然后让report处理结果
     */
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
		//最终调用report返回结果
        return report(s);
    }

    /**
     * 指点时间返回
     */
    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING &&
            (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        return report(s);
    }

//主要是通过中断和超时等待结果或者正常完成或者终止。
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
		//获取处理时间
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
		
        WaitNode q = null;
		//判断节点是否在链表中
        boolean queued = false;
		//自旋
        for (;;) {
			//1.线程已经中断判断是否已经被中断，如果中断则移除该任务
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            //2.如果线程已经完成，则直接把当前waitNode置空，并且返回当前状态
            int s = state;
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            }
			//如果当前状态处于计算机结果中，则线程进行礼让(让出cpu时间片)
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
			//如果节点等null，则创建节点
            else if (q == null)
                q = new WaitNode();
			//如果发现不在链表中则添加到链表头部
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);
			//如果等待有时间限制，则判断是否已经超时，如果超时则移除节点,并返回状态
            else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
				//如果没超时则，设置一个有效时间，通过LockSupport把当前线程挂起.
                LockSupport.parkNanos(this, nanos);
            }
            else
				//挂起线程。
                LockSupport.park(this);
        }
    }

	
    //内部方法主要用来返回结果，它会判断当前状态是否是正常，如果正常直接返回，否则抛出一个异常
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

typescript复制代码//用于取消任务执行，是否支持中断取消
    public boolean cancel(boolean mayInterruptIfRunning) {
	//如果表示新建状态，或者通过CAS修改状态为中断或者取消状态失败，则直接返回取消失败
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    //如果是可中断处理的则获取执行callable的线程执行中断处理
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // 调用UNSAFE修改stat状态的偏移量为中断中断状态。
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
		//最后调用finishCompletion 处理结果。
            finishCompletion();
        }
        return true;
    }

    //判断是否以及取消
    public boolean isCancelled() {
        return state >= CANCELLED;
    }
    //判断是否以及完成任务，会发现其他几种除了中间状态其实都代表着任务已经完成了（可能是异常完成或者中断）
    public boolean isDone() {
        return state != NEW;
    }
	
	//任务执行完成或者取消后调用该方法收尾。
	private void finishCompletion() {
        // 这里主要是移除所以等待线程，等待GC回收
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
						//把等待线程唤醒。
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
      //这里调用了一个done方法这个方法是抽象方法主要是用于扩展用的，如果你在任务执行完成后还需要做一些其他操作则可以实现该接口处理，
	  //guava中ListenablFuture就是基于这个接口任务完成后回调done方法去执行监听器执行处理。
        done();
    
        callable = null;        // 置空等待回收
    }


    //抽象方法留给子类扩展使用，如果完成任务后需要做额外操作可以通过该接口扩展。
    protected void done() { }

typescript复制代码 //该方法是当线程执行完run方法之后会调用set方法使用cas修改状态、设置结果，处理善后工作.
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

    //处理异常结果的方法
    protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

   //实现了RunnableFuture接口的run方法，线程池内部执行是会调用该接口
   //这个接口干了啥？
   //1. 先判断当前线程的状态，如果表示new则证明已经被调用过
   // 
   //2. 如果没执行过，尝试把当前线程设置为执行这个任务的线程,并且 通过cas把执行线程修改成功则，继续执行，否则cas修改失败证明，这个执行任务线程已经被其他线程设置(已经有线程执行这个操作了)
   //修改成功后，执行 callable的call方法执行任务。
    public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {//如果异常则把异常信息处理到结果中
                    result = null;
                    ran = false;
                    setException(ex);
                }
				//如果正常完成着调用set方法把结果设置到全局outCome
                if (ran)
                    set(result);
            }
        } finally {
            //最后执行完后要置空，防止并非调用call或者run方法
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;//重新获取，可能有其他线程修改了，然后进行中断中间状态或者中断状态
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

	//处理来只取消的中断处理，如果处于中断中则调用线程礼让(方法)方法让出cpu
    private void handlePossibleCancellationInterrupt(int s) {

        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield(); // wait out pending interrupt
    }

//该方法是定时线程池中执行run方法会调用这个方法，定时任务中执行一次完成后需要把状态重新设置为new，为下一轮执行做准备。
    protected boolean runAndReset() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return false;
        boolean ran = false;
        int s = state;
        try {
            Callable<V> c = callable;
            if (c != null && s == NEW) {
                try {
                    c.call(); // don't set result
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;
    }

ini复制代码  /**
     * 单链表结构，用来记录一个堆栈中的等待线程
     */
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }



    //移除等待的线程节点
    private void removeWaiter(WaitNode node) {
        if (node != null) {
            node.thread = null;
            retry:
            for (;;) {          // restart on removeWaiter race
                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;
                    if (q.thread != null)
                        pred = q;
                    else if (pred != null) {
                        pred.next = s;
                        if (pred.thread == null) // check for race
                            continue retry;
                    }
                    else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                          q, s))
                        continue retry;
                }
                break;
            }
        }
    }

    // Unsafe 获取 任务状态stat ,已经当前任务执行的线程和等待的线程
    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = FutureTask.class;
            stateOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("state"));
            runnerOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("runner"));
            waitersOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("waiters"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

typescript复制代码  class A
    {
        public A()
        {
            Console.WriteLine("Create A Class");
        }
        ~A()
        {
            Console.WriteLine("Kill A Class");
        }
    }

css复制代码    A a = new A()

scss复制代码CHECK_ALLOC_AND_POSSIBLY_REGISTER_FOR_FINALIZATION(newAlloc, size, flags & GC_ALLOC_FINALIZE);


#define CHECK_ALLOC_AND_POSSIBLY_REGISTER_FOR_FINALIZATION(_object, _size, _register) do {  \
    if ((_object) == NULL || ((_register) && !REGISTER_FOR_FINALIZATION(_object, _size)))   \
    {                                                                                       \
        STRESS_LOG_OOM_STACK(_size);                                                        \
        return NULL;                                                                        \
    }                                                                                       \
} while (false)

scss复制代码#define REGISTER_FOR_FINALIZATION(_object, _size) \
    hp->finalize_queue->RegisterForFinalization (0, (_object), (_size))

ini复制代码bool CFinalize::Initialize()
{
    CONTRACTL {
        NOTHROW;
        GC_NOTRIGGER;
    } CONTRACTL_END;


    m_Array = new (nothrow)(Object*[100]);


    if (!m_Array)
    {
        ASSERT (m_Array);
        STRESS_LOG_OOM_STACK(sizeof(Object*[100]));
        if (GCConfig::GetBreakOnOOM())
        {
            GCToOSInterface::DebugBreak();
        }
        return false;
    }
    m_EndArray = &m_Array[100];


    for (int i =0; i < FreeList; i++)
    {
        SegQueueLimit (i) = m_Array;
    }
    m_PromotedCount = 0;
    lock = -1;
#ifdef _DEBUG
    lockowner_threadid.Clear();
#endif // _DEBUG


    return true;
}

scss复制代码CFinalize::RegisterForFinalization (int gen, Object* obj, size_t size)
{
    CONTRACTL {
        NOTHROW;
        GC_NOTRIGGER;
    } CONTRACTL_END;


    EnterFinalizeLock();


    // Adjust gen
    unsigned int dest = gen_segment (gen);


    // Adjust boundary for segments so that GC will keep objects alive.
    Object*** s_i = &SegQueue (FreeList);
    if ((*s_i) == m_EndArray)
    {
        if (!GrowArray())
        {
            LeaveFinalizeLock();
            if (method_table(obj) == NULL)
            {
                // If the object is uninitialized, a valid size should have been passed.
                assert (size >= Align (min_obj_size));
                dprintf (3, (ThreadStressLog::gcMakeUnusedArrayMsg(), (size_t)obj, (size_t)(obj+size)));
                ((CObjectHeader*)obj)->SetFree(size);
            }
            STRESS_LOG_OOM_STACK(0);
            if (GCConfig::GetBreakOnOOM())
            {
                GCToOSInterface::DebugBreak();
            }
            return false;
        }
    }
    Object*** end_si = &SegQueueLimit (dest);
    do
    {
        //is the segment empty?
        if (!(*s_i == *(s_i-1)))
        {
            //no, swap the end elements.
            *(*s_i) = *(*(s_i-1));
        }
        //increment the fill pointer
        (*s_i)++;
        //go to the next segment.
        s_i--;
    } while (s_i > end_si);


    // We have reached the destination segment
    // store the object
    **s_i = obj;
    // increment the fill pointer
    (*s_i)++;


    LeaveFinalizeLock();


    return true;
}

java复制代码//ReetrantLock的全局AQS子类.
private final Sync sync;



//我们看看Syn这个静态抽象类干了那几件事：
//1.首先是继承了AQS这个抽象同步器类
//2.实现了非公平锁尝试获取锁操作的nonfairTryAcquire方法，
//3.实现了tryRelease尝试释放锁方法 (ps:获取锁和释放锁操作都是模板方法的体现）
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;

    abstract void lock();

	// 首先获取当前线程，然后通过getState()方法获取全局state变量(ps：该变量是AQS的一个全局遍历，它可以判断当前锁被某个线程冲入几次)
	//然后通过cas修改这个state从0改为1,如果成功则该线程获取锁成功，把他设置为当前锁持有者线程
	//如果不成功，则判断当前锁持有者线程是否是自己，如果是者重入，修改state数量。
    final boolean nonfairTryAcquire(int acquires) {
		//获取当前线程，已经当前锁的状态state
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
			//如果0则还没有线程获得锁，尝试修改状态，如果成功把自己设置为锁持有者线程
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
		//判断当前锁持有者是否是自己，如果是进行重入操作
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
	//尝试释放锁方法，
	//1.计算需要释放释放个数与当前状态的差值，然后判断当前线程释放是锁持有者如果不是抛异常
	//2.如果差值等于0者释放成功把锁持有者线程置空，让出位置给下一个线程
	//3.最后修改state状态。
    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
    ...省略相关判断线程以及状态的简单方法
}

//非公平锁类，Sync的子类,实现了非公平锁加锁方法Lock的业务逻辑
//1.先通过cas修改状态修改成功者把当前线程设置为锁持有者线程
//2.调用acquire获取锁，该方法是AQS实现的
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    final void lock() {
		//cas 修改state从0-1如果成功，则获得锁，否则调用acquire方法进行后续尝试获取锁操作。
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
	//尝试获取锁方法，是由AQS的acquire方法调用，(ps：你可以发现模板方法者这里用得很多。）
	//然后调用父类Sync的nonfairTryAcquire非公平锁尝试获取方法
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

/**
 * 公平锁，公平锁不会先尝试修改state状态而是通过acquire去获取。
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        acquire(1);
    }
    //这个方法是公平锁尝试获取锁的方法
	//1.获取当前线程和锁状态state,判断队列里是否有等待线程，如果没有并且cas修改成功则获得锁，修改为持有者线程
	//2.判断当前锁持有者是否是自己，如果是进行重入操作
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

java复制代码//1.第一步是调用ReetrantLock的lock方法
public void lock() {
    sync.lock();
}
//第二步是调用Sync子类的lock方法
final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}
//第三步是调用AQS的acquire方法，先尝试获取锁，如果获取不到则生成一个waiter节点，然后
public final void acquire(int arg) {
//这个if，tryAcquire失败后，进行acquireqQueued去队列来从头节点的下一个节点开始重新tryAcquire获取如果成功则把当前节点中断，获得锁的线程执行。
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}




//第四步是调用NonfairSync的tryAcquire方法
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}


//第五步是调用Sync的nonfairTryAcquire方法,该部分源码前面已经介绍了。
final boolean nonfairTryAcquire(int acquires){}




//第六步如果还是失败则调用addWaiter
private Node addWaiter(Node mode) {
//创建当前线程创建成一个waiter节点
    Node node = new Node(Thread.currentThread(), mode);
    // 获取AQS队列的尾节点，然后把当前node 节点在尾部添加并且修改为节点为当前新node
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
//这里存在并发操作，可能为节点变为null，就修改失败
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
//cas修改为尾节点可能会失败，enq失败后初始化操作
    enq(node);
    return node;
}
//自旋，并且判断尾节点是否是null,如果是null，则当前没有等待节点则需要初始化一个节点作为头节点，该节点类似于哨兵作用。
//头节点是不干活的
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // 创建一个node作为头节点，进行初始化
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
//重新插入前面的node节点。
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
//第七步，调用AQS的acquireQueued，
//1. 节点添加到队列后，自旋，获取当前node节点的前一个节点的，
//2.如果当前的node的前一个节点是头节点，则证明队列里只有一个线程在等待，进行调用tryAcquire再次尝试获取锁，如果成功则把队列节点置空并且返回重新设置头节点
//3.获取失败，准备挂起，在挂起之前则调用shouldParkAfterFailedAcquire，主要是处理有一些等待节点已经被唤醒者需要移出队列的操作
//4.处理完后调用parkAndCheckInterrupt 进行挂起操作
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
//处理被唤醒的节点，和挂起当前线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
//
        if (failed)
            cancelAcquire(node);
    }
}
//处理已经被唤醒的节点，
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
      //一直while找到等待状态低于的节点然后把当前节点插入为下一个节点。
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
//调用LockSupport.park挂起当前线程
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

java复制代码第一步：先调用unlock
public void unlock() {
    sync.release(1);
}
第二步： 调用AQS的release方法，模板方法
public final boolean release(int arg) {
//调用Syn的tryRelease尝试获取锁。
    if (tryRelease(arg)) {
        Node h = head;
//如果头节点不是空而且状态不是0则尝试唤醒队列的后续节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
第三步，唤醒后续节点
private void unparkSuccessor(Node node) {
  
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
//一直遍历找到可以唤醒的jied
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
//通过unpark唤醒
    if (s != null)
        LockSupport.unpark(s.thread);
}

ini复制代码	以N=1234532为例：
	N = 1 2 3 4 5 3 2
	X  = 1 2 3 4 5 _ _

ini复制代码	N = 1 2 3 4 5 3 2
	X  = 1 2 3 4 4 _ _

ini复制代码	N = 3 3 3 2 1
	X = 3 3 2 9 9（之前做法得到的结果）

ini复制代码	N = 3 3 3 2 1
	X = 2 9 9 9 9

ini复制代码class Solution {
public:
    int monotoneIncreasingDigits(int N) {
        string str = to_string(N);
        int len = str.size();
        if (len == 1) return N;
        for (int i = 1;i < len;++i){
            if (str[i] < str[i-1]){//找到违反单调不减的下标
                int t = i;
                while (t>=1 && str[t-1]>str[t]){//向前检查
                    --str[t-1];
                    --t;
                }
                for (int k =t+1;k < len;++k)//t之后的数字取9
                    str[k] = '9';
                break;
            }
        }
        cout << str << endl;
        return atoi(str.c_str());
    }
};

python复制代码class Solution(object):
    def monotoneIncreasingDigits(self, N):
        """
        :type N: int
        :rtype: int
        """
        sn = str(N)
        size = len(sn)
        flag = False
        for x in range(size - 1):
            if sn[x] > sn[x + 1]:
                flag = True
                break
        if not flag: return N
        while x > 0 and sn[x - 1] == sn[x]: x -= 1
        y = len(sn) - x - 1
        return (N / (10 ** y)) * (10 ** y) - 1

shell复制代码replicaof  172.16.19.3  6379

shell复制代码# 命令操作类似：
replicatof 所选从库IP 6379

java复制代码@Aspect
public class PerformanceTraceAspect {
    private final Log logger = LogFactory.getLog(PerformanceTraceAspect.class);

    @Pointcut("execution(public void *.method1()) || execution(public void *.method2())")
    public void pt() {}

    @Around("pt()")
    public Object performanceTrace(ProceedingJoinPoint joinPoint) throws Throwable {
        StopWatch watch = new StopWatch();
        try {
            watch.start();
            return joinPoint.proceed();
        } finally {
            watch.stop();
            if (logger.isInfoEnabled()) {
                logger.info("PT in method[" + joinPoint.getSignature().getName() + "]>>>" + watch.toString());
            }
        }

    }
}

java复制代码AspectJProxyFactory weaver = new AspectJProxyFactory();
weaver.setProxyTargetClass(true);
weaver.setTarget(new Foo());
weaver.addAspect(PerformanceTraceAspect.class); //将切面添加进去
Object proxy = weaver.getProxy();

xml复制代码<aop:aspectj-autoproxy proxy-target-clas="true">
</aop:aspectj-autoproxy>

java复制代码@Pointcut("execution(public void *.method1()) || execution(public void *.method2())")
public void pt() {}

erlang复制代码before1
before2
task
after returning 2
after returning 1

java复制代码@Test
public void testRedis(){
    ValueOperations<String, String> operations = redisTemplate.opsForValue();
    operations.set("name", "tom");
    System.out.println(operations.get("name"));
}

java复制代码//Redis工具类
@Component
public final class RedisUtil{
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    //使用redisTemplate实现数据操作
    ...
}

java复制代码//String类型数据的放入和获取
public boolean set(String key, Object value) {
    try {
        redisTemplate.opsForValue().set(key, value);
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
public boolean set(String key, Object value, long time) {
    try {
        if (time > 0) {
            redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
        } else {
            set(key, value);
        }
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
public Object get(String key) {
    return key == null ? null : redisTemplate.opsForValue().get(key);
}

java复制代码//hash数据的放入、获取等操作
public boolean hset(String key, Map<String, Object> map) {
    try {
        redisTemplate.opsForHash().putAll(key, map);
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
public boolean hset(String key, Map<String, Object> map, long time) {
    try {
        redisTemplate.opsForHash().putAll(key, map);
        if (time > 0) {
            expire(key, time);
        }
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
public Map<Object, Object> hget(String key) {
    return redisTemplate.opsForHash().entries(key);
}
//针对hash中的指定key的操作
public boolean hItemSet(String key, String item, Object value) {
    try {
        redisTemplate.opsForHash().put(key, item, value);
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
public Object hItemGet(String key, String item) {
    return redisTemplate.opsForHash().get(key, item);
}
​
public boolean hItemHas(String key, String item) {
    return redisTemplate.opsForHash().hasKey(key, item);
}
public void hItemDel(String key, Object... item) {
    redisTemplate.opsForHash().delete(key, item);
}

java复制代码//设置指定key缓存有效期
public boolean expire(String key, long time) {
    try {
        if (time > 0) {
            redisTemplate.expire(key, time, TimeUnit.SECONDS);
        }
        return true;
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
​
//获取指定key的有效期
public long getExpire(String key) {
    return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
​
//判断指定key在缓存中是否存在
public boolean hasKey(String key) {
    try {
        return redisTemplate.hasKey(key);
    } catch (Exception e) {
        e.printStackTrace();
        return false;
    }
}
​
//删除指定key
@SuppressWarnings("unchecked")
public void del(String... key) {
    if (key != null && key.length > 0) {
        if (key.length == 1) {
            redisTemplate.delete(key[0]);
        } else {
            redisTemplate.delete(Arrays.asList(key));
        }
    }
}

Python复制代码async def app(scope, receive, send):
    # 一个最简单的ASGI应用程序
    assert scope['type'] == 'http'
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })


if __name__ == "__main__":
    # uvicorn服务
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")

Python复制代码from starlette.applications import Starlette
from starlette.middleware.gzip import GZipMiddleware


app: Starlette = Starlette()


@app.route("/")
def demo_route() -> None: pass


@app.websocket_route("/")
def demo_websocket_route() -> None: pass


@app.add_exception_handlers(404)
def not_found_route() -> None: pass


@app.on_event("startup")
def startup_event_demo() -> None: pass


@app.on_event("shutdown")
def shutdown_event_demo() -> None: pass


app.add_middleware(GZipMiddleware)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000)

Python复制代码class Starlette:
    def __init__(
        self,
        debug: bool = False,
        routes: typing.Sequence[BaseRoute] = None,
        middleware: typing.Sequence[Middleware] = None,
        exception_handlers: typing.Dict[
            typing.Union[int, typing.Type[Exception]], typing.Callable
        ] = None,
        on_startup: typing.Sequence[typing.Callable] = None,
        on_shutdown: typing.Sequence[typing.Callable] = None,
        lifespan: typing.Callable[["Starlette"], typing.AsyncGenerator] = None,
    ) -> None:
        """
        :param debug: 决定是否启用debug功能
        :param route: 一个路由列表, 提供HTTP和WebSocket服务.
        :param middleware: 中间件列表， 应用于每个请求
        :param exception_handler: 存放异常回调的字典， 键为HTTP状态码， 值为回调函数
        :on_startup: 启动时调用的回调函数
        :on_shutdown: 关闭时的回调函数
        :lifespan: ASGI中的lifespan功能
        """

        # 这里表示如果有传入lifespan， 则不可传入on_startup以及on_shutdown
        # 因为本质上starlette的通过把on_start_up和on_shutdown转为一个lifespan来接收uvicorn调用的
        assert lifespan is None or (
            on_startup is None and on_shutdown is None
        ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."

        # 初始化变量
        self._debug = debug
        self.state = State()
        self.router = Router(
            routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan
        )
        self.exception_handlers = (
            {} if exception_handlers is None else dict(exception_handlers)
        )
        self.user_middleware = [] if middleware is None else list(middleware)
        # 构建中间件
        self.middleware_stack = self.build_middleware_stack()

Python复制代码class Starlette:
    def build_middleware_stack(self) -> ASGIApp:
        debug = self.debug
        error_handler = None
        exception_handlers = {}

        # 解析异常处理的回调， 分别存放在error_handler和exception_handlers
        # 只有HTTP状态码为500的才会存入到error_handler
        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value

        # 为不同种类的中间件排好顺序
        # 第一层为ServerErrorMiddleware, 它能在发现异常的时候打印错误堆栈， 或者在debug模式的时候展示错误页面， 方便调试
        # 第二层是用户中间件层， 用户自己注册的所有中间件都会存放在这里
        # 第三层是ExceptionMiddleware, 它是异常处理层, 它会处理路由执行时抛出的所有异常
        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [
                Middleware(
                    ExceptionMiddleware, handlers=exception_handlers, debug=debug
                )
            ]
        )

        # 最后把中间件装填到app中
        app = self.router
        for cls, options in reversed(middleware):
            # cls是中间件类本身， options也就是我们传的参数
            # 可以看出中间件本身也是一个ASGIAPP， 装填中间件就是一个ASGI APP套上另外一个ASGI APP， 一直套娃。
            app = cls(app=app, **options)
        
        # 由于中间件的装填方式是不断的套娃， 而调用是不断的通过`call_next`调用装填它的上级ASGI APP， 所以要采用逆序的方法
        return app

Python复制代码class Starlette:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope["app"] = self
        await self.middleware_stack(scope, receive, send)

Python复制代码class BaseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        pass

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        pass

Python复制代码@app.app_exception_handlers(404)
def not_found_route() -> None: pass

Python复制代码class ExceptionMiddleware:
    def __init__(
        self, app: ASGIApp, handlers: dict = None, debug: bool = False
    ) -> None:
        self.app = app
        self.debug = debug  # TODO: We ought to handle 404 cases if debug is set.
        # startletter是支持HTTP状态码和Exception两种类型
        self._status_handlers = {}  # type: typing.Dict[int, typing.Callable]
        self._exception_handlers = {
            HTTPException: self.http_exception
        }  # type: typing.Dict[typing.Type[Exception], typing.Callable]
        if handlers is not None:
            for key, value in handlers.items():
                self.add_exception_handler(key, value)

    def add_exception_handler(
        self,
        exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
        handler: typing.Callable,
    ) -> None:
        # 用户通过调用startlette app的方法挂载的异常回调最后都是通过该方法挂载到类里面的_status_handlers或者是_exception_handler里面
        if isinstance(exc_class_or_status_code, int):
            self._status_handlers[exc_class_or_status_code] = handler
        else:
            assert issubclass(exc_class_or_status_code, Exception)
            self._exception_handlers[exc_class_or_status_code] = handler

    def _lookup_exception_handler(
        self, exc: Exception
    ) -> typing.Optional[typing.Callable]:
        # 查找注册异常相关的回调函数, 通过mro发现异常的对应回调函数
        # 
        # 用户挂载的可能是一个基类， 后续在遇到挂载异常的子类时， 也会调用基类注册的回调
        # 比如用户注册了一个基类， 然后会有用户异常和系统异常两个异常都继承于这个基类
        # 后续函数抛出用户异常或系统异常时， 都会执行到基类注册的对应回调
        for cls in type(exc).__mro__:
            if cls in self._exception_handlers:
                return self._exception_handlers[cls]
        return None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # 熟悉的ASGI 调用方法
        if scope["type"] != "http":
            # 不支持websocket请求
            await self.app(scope, receive, send)
            return

        # 防止同一个响应产生多个异常
        response_started = False

        async def sender(message: Message) -> None:
            nonlocal response_started

            if message["type"] == "http.response.start":
                response_started = True
            await send(message)

        try:
            # 调用下一个ASGI APP
            await self.app(scope, receive, sender)
        except Exception as exc:
            handler = None

            if isinstance(exc, HTTPException):
                # 如果是HTTPException异常， 则从注册HTTP回调字典中寻找
                handler = self._status_handlers.get(exc.status_code)

            if handler is None:
                # 如果是普通的异常， 则从异常回调字典去寻找
                handler = self._lookup_exception_handler(exc)

            if handler is None:
                # 找不到对应的异常， 则往上面抛
                raise exc from None

            # 一个响应只接收一次异常处理
            if response_started:
                msg = "Caught handled exception, but response already started."
                raise RuntimeError(msg) from exc

            request = Request(scope, receive=receive)
            if asyncio.iscoroutinefunction(handler):
                response = await handler(request, exc)
            else:
                response = await run_in_threadpool(handler, request, exc)
            # 通过回调函数生成的response处理请求
            await response(scope, receive, sender)

Python复制代码class DemoMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: ASGIApp,
    ) -> None:
        super(DemoMiddleware, self).__init__(app)

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # before
        response: Response = await call_next(request)
        # after
        return response

Python复制代码class BaseHTTPMiddleware:
    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:
        # 赋值下一级的ASGI app
        self.app = app
        # 如果用户有传dispatch， 就使用用户传的函数， 否则使用自身的dispatch
        # 一般用户都是继承于BaseHTTPMiddleware, 然后复写dispatch方法
        self.dispatch_func = self.dispatch if dispatch is None else dispatch

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        ASGI 标准的函数签名函数， 代表着ASGI的请求会从这里进来
        """
        if scope["type"] != "http":
            # 如果类型不是http的， 则不会走中间件(也就是websocket的不支持)
            # 要支持websocket的话， 中间件就很难这样实现了， 我在实现rap框架时， 为了支持类websocket的流量中间件处理， 牺牲了一些功能才可以实现
            await self.app(scope, receive, send)
            return

        # 通过scope生成request对象
        request = Request(scope, receive=receive)
        # 进入dispatch逻辑， 也就是用户的处理逻辑
        # 这个逻辑得到的respone实际上是call_next函数生成的， dispatch函数只做了传递的作用
        response = await self.dispatch_func(request, self.call_next)
        # 根据生成的response, 返回数据到上一层
        await response(scope, receive, send)

    async def call_next(self, request: Request) -> Response:
        loop = asyncio.get_event_loop()
        # 通过queue生产消费模式来获取下一级的消息
        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()

        scope = request.scope
        # 通过request.receive对象把uvicorn的receive对象传过来
        # 这里用到的receive对象还是uvicorn初始化时的receive对象
        receive = request.receive
        send = queue.put

        async def coro() -> None:
            try:
                await self.app(scope, receive, send)
            finally:
                # 这个put操作能确保get那边不会被卡死
                await queue.put(None)

        # 通过loop.create_task， 在另一个协程跑下一个ASGI APP
        task = loop.create_task(coro())
        # 等待下一个ASGI APP的返回
        message = await queue.get()
        if message is None:
            # 如果拿到是空的， 则代表下一个ASGI APP没有返回响应， 这时可能出错， 
            # 通过调用task.result()， 如果该协程出现异常, 则会把该协程的错误抛出来
            task.result()
            # 如果没有异常抛出来， 就可能是用户写错等原因， 返回了一个空响应, 
            # 这时候是没办法返回响应给客户端的， 需要自己制造一个异常， 方便后续生成一个500的响应
            raise RuntimeError("No response returned.")
        
        # ASGI处理响应的时候会分多步走， 正常情况下， 上面的queue.get， 是获取响应的第一步
        assert message["type"] == "http.response.start"

        async def body_stream() -> typing.AsyncGenerator[bytes, None]:
            # 其他的处理会交给body_stream函数处理
            # 这个方法所做的就是一直返回数据流
            while True:
                message = await queue.get()
                if message is None:
                    break
                assert message["type"] == "http.response.body"
                yield message.get("body", b"")
            task.result()

        # 将body_stream函数放到Response方法中
        # response本身也是一个类ASGI APP的类， 用户根据教程， 在dispatch方法中通过call_next获得response对象， 
        # 并在最后返回， 所以这个reponse对象将会交给__call__方法中进行处理。
        response = StreamingResponse(
            status_code=message["status"], content=body_stream()
        )
        response.raw_headers = message["headers"]
        return response

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        raise NotImplementedError()  # pragma: no cover

Python复制代码@app.exception_handlers(500)
def not_found_route() -> None: pass

Python复制代码class Router:
    def __init__(
        self,
        routes: typing.Sequence[BaseRoute] = None,
        redirect_slashes: bool = True,
        default: ASGIApp = None,
        on_startup: typing.Sequence[typing.Callable] = None,
        on_shutdown: typing.Sequence[typing.Callable] = None,
        lifespan: typing.Callable[[typing.Any], typing.AsyncGenerator] = None,
    ) -> None:
        # 装填starlette初始化时的信息
        self.routes = [] if routes is None else list(routes)
        self.redirect_slashes = redirect_slashes
        self.default = self.not_found if default is None else default
        self.on_startup = [] if on_startup is None else list(on_startup)
        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)

        async def default_lifespan(app: typing.Any) -> typing.AsyncGenerator:
            await self.startup()
            yield
            await self.shutdown()

        # 如果初始化lifespan为空， 则把on_startup和on_shuatdown转化为lifespan
        self.lifespan_context = default_lifespan if lifespan is None else lifespan

    async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
        """匹配不到路由执行的逻辑"""
        if scope["type"] == "websocket":
            # websocket匹配失败
            websocket_close = WebSocketClose()
            await websocket_close(scope, receive, send)
            return

        # If we're running inside a starlette application then raise an
        # exception, so that the configurable exception handler can deal with
        # returning the response. For plain ASGI apps, just return the response.
        if "app" in scope:
            # 在starlette.applications的__call__方法可以看到starlette把自身存入scope中
            # 这里抛出异常后， 可以被ServerErrorMiddleware捕获
            raise HTTPException(status_code=404)
        else:
            # 对于不是starlette调用的， 直接返回错误
            response = PlainTextResponse("Not Found", status_code=404)
        await response(scope, receive, send)

    async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Handle ASGI lifespan messages, which allows us to manage application
        startup and shutdown events.
        """
        # lifespan执行的逻辑， 在执行的时候starlette会与ASGI服务器进行通信， 但目前这样的代码估计还有一些待开发的功能
        first = True
        app = scope.get("app")
        await receive()
        try:
            if inspect.isasyncgenfunction(self.lifespan_context):
                async for item in self.lifespan_context(app):
                    assert first, "Lifespan context yielded multiple times."
                    first = False
                    await send({"type": "lifespan.startup.complete"})
                    await receive()
            else:
                for item in self.lifespan_context(app):  # type: ignore
                    assert first, "Lifespan context yielded multiple times."
                    first = False
                    await send({"type": "lifespan.startup.complete"})
                    await receive()
        except BaseException:
            if first:
                exc_text = traceback.format_exc()
                await send({"type": "lifespan.startup.failed", "message": exc_text})
            raise
        else:
            await send({"type": "lifespan.shutdown.complete"})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        The main entry point to the Router class.
        """
        # 匹配以及执行路由的主要函数
        # 目前只支持http, websocket, lifespan三种类型
        assert scope["type"] in ("http", "websocket", "lifespan")

        # 初始化router到scope中
        if "router" not in scope:
            scope["router"] = self

        if scope["type"] == "lifespan":
            # 执行lifespan逻辑
            await self.lifespan(scope, receive, send)
            return

        partial = None

        # 进行路由匹配
        for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                # 如果是完整匹配（url匹配， method匹配）
                # 则进行路由正常处理
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return
            elif match == Match.PARTIAL and partial is None:
                # 如果是不完整匹配（url匹配， method不匹配）
                # 则保留值， 继续匹配
                partial = route
                partial_scope = child_scope

        if partial is not None:
            # 如果存在不完整匹配的路由, 也继续执行， 但这时路由会响应HTTP 方法不对的错误
            scope.update(partial_scope)
            await partial.handle(scope, receive, send)
            return

        if scope["type"] == "http" and self.redirect_slashes and scope["path"] != "/":
            # 未匹配的情况, 判断重定向
            redirect_scope = dict(scope)
            if scope["path"].endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match != Match.NONE:
                    # 再次进行匹配, 如果结果不为空, 则发送重定向response
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return

        # 上面流程都没命中时， 则代表没有找到任务路由， 这时候会执行默认路由， 默认的默认路由是404 not found
        await self.default(scope, receive, send)

Python复制代码class BaseRoute:
    def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
        # 一个标准的匹配函数签名， 每个Route都要返回一个(Match, Scope)的元祖
        # Match有3种， 分别是
        #   NONE： 没有匹配到
        #   PARTIAL： 部分匹配（url匹配了， method匹配失败）
        #   FULL： 完全匹配（url和method都匹配成功）
        # Scope基本上都会返回如下格式, 不过Mount返回的内容更多：
        #   {"endpoint": self.endpoint, "path_params": path_params}
        raise NotImplementedError()  # pragma: no cover

    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        # 根据名字生成反向查找
        raise NotImplementedError()  # pragma: no cover

    async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
        # 被Router匹配后可以调用的函数
        raise NotImplementedError()  # pragma: no cover

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        A route may be used in isolation as a stand-alone ASGI app.
        This is a somewhat contrived case, as they'll almost always be used
        within a Router, but could be useful for some tooling and minimal apps.
        """
        # 如果该路由被当做ASGI APP单独调用， 则自己进行匹配并响应结果
        match, child_scope = self.matches(scope)
        if match == Match.NONE:
            if scope["type"] == "http":
                response = PlainTextResponse("Not Found", status_code=404)
                await response(scope, receive, send)
            elif scope["type"] == "websocket":
                websocket_close = WebSocketClose()
                await websocket_close(scope, receive, send)
            return

        scope.update(child_scope)
        await self.handle(scope, receive, send)

Python复制代码├── middleware                       # 中间件
├── applications.py                  # 启动的应用程序
├── authentication.py                # 验证相关
├── background.py                    # 封装后台任务， 会在返回响应后执行
├── concurrency.py                   # 一些小的asyncio相关的封装， 在新版本中， 直接使用了anyio库来代替
├── config.py                        # 配置
├── convertors.py                    # 一些类型的转换方法
├── datastructures.py                # 一些数据结构， 比如Url, Header, Form, QueryParam, State等等
├── endpoints.py                     # 支持cbv的路由以及一个稍微高级点的Websocket封装
├── exceptions.py                    # 异常处理
├── formparsers.py                   # Form，File之类的解析
├── graphql.py                       # 负责处理graphql相关的
├── __init__.py
├── py.typed                         # starlette需要用到的TypeHints
├── requests.py                      # 请求, 供用户获取数据
├── responses.py                     # 响应, 负责初始化Header和Cookies， 同时根据不同的Respnose类生成响应数据， 然后有个类ASGI调用接口， 该接口会发送ASGI协议到uvicorn服务， 发送完后如果有backgroud task, 则执行backgroud task， 直到执行完成， 该响应流程才结束。
├── routing.py                       # 路由
├── schemas.py                       # OpenApi相关的Schemas
├── staticfiles.py                   # 静态文件
├── status.py                        # HTTP状态码
├── templating.py                    # 基于jinja的模板响应
├── testclient.py                    # 测试客户端
├── types.py                         # 类型
└── websockets.py                    # websocket

Python复制代码async def stream(self) -> typing.AsyncGenerator[bytes, None]:
    # 如果已经读取过的， 则从缓存中获取数据
    if hasattr(self, "_body"):
        yield self._body
        yield b""
        return

    if self._stream_consumed:
        raise RuntimeError("Stream consumed")

    self._stream_consumed = True
    while True:
        # 从ASGI容器的receive循环获取数据
        message = await self._receive()
        if message["type"] == "http.request":
            body = message.get("body", b"")
            if body:
                # 获取的数据不为空就返回数据
                yield body
            if not message.get("more_body", False):
                # 代表body数据已经被获取完
                break
        elif message["type"] == "http.disconnect":
            # 代表与客户端的连接已经关闭了
            self._is_disconnected = True
            # 抛出异常， 用户调用await request.body()   await request.json()之类的会抛出异常
            raise ClientDisconnect()
    # 返回空字节，标记结束
    yield b""

Python复制代码import asyncio
from starlette.applications import Starlette
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

app: Starlette = Starlette()


class DemoMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        print(request, await request.body())
        return await call_next(request)


app.add_middleware(DemoMiddleware)


@app.route("/")
async def demo(request: Request) -> JSONResponse:
    try:
        await asyncio.wait_for(request.body(), 1)
        return JSONResponse({"result": True})
    except asyncio.TimeoutError:
        return JSONResponse({"result": False})


if __name__ == "__main__":
    import uvicorn  # type: ignore

    uvicorn.run(app)

Bash复制代码-> curl http://127.0.0.1:8000
{"result":false}

Python复制代码async def proxy_get_body(request: Request) -> bytes:
    async def receive() -> Message:
        return {"type": "http.request", "body": body}

    body = await request.body()
    request._receive = receive
    return body

Python复制代码from starlette.testclient import TestClient
from requests import Response  # type: ignore


app: Starlette = Starlette()
with TestClient(app) as client:
    response: Response = client.get("/")

Python复制代码response: Response = TestClient(app).get("/")