當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

HashedWheelTimer时间轮原理分析

發布時間：2024/4/11 编程问答 55 豆豆

生活随笔收集整理的這篇文章主要介紹了 HashedWheelTimer时间轮原理分析小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

HashedWheelTimer時間輪是一個高性能，低消耗的數據結構，它適合用非準實時，延遲的短平快任務，例如心跳檢測。

概要

時間輪是一種非常驚艷的數據結構。其在Linux內核中使用廣泛，是Linux內核定時器的實現方法和基礎之一。Netty內部基于時間輪實現了一個HashedWheelTimer來優化I/O超時的檢測，本文將詳細分析HashedWheelTimer的使用及原理。

背景

由于Netty動輒管理100w+的連接，每一個連接都會有很多超時任務。比如發送超時、心跳檢測間隔等，如果每一個定時任務都啟動一個Timer，不僅低效，而且會消耗大量的資源。
在Netty中的一個典型應用場景是判斷某個連接是否idle，如果idle（如客戶端由于網絡原因導致到服務器的心跳無法送達），則服務器會主動斷開連接，釋放資源。得益于Netty NIO的優異性能，基于Netty開發的服務器可以維持大量的長連接，單臺8核16G的云主機可以同時維持幾十萬長連接，及時掐掉不活躍的連接就顯得尤其重要。

看看官方文檔說明：

A optimized for approximated I/O timeout scheduling. You can increase
or decrease the accuracy of the execution timing by

specifying smaller or larger tick duration in the constructor. In most
network applications, I/O timeout does not need to be accurate.
Therefore, the default tick duration is 100 milliseconds and you will
not need to try different configurations in most cases.

這種方案也不是Netty憑空造出來的，而是根據George Varghese和Tony Lauck在1996年的論文實現的，有興趣的可以閱讀一下。
論文下載
論文PPT

應用場景

HashedWheelTimer本質是一種類似延遲任務隊列的實現，那么它的特點就是上述所說的，適用于對時效性不高的，可快速執行的，大量這樣的“小”任務，能夠做到高性能，低消耗。
例如：

心跳檢測
session、請求是否timeout
業務場景則有：
用戶下單后發短信
下單之后15分鐘，如果用戶不付款就自動取消訂單

簡單使用

如果之前沒用過，先看看用法有一個大體的感受，

@Slf4j public class HashedWheelTimerTest {private CountDownLatch countDownLatch = new CountDownLatch(2);@Testpublic void test1() throws Exception {//定義一個HashedWheelTimer，有16個格的輪子，每一秒走一個一個格子HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 16);//把任務加到HashedWheelTimer里，到了延遲的時間就會自動執行timer.newTimeout((timeout) -> {log.info("task1 execute");countDownLatch.countDown();}, 500, TimeUnit.MILLISECONDS);timer.newTimeout((timeout) -> {log.info("task2 execute");countDownLatch.countDown();}, 2, TimeUnit.SECONDS);countDownLatch.await();timer.stop();} }

需要引入netty-all.jar包
使用上跟ScheduledExecutorService差不多。

實現原理

源碼基于netty-all.4.1.34.Final

數據結構

時間輪其實就是一種環形的數據結構，可以想象成時鐘，分成很多格子，一個格子代表一段時間（這個時間越短，Timer的精度越高）。并用一個鏈表保存在該格子上的到期任務，同時一個指針隨著時間一格一格轉動，并執行相應格子中的到期任務。任務通過取模決定放入哪個格子。如下圖所示：

假設一個格子是1秒，則整個wheel能表示的時間段為8s，假如當前指針指向2，此時需要調度一個3s后執行的任務，顯然應該加入到(2+3=5)的方格中，指針再走3次就可以執行了；如果任務要在10s后執行，應該等指針走完一個round零2格再執行，因此應放入4，同時將round（1）保存到任務中。檢查到期任務時應當只執行round為0的，格子上其他任務的round應減1。

再回頭看看構造方法的三個參數分別代表

tickDuration
每一tick的時間
timeUnit
tickDuration的時間單位
ticksPerWheel
就是輪子一共有多個格子，即要多少個tick才能走完這個wheel一圈。

對于HashedWheelTimer的數據結構在介紹完源碼之后有圖解。

初始化

HashedWheelTimer整體代碼不難，慢慢看應該都可以看懂
我們從HashedWheelTimer的構造方法入手，先說明一下

構造方法

//附上文檔說明，自行閱讀 /*** Creates a new timer.** @param threadFactory a {@link ThreadFactory} that creates a* background {@link Thread} which is dedicated to* {@link TimerTask} execution.* @param tickDuration the duration between tick* @param unit the time unit of the {@code tickDuration}* @param ticksPerWheel the size of the wheel* @param leakDetection {@code true} if leak detection should be enabled always,* if false it will only be enabled if the worker thread is not* a daemon thread.* @param maxPendingTimeouts The maximum number of pending timeouts after which call to* {@code newTimeout} will result in* {@link java.util.concurrent.RejectedExecutionException}* being thrown. No maximum pending timeouts limit is assumed if* this value is 0 or negative.* @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}* @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0*/ //threadFactory默認是用Executors.defaultThreadFactory()，太懶了 //tickDuration，unit，ticksPerWheel核心參數，之前已經說過了 //leakDetection內存泄漏檢查 //maxPendingTimeouts準備執行的任務數，默認是-1，即不限制。如果并發量真的很高，可以設置一下，防止OOM public HashedWheelTimer(ThreadFactory threadFactory,long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,long maxPendingTimeouts) {if (threadFactory == null) {throw new NullPointerException("threadFactory");}if (unit == null) {throw new NullPointerException("unit");}if (tickDuration <= 0) {throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);}if (ticksPerWheel <= 0) {throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);}//初始化時間輪（1）wheel = createWheel(ticksPerWheel);mask = wheel.length - 1;this.tickDuration = unit.toNanos(tickDuration);//要求tickDuration * wheel.length < Long.MAX_VALUE，我猜測是因為擔心有類似的計算而導致溢出（但實際我沒找到）if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d",tickDuration, Long.MAX_VALUE / wheel.length));}//創建Work執行線程（2）workerThread = threadFactory.newThread(worker);//默認是啟動內存泄露檢測（我還不是很清楚具體原理）leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;this.maxPendingTimeouts = maxPendingTimeouts;//HashedWheelTimer實例數限制，因為HashedWheelTimer是一個非常消耗內存的對象，如果超過64個則會警告if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {reportTooManyInstances();} }

createWheel

private static HashedWheelBucket[] createWheel(int ticksPerWheel) {if (ticksPerWheel <= 0) {throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);}if (ticksPerWheel > 1073741824) {throw new IllegalArgumentException("ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);}//格子數向2的N次方數靠齊ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];for (int i = 0; i < wheel.length; i ++) {//初始化每一個bucketwheel[i] = new HashedWheelBucket();}return wheel; } //循環直到小于2的N次方 private static int normalizeTicksPerWheel(int ticksPerWheel) {int normalizedTicksPerWheel = 1;while (normalizedTicksPerWheel < ticksPerWheel) {normalizedTicksPerWheel <<= 1;}return normalizedTicksPerWheel; }

wheel其實是一個HashedWheelBucket數組

HashedWheelBucket

/** * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no * extra object creation is needed. */ private static final class HashedWheelBucket {// Used for the linked-list datastructureprivate HashedWheelTimeout head;private HashedWheelTimeout tail; }

bucket的結構是一個帶有頭節點指針和尾節點指針的linked-list

newTimeout

HashedWheelTimer初始化后，看看怎樣增加一個任務（在HashedWheelTimer內部統一叫HashedWheelTimeout，縮寫timeout，從名字已經可以看出HashedWheelTimer的作用）

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {if (task == null) {throw new NullPointerException("task");}if (unit == null) {throw new NullPointerException("unit");}//記錄待處理數，這個數字沒有實際作用，或可用于監控long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();//如果大于maxPendingTimeouts則報錯，默認-1，即不限制if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {pendingTimeouts.decrementAndGet();throw new RejectedExecutionException("Number of pending timeouts ("+ pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "+ "timeouts (" + maxPendingTimeouts + ")");}//（1）start();//計算這個timeout的執行時間，公式=當前時間 + 延遲時間 - wheelTimer的啟動時間，單位納秒，很直觀吧long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;//初始化timeoutHashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);//timeouts是一個MpscQueue隊列。這里注意了，新加入的timeout不是立即加到wheel中的，而是先加入到一個隊列，在tick的時候再從隊列取出來加入到wheel。//原因是，我猜測一是做緩沖作用，二是避免在插入timeout和執行timeout時有并發的沖突，特別是對linked-list的操作，如果采用加鎖的話，而執行該bucket所有timeout的時間不能保證，可能反而會阻塞到用戶。timeouts.add(timeout);return timeout; }

如果時間輪沒有啟動，那么就調用start方法啟動時間輪，啟動時間輪之后會為startTime設置為當前時間

計算延遲時間deadline

將task任務封裝到HashedWheelTimeout中，然后添加到timeouts隊列中進行緩存

//（1） //start方法為了保證wheelTimer在運行的狀態（如果是關閉狀態，那么直接拋異常出去），且wheelTimer已經初始化完成 //大家是不是很好奇wheelTimer的初始化為什么要在newTimeout，即加入第一個timeout后才去做。可以沿著延遲初始化或懶加載的思路去想，如果一個HashedWheelTimer初始化后一直沒有timeout加入，在那里空轉而白白浪費CPU資源就不好了。 public void start() {switch (WORKER_STATE_UPDATER.get(this)) {case WORKER_STATE_INIT:if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {//啟動WorkerworkerThread.start();}break;case WORKER_STATE_STARTED:break;case WORKER_STATE_SHUTDOWN:throw new IllegalStateException("cannot be started once stopped");default:throw new Error("Invalid WorkerState");}//因為Worker是異步執行的，會一直等待Worker的初始化while (startTime == 0) {try {//是一個CountDownLatchstartTimeInitialized.await();} catch (InterruptedException ignore) {// Ignore - it will be ready very soon.}} }

start方法會根據當前的workerState狀態來啟動時間輪。并且用了startTimeInitialized來控制線程的運行，如果workerThread沒有啟動起來，那么newTimeout方法會一直阻塞在運行start方法中。如果不阻塞，newTimeout方法會獲取不到startTime。

HashedWheelTimeout

timeout的結構

private static final class HashedWheelTimeout implements Timeout {//timeout的狀態有初始化，取消，已執行private static final int ST_INIT = 0;private static final int ST_CANCELLED = 1;private static final int ST_EXPIRED = 2;//用來支持CAS操作原子類private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");//所屬timerprivate final HashedWheelTimer timer;//需要執行的Runnableprivate final TimerTask task;//執行時間private final long deadline;@SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })private volatile int state = ST_INIT;//時間輪的層數long remainingRounds;//在linked-list中該節點指向的前后節點HashedWheelTimeout next;HashedWheelTimeout prev;// The bucket to which the timeout was addedHashedWheelBucket bucket;HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {this.timer = timer;this.task = task;this.deadline = deadline;} }

Worker run

一個HashedWheelTimer只有一個Worker線程。看看Worker的初始化

private final Worker worker = new Worker();

Worker繼承Runnable，我們看run方法

public void run() {//獲取啟動的時間作為開始時間startTime = System.nanoTime();if (startTime == 0) {// We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.startTime = 1;}// Notify the other threads waiting for the initialization at start().startTimeInitialized.countDown();do {//等待下一個tick（1）final long deadline = waitForNextTick();if (deadline > 0) {//找到該處理的bucket下標，因為wheel.length是2的N次方，mask為length - 1，所以這里相當于取模操作，性能比%高int idx = (int) (tick & mask);//處理掉已經取消的timeout（2）processCancelledTasks();//找到當前tick對應哪個bucketHashedWheelBucket bucket =wheel[idx];//把隊列的timeout放到wheel里（3）transferTimeoutsToBuckets();//處理當前bucket所有的timeout（4）bucket.expireTimeouts(deadline);//tick加一tick++;}//注意一下處理的順序，也講究的，先處理掉被取消的timeout，再把隊列的加進來，再處理，后面兩步不能反轉，因為有可能隊列里的timeout是下一tick執行的//循環直到wheelTimer被關閉} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);//wheelTimer被關閉后的處理//取出每一個bucket里還沒被執行的timeout，放到unprocessedTimeouts中for (HashedWheelBucket bucket: wheel) {bucket.clearTimeouts(unprocessedTimeouts);}//把隊列里的timeout放到unprocessedTimeouts中//PS：這個unprocessedTimeouts暫時只是做記錄用，做監控時或可用到for (;;) {HashedWheelTimeout timeout = timeouts.poll();if (timeout == null) {break;}if (!timeout.isCancelled()) {unprocessedTimeouts.add(timeout);}}//還要處理掉中間被取消的timeoutprocessCancelledTasks();}

時間輪運行的時候首先會記錄一下啟動時間（startTime），然后調用startTimeInitialized釋放外層的等待線程；

進入dowhile循環，調用waitForNextTick睡眠等待到下一次的tick指針的跳動，并返回當前時間減去startTime作為deadline

由于mask= wheel.length -1 ，wheel是2的次方數，所以可以直接用tick & mask 計算出此次在wheel中的槽位

調用processCancelledTasks將cancelledTimeouts隊列中的任務取出來，并將當前的任務從時間輪中移除

調用transferTimeoutsToBuckets方法將timeouts隊列中緩存的數據取出加入到時間輪中

運行目前指針指向的槽位中的bucket鏈表數據

時間輪指針跳動

//（1）private long waitForNextTick() {//計算下一個tick的時間，很簡單一看就懂long deadline = tickDuration * (tick + 1);for (;;) {final long currentTime = System.nanoTime() - startTime;//根據當前計算需要sleep的時間。這里加了999999是因為向上取整了1毫秒，假如距離下一個tick的時間為2000010納秒，那如果sleep 2毫秒是不夠的，所以需要多sleep 1毫秒。long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;//sleepTimeMs <=0 說明下一個tick的時間到了，說明上一個tick執行的時間“太久”了，所以直接返回就好了，不需要sleepif (sleepTimeMs <= 0) {//currentTime == Long.MIN_VALUE 這個判斷不是很理解if (currentTime == Long.MIN_VALUE) {return -Long.MAX_VALUE;} else {return currentTime;}}// Check if we run on windows, as if thats the case we will need// to round the sleepTime as workaround for a bug that only affect// the JVM if it runs on windows.//// See https://github.com/netty/netty/issues/356//這里是為了處理在windows系統上的一個bug，如果sleep不夠10ms則要取整if (PlatformDependent.isWindows()) {sleepTimeMs = sleepTimeMs / 10 * 10;}//直接sleep等待try {Thread.sleep(sleepTimeMs);} catch (InterruptedException ignored) {//Worker被中斷，如果是關閉了則返回負數，表示不會執行下一個tickif (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {return Long.MIN_VALUE;}}}}

在時鐘的秒鐘上面秒與秒之間的時間是需要等待的，那么waitForNextTick這個方法就是根據當前的時間計算出跳動到下個時間的間隔時間，并進行sleep操作，然后返回當前時間距離時間輪啟動時間的時間段。

//（2）private void processCancelledTasks() {for (;;) {//cancelledTimeouts也是一個MpscQueue，調用timeout的cancel方法會把timeout加進去//把取消的timeout取出來HashedWheelTimeout timeout = cancelledTimeouts.poll();if (timeout == null) {// all processedbreak;}try {//移除掉（2.1）timeout.remove();} catch (Throwable t) {if (logger.isWarnEnabled()) {logger.warn("An exception was thrown while process a cancellation task", t);}}}}//（2.1）void remove() {HashedWheelBucket bucket = this.bucket;if (bucket != null) {//調用bucket的remove方法（2.2）bucket.remove(this);} else {timer.pendingTimeouts.decrementAndGet();}}//（2.2）//類似鏈表刪除節點的操作public HashedWheelTimeout remove(HashedWheelTimeout timeout) {HashedWheelTimeout next = timeout.next;// remove timeout that was either processed or cancelled by updating the linked-listif (timeout.prev != null) {timeout.prev.next = next;}if (timeout.next != null) {timeout.next.prev = timeout.prev;}if (timeout == head) {// if timeout is also the tail we need to adjust the entry tooif (timeout == tail) {tail = null;head = null;} else {head = next;}} else if (timeout == tail) {// if the timeout is the tail modify the tail to be the prev node.tail = timeout.prev;}// null out prev, next and bucket to allow for GC.timeout.prev = null;timeout.next = null;timeout.bucket = null;timeout.timer.pendingTimeouts.decrementAndGet();return next;}

轉移任務到時間輪中

在調用時間輪的方法加入任務的時候并沒有直接加入到時間輪中，而是緩存到了timeouts隊列中，所以在運行的時候需要將timeouts隊列中的任務轉移到時間輪數據的鏈表中

//（3）private void transferTimeoutsToBuckets() {//最多取隊列的100000的元素出來for (int i = 0; i < 100000; i++) {HashedWheelTimeout timeout = timeouts.poll();if (timeout == null) {// all processedbreak;}//如果timeout被取消了則不做處理if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {// Was cancelled in the meantime.continue;}//計算位于實踐論的層數long calculated = timeout.deadline / tickDuration;timeout.remainingRounds = (calculated - tick) / wheel.length;//就是timeout已經到期了，也不能放到之前的tick中final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.//計算所在bucket下標，并放進去int stopIndex = (int) (ticks & mask);HashedWheelBucket bucket = wheel[stopIndex];//又是類似鏈表插入節點的操作bucket.addTimeout(timeout);}}

在這個轉移方法中，寫死了一個循環，每次都只轉移10萬個任務。

然后根據HashedWheelTimeout的deadline延遲時間計算出時間輪需要運行多少次才能運行當前的任務，如果當前的任務延遲時間大于時間輪跑一圈所需要的時間，那么就計算需要跑幾圈才能到這個任務運行。

最后計算出該任務在時間輪中的槽位，添加到時間輪的鏈表中。

運行時間輪中的任務

當指針跳到時間輪的槽位的時間，會將槽位的HashedWheelBucket取出來，然后遍歷鏈表，運行其中到期的任務。

//（4）public void expireTimeouts(long deadline) {HashedWheelTimeout timeout = head;//把bucket的所有timeout取出來執行while (timeout != null) {HashedWheelTimeout next = timeout.next;if (timeout.remainingRounds <= 0) {next = remove(timeout);if (timeout.deadline <= deadline) {//timeout的真正執行timeout.expire();} else {// The timeout was placed into a wrong slot. This should never happen.throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));}//該timeout被取消了則移除掉 } else if (timeout.isCancelled()) {next = remove(timeout);//否則層數減一，等待下一輪的到來} else {timeout.remainingRounds --;}timeout = next;}}

HashedWheelBucket是一個鏈表，所以我們需要從head節點往下進行遍歷。如果鏈表沒有遍歷到鏈表尾部那么就繼續往下遍歷。

獲取的timeout節點節點，如果剩余輪數remainingRounds大于0，那么就說明要到下一圈才能運行，所以將剩余輪數減一；

如果當前剩余輪數小于等于零了，那么就將當前節點從bucket鏈表中移除，并判斷一下當前的時間是否大于timeout的延遲時間，如果是則調用timeout的expire執行任務。

圖解

畫了個圖給大家體會一下

MpscQueue隊列

HashedWheelTimer用到的timeouts和cancelledTimeouts都是一種MpscQueue隊列的數據結構。
MpscQueue全稱Multi-Producer Single-Consumer Queue，從名字看出，是一種適合于多個生產者，單個消費者的高并發場景的高性能的，無鎖的隊列，原來Netty是自己實現了一個，但在最新的版本用了JCTools的，大家有興趣可以了解一下。

多層時間輪

當時間跨度很大時，提升單層時間輪的 tickDuration 可以減少空轉次數，但會導致時間精度變低，層級時間輪既可以避免精度降低，又避免了指針空轉的次數。如果有時間跨度較長的定時任務，則可以交給層級時間輪去調度。
設想一下一個定時了 3 天，10 小時，50 分，30 秒的定時任務，在 tickDuration = 1s 的單層時間輪中，需要經過：3246060+106060+5060+30 次指針的撥動才能被執行。但在 wheel1 tickDuration = 1 天，wheel2 tickDuration = 1 小時，wheel3 tickDuration = 1 分，wheel4 tickDuration = 1 秒的四層時間輪中，只需要經過 3+10+50+30 次指針的撥動。

如圖所示:

缺點

HashedWheelTimer也有一些缺點，在使用場景上要注意一下

Netty的HashedWheelTimer只支持單層的時間輪
當前一個任務執行時間過長的時候，會影響后續任務的到期

延遲任務方案對比

HashedWheelTimer本質上也是一個延遲隊列，我們跟其他延遲類解決方案對比一下

數據庫輪詢
比較常用的一種方法，數據先保存在數據庫中，然后啟動一個定時Job，根據時間或狀態把數據撈出來，處理后再更新回數據庫。這種方式很簡單，不會引入其他的技術，開發周期短。如果數據量比較大，千萬級甚至更多，插入頻率很高的話，上面的方式在性能上會出現一些問題，查找和更新對會占用很多時間，輪詢頻率高的話甚至會影響數據入庫。如果數據量進一步增大，那掃數據庫肯定就不行了。另一方面，對于訂單這類數據，我們也許會遇到分庫分表，那上述方案就會變得過于復雜，得不償失。
不過，優點是數據得到持久化，有問題可以查看。
DelayQueue
DelayQueue本質是PriorityQueue，每次插入或刪除任務都要調整堆，復雜度是O(logN)，相對HashedWheelTimer的O(1)來說有性能消耗。
ScheduledExecutorService
其本質也是類似DelayQueue，不過ScheduledExecutorService是多線程的方式執行，可以基本保證其他任務的準時進行。ScheduledExecutorService封裝較好，方便使用，還支持周期性任務。

總結