數(shù)據(jù)結(jié)構(gòu)與算法--關(guān)鍵路徑

關(guān)鍵路徑與無(wú)環(huán)加權(quán)有向圖的最長(zhǎng)路徑

現(xiàn)在考慮一個(gè)這樣的問(wèn)題：你今天事情比較多沐批，要洗衣服、做作業(yè)還要燒水洗澡豺总，之后出去找朋友玩匀归。假設(shè)洗衣服要20分鐘，燒水要30分鐘运敢，做作業(yè)的話你把朋友做好的帶回來(lái)抄校仑，只需要10分鐘。你想能早些去找朋友传惠，但在那之前又必須將那些事做完迄沫，你要怎么安排呢？很容易想到卦方，這三者同時(shí)進(jìn)行：打好水開始燒水羊瘩，衣服扔進(jìn)洗衣機(jī)，回書桌抄作業(yè)...20分鐘后作業(yè)寫完了盼砍，衣服也洗好了尘吗，水還有10分鐘水才燒開，利用這時(shí)間把洗好的衣服晾曬好浇坐，差不多水也燒開了睬捶，好了最后去洗澡。簡(jiǎn)直一氣呵成近刘，這是我們能花費(fèi)的最少時(shí)間了擒贸，在這個(gè)例子中剛好等于所有任務(wù)中持續(xù)時(shí)間最長(zhǎng)的那個(gè)臀晃。（你做完了作業(yè)才想起來(lái)去燒水，花費(fèi)不止半小時(shí)吧）

由此引申出一個(gè)更為廣泛的問(wèn)題介劫，給定一組需要完成的任務(wù)和每個(gè)任務(wù)所需的時(shí)間徽惋，以及一組關(guān)于任務(wù)完成的先后次序的優(yōu)先級(jí)限制。在滿足限制條件的前提下應(yīng)該如何在若干相同的處理器上（數(shù)量不限座韵，可并行處理多個(gè)任務(wù)）安排任務(wù)并在最短的時(shí)間內(nèi)完成所有的任務(wù)寂曹？

此問(wèn)題的提出主要是為了解決并行任務(wù)調(diào)度，使得完成所有任務(wù)的總時(shí)間最短回右。待處理的任務(wù)總數(shù)可能成百上千，因此需要一個(gè)算法幫我們快速規(guī)劃一個(gè)調(diào)度方案：按照怎樣的順序執(zhí)行這些任務(wù)漱挚，哪些任務(wù)可以同時(shí)處理翔烁，如何使得耗費(fèi)的總時(shí)間最短？正好存在一種叫做“關(guān)鍵路徑”的方法可以證明這個(gè)問(wèn)題與無(wú)環(huán)加權(quán)有向圖的最長(zhǎng)路徑問(wèn)題等價(jià)旨涝。

關(guān)鍵路徑：把路徑上各個(gè)任務(wù)所持續(xù)的時(shí)間之和稱為路徑長(zhǎng)度蹬屹，從起點(diǎn)到終點(diǎn)的所有路徑中，具有最長(zhǎng)路徑長(zhǎng)度的路徑稱為關(guān)鍵路徑白华，關(guān)鍵路徑中的各個(gè)任務(wù)稱為關(guān)鍵任務(wù)慨默。上面的例子中，燒水就是個(gè)關(guān)鍵任務(wù)弧腥。

首先厦取，按照關(guān)鍵路徑的順序執(zhí)行任務(wù)，一定能保證所有的任務(wù)都能完成管搪，且此時(shí)花費(fèi)的總時(shí)間最短虾攻。有些活動(dòng)順序進(jìn)行，有些活動(dòng)并行進(jìn)行更鲁。從起點(diǎn)到各個(gè)頂點(diǎn)霎箍，以至從起點(diǎn)到終點(diǎn)的有向路徑可能不止一條，這些路徑的長(zhǎng)度也不盡相同澡为。這若干條從起點(diǎn)到終點(diǎn)的路徑可以看做一個(gè)生產(chǎn)過(guò)程的幾條不同的生產(chǎn)線漂坏，必須每條生產(chǎn)線都完工，整個(gè)生產(chǎn)過(guò)程才算結(jié)束媒至，也就是不論如何你都得等那條花費(fèi)時(shí)間最長(zhǎng)的流水線做完顶别，整個(gè)生產(chǎn)才可能完工。現(xiàn)在由于可以同時(shí)處理多個(gè)任務(wù)塘慕，在花費(fèi)時(shí)間最長(zhǎng)的流水線工作過(guò)程中筋夏，其他流水線一定會(huì)提前完工（最多也就是同時(shí)），因此花費(fèi)時(shí)間最長(zhǎng)的流水線做完后图呢，整個(gè)工程也隨之竣工了条篷。假設(shè)花費(fèi)時(shí)間最長(zhǎng)的那條流水線所用的時(shí)間是M骗随，這就是說(shuō)，不管怎么安排赴叹，都需要至少M(fèi)的時(shí)間才能竣工鸿染，而這已經(jīng)是最短時(shí)間了。

再舉個(gè)例子乞巧，你和朋友們約好去某個(gè)地方聚餐涨椒。有些朋友到的比較早，有些朋友到得比較晚绽媒，但是不管怎么樣蚕冬，我們都要等到最后一個(gè)朋友到目的地，這樣大家才算是聚齊了是辕。

說(shuō)了半天囤热，求并行任務(wù)調(diào)度中的關(guān)鍵路徑，實(shí)際上就是求從起點(diǎn)到終點(diǎn)的最長(zhǎng)路徑获三。

通過(guò)求解最長(zhǎng)路徑得到關(guān)鍵路徑

通過(guò)上面的討論旁蔼，現(xiàn)在只需求最長(zhǎng)路徑，就能得到關(guān)鍵路徑疙教。我們知道任務(wù)調(diào)度必須要求圖是無(wú)環(huán)的棺聊，因此可以使用求無(wú)環(huán)加權(quán)有向圖的最短路徑的方法求最長(zhǎng)路徑。

具體方法是：復(fù)制原圖得到一個(gè)副本贞谓，將副本的所有邊的權(quán)重取相反數(shù)限佩，求副本的最短路徑實(shí)際上就是原圖的最長(zhǎng)路徑。

或者一個(gè)更為簡(jiǎn)單的方法：修改邊的放松方法裸弦。改為distTo[v] + edge.weight() > distTo[w]（求最短路徑的不等號(hào)是<）犀暑，即：有比原來(lái)到w更長(zhǎng)的路徑就更新。同時(shí)初始化的時(shí)候烁兰，distTo[i]從原來(lái)的正無(wú)窮改成負(fù)無(wú)窮耐亏。

求無(wú)環(huán)加權(quán)有向圖的最短路徑，可以按照拓補(bǔ)排序依次放松頂點(diǎn)沪斟。詳細(xì)地見我上一遍文章广辰，只需改前述兩個(gè)地方，就能求得最長(zhǎng)路徑主之。

package Chap7;

import java.util.LinkedList;

/**
 * 求無(wú)環(huán)有向圖的最長(zhǎng)路徑
 */
public class AcycliLP {
    private DiEdge[] edgeTo;
    private double[] distTo;

    public AcycliLP(EdgeWeightedDiGraph<?> graph, int s) {
        edgeTo = new DiEdge[graph.vertexNum()];
        distTo = new double[graph.vertexNum()];

        for (int i = 0; i < graph.vertexNum(); i++) {
            // 1. 改成了負(fù)無(wú)窮
            distTo[i] = Double.NEGATIVE_INFINITY;
        }

        distTo[s] = 0.0;
        // 以上是初始化
        TopoSort topo = new TopoSort(graph);

        if (!topo.isDAG()) {
            throw new RuntimeException("該圖存在有向環(huán)择吊，本算法無(wú)法處理！");
        }

        for (int v : topo.order()) {
            relax(graph, v);
        }
    }

    private void relax(EdgeWeightedDiGraph<?> graph, int v) {
        for (DiEdge edge : graph.adj(v)) {
            int w = edge.to();
            // 2槽奕、若路徑更長(zhǎng)就更新
            if (distTo[v] + edge.weight() > distTo[w]) {
                distTo[w] = distTo[v] + edge.weight();
                edgeTo[w] = edge;
            }
        }
    }

    public double distTo(int v) {
        return distTo[v];
    }

    public boolean hasPathTo(int v) {
        return distTo[v] != Double.NEGATIVE_INFINITY;
    }

    public Iterable<DiEdge> pathTo(int v) {
        if (hasPathTo(v)) {
            LinkedList<DiEdge> path = new LinkedList<>();
            for (DiEdge edge = edgeTo[v]; edge != null; edge = edgeTo[edge.from()]) {
                path.push(edge);
            }
            return path;
        }
        return null;
    }
}

好几睛，可以求得關(guān)鍵路徑了。現(xiàn)在來(lái)看一個(gè)任務(wù)調(diào)度的例子粤攒，如何利用上面的實(shí)現(xiàn)來(lái)安排任務(wù)所森。

現(xiàn)在要先將其轉(zhuǎn)換為圖囱持。由于有些工程不好看出哪個(gè)任務(wù)是最先開工的，哪個(gè)是收尾的任務(wù)（比如上圖）焕济。在不同的任務(wù)表中纷妆，每個(gè)任務(wù)都可能成為起點(diǎn)或終點(diǎn)。為了可以應(yīng)付各種任務(wù)表晴弃，不妨設(shè)置虛擬的起點(diǎn)和終點(diǎn)掩幢。因?yàn)槊總€(gè)任務(wù)都可能最先開工，所以設(shè)置一個(gè)虛擬起點(diǎn)可以指向圖中所有頂點(diǎn)上鞠，且權(quán)值都為0际邻；因?yàn)槊總€(gè)頂點(diǎn)都可能作為收尾任務(wù)，因此所有頂點(diǎn)指向一個(gè)虛擬的終點(diǎn)芍阎，權(quán)值是這些頂點(diǎn)代表的任務(wù)所持續(xù)的時(shí)間枯怖。這樣我們也不用在乎任務(wù)表中哪個(gè)任務(wù)最先開工、最后收尾的關(guān)系夠不夠明確了能曾，設(shè)置了虛擬起點(diǎn)和終點(diǎn)后，只要求得從起點(diǎn)到終點(diǎn)的最長(zhǎng)路徑肿轨，中間走過(guò)的路徑的就是各個(gè)任務(wù)執(zhí)行的順序寿冕。

加入虛擬頂點(diǎn)后，上面的任務(wù)表其實(shí)就是下圖椒袍。

各條從s到t的路徑中（想象成各條生產(chǎn)線）驼唱，找出最長(zhǎng)的那條（費(fèi)時(shí)最長(zhǎng)的那條生產(chǎn)線），這條0 -> 9 -> 6 -> 8 -> 2就是關(guān)鍵路徑驹暑，按照這個(gè)順序執(zhí)行任務(wù)就能使得完成整個(gè)工程總時(shí)間最短玫恳。

我們用代碼測(cè)試一下。

package Chap7;

public class CPM {
    private AcycliLP lp;
    private int s; // 虛擬的起點(diǎn)
    private int t; // 虛擬的終點(diǎn)
    private int jobsNum; // 任務(wù)個(gè)數(shù)

    public CPM(double[] jobDuration, int[][] successorAfter) {
        jobsNum = jobDuration.length;
        // 設(shè)置兩個(gè)虛擬頂點(diǎn)优俘，代表起點(diǎn)和終點(diǎn)
        EdgeWeightedDiGraph<?> graph = new EdgeWeightedDiGraph<>(jobsNum + 2);

        s = jobDuration.length; // 起點(diǎn)
        t = s + 1; // 終點(diǎn)

        for (int i = 0; i < jobsNum; i++) {
            // 每個(gè)頂點(diǎn)都可能成為最先開工的京办，所以虛擬起點(diǎn)指向所有頂點(diǎn)，且費(fèi)時(shí)都為0
            graph.addDiEdge(new DiEdge(s, i, 0.0));
            // 每個(gè)頂點(diǎn)都可能成為工程收尾的活動(dòng)帆焕，所有頂點(diǎn)都指向該虛擬終點(diǎn)惭婿，費(fèi)時(shí)自然是每個(gè)活動(dòng)所持續(xù)的時(shí)間
            graph.addDiEdge(new DiEdge(i, t, jobDuration[i]));
            // 任務(wù)i必須在任務(wù)j之前完成， 即加入i -> j的有向邊
            for (int j = 0; j < successorAfter[i].length; j++) {
                int successor = successorAfter[i][j];
                graph.addDiEdge(new DiEdge(i, successor, jobDuration[i]));
            }
            // 找到到每個(gè)活動(dòng)的最長(zhǎng)路徑
            lp = new AcycliLP(graph, s);
        }

    }

    public void printJobExecuteOrder() {
        System.out.println("各任務(wù)開始時(shí)間表：");
        for (int i = 0; i < jobsNum; i++) {
            System.out.println(i + ": " + lp.distTo(i));
        }
        System.out.println("\n按照以下順序執(zhí)行任務(wù)叶雹，開始時(shí)間相同的任務(wù)同時(shí)執(zhí)行财饥。");
        for (DiEdge edge : lp.pathTo(t)) {
            // 遇到起點(diǎn)不打印箭頭
            if (edge.from() == s) {
                System.out.print(edge.to());
            }
            // 最后一個(gè)任務(wù)在前一個(gè)頂點(diǎn)的就打印過(guò)了，遇到最后一條邊換行就行
            else if (edge.to() == t) {
                System.out.println();
            } else {
                System.out.print(" -> " + edge.to());
            }
        }

        System.out.println("總共需要" + lp.distTo(t));
    }

    public static void main(String[] args) {
        // 每個(gè)任務(wù)的持續(xù)時(shí)間
        double[] duration = {41.0, 51.0, 50.0, 36.0, 38.0, 45.0, 21.0, 32.0, 32.0, 29.0};
        // 必須在這些任務(wù)之前完成折晦，如successorAfter[0]表示任務(wù)0的后繼任務(wù)1钥星、7、9满着，也就是說(shuō)0必須在1谦炒、7贯莺、9之前做完
        // {} 表示該任務(wù)不要求在哪個(gè)任務(wù)執(zhí)行前就得完成，說(shuō)明它可能是作為收尾的任務(wù)
        int[][] successorAfter = {{1, 7, 9}, {2}, {}, {}, {}, {}, {3, 8}, {3, 8}, {2}, {4, 6}};
        CPM cpm = new CPM(duration, successorAfter);
        cpm.printJobExecuteOrder();
    }

}

根據(jù)任務(wù)表编饺，給圖增加邊乖篷。

• 虛擬起點(diǎn)到所有頂點(diǎn)的邊，且權(quán)值為0透且；
• 所有頂點(diǎn)到虛擬終點(diǎn)的邊撕蔼，且權(quán)值為頂點(diǎn)任務(wù)持續(xù)時(shí)間。
• 某任務(wù)v必須在一些任務(wù)之前完成的邊秽誊，且權(quán)值為任務(wù)v的持續(xù)時(shí)長(zhǎng)鲸沮。比如0必須在1、7锅论、9之前讼溺，則增加0 -> 1，0 -> 7最易，0 -> 9邊怒坯，且權(quán)值都為任務(wù)0的持續(xù)時(shí)長(zhǎng)。

接下來(lái)通過(guò)AcycliLP類求得關(guān)鍵路徑藻懒，distTo[i]就表示i任務(wù)的開始時(shí)間剔猿，distTo[t]表示做完整個(gè)工程需要的最短時(shí)間；pathTo[i]表示執(zhí)行到任務(wù)i的關(guān)鍵路徑嬉荆，自然pathTo[t]就是整個(gè)工程的關(guān)鍵路徑归敬。

上面的代碼會(huì)打印如下結(jié)果：

各任務(wù)開始時(shí)間表：
0: 0.0
1: 41.0
2: 123.0
3: 91.0
4: 70.0
5: 0.0
6: 70.0
7: 41.0
8: 91.0
9: 41.0

按照以下順序執(zhí)行任務(wù)，開始時(shí)間相同的任務(wù)同時(shí)執(zhí)行鄙早。
0 -> 9 -> 6 -> 8 -> 2
總共需要173.0

可以清楚地看到每個(gè)任務(wù)應(yīng)該在什么時(shí)刻開始動(dòng)工汪茧，只要按照0 -> 9 -> 6 -> 8 -> 2這樣順序執(zhí)行就好。而且限番，有些任務(wù)開始時(shí)間一樣舱污，表示我們應(yīng)該同時(shí)執(zhí)行它們以節(jié)省時(shí)間。所以最后的方案是：0弥虐、5最先并同時(shí)執(zhí)行慌闭，只要0做完了，就開始同時(shí)執(zhí)行1躯舔、7驴剔、9，只要9做完了立刻同時(shí)執(zhí)行4粥庄、6丧失，一旦6做完了同時(shí)執(zhí)行3、8惜互，8只要完工緊接著做2布讹，最后等待2也終于做完琳拭，整個(gè)工程竣工！需要的最短時(shí)間為173.0描验。

關(guān)鍵路徑中的任務(wù)都是做完后就立刻執(zhí)行下一個(gè)的（比如0執(zhí)行完立刻執(zhí)行9白嘁，9完工立刻執(zhí)行6...），不用等待和它一起開工的其他任務(wù)也做完膘流，因?yàn)檫@些任務(wù)在花費(fèi)最長(zhǎng)時(shí)間的這條關(guān)鍵路徑上絮缅，它們肯定能在這期間做完的；而且關(guān)鍵路徑上到每個(gè)關(guān)鍵任務(wù)的路徑已經(jīng)是最長(zhǎng)了呼股，在執(zhí)行某個(gè)關(guān)鍵任務(wù)時(shí)耕魄，必須在它之前執(zhí)行的那些任務(wù)必定已經(jīng)完成，因此各個(gè)關(guān)鍵任務(wù)之間是不會(huì)有空閑和等待時(shí)間的彭谁。如下圖是上面任務(wù)表的執(zhí)行流程

0 -> 9 -> 6 -> 8 -> 2一氣呵成吸奴，中間毫無(wú)停頓。而且其他任務(wù)在這條生產(chǎn)線執(zhí)行過(guò)程中均已完成缠局！

by @sunhaiyu

2017.9.29