網上有很多《使用swagger2構建API文檔》的文章，該文檔是一個在線文檔，需要使用HTTP訪問。但是在我們日常使用swagger接口文檔的時候，有的時候需要接口文檔離線訪問，如將文檔導出為html、markdown格式。又或者我們不希望應用系統與swagger接口文檔使用同一個服務，而是導出HTML之後單獨部署，這樣做保證了對接口文檔的訪問不影響業務系統，也一定程度提高了接口文檔的安全性。核心的實現過程就是：

• 在swagger2接口文檔所在的應用內，利用swagger2markup將接口文檔導出為adoc文件，也可以導出markdown文件。
• 然後將adoc文件轉換為靜態的html格式，可以將html發布到nginx或者其他的web應用容器，提供訪問（本文不會講html靜態部署，只講HTML導出)。

注意：adoc是一種文件格式，不是我的筆誤。不是doc文件也不是docx文件。

一、maven依賴類庫

在已經集成了swagger2的應用內，通過maven坐標引入相關依賴類庫,pom.xml代碼如下：

<dependency>
    <groupId>io.github.swagger2markup</groupId>
    <artifactId>swagger2markup</artifactId>
    <version>1.3.1</version>
</dependency>
<dependency>
    <groupId>io.swagger</groupId>
    <artifactId>swagger-core</artifactId>
    <version>1.5.16</version>
</dependency>
<dependency>
    <groupId>io.swagger</groupId>
    <artifactId>swagger-models</artifactId>
    <version>1.5.16</version>
</dependency>

swagger2markup用於將swagger2在線接口文檔導出為html,markdown,adoc等格式文檔，用於靜態部署或離線閱讀。其中第一個maven坐標是必須的。后兩個maven坐標，當你在執行後面的代碼過程中報下圖中的ERROR，或者有的類無法import的時候使用。

產生異常的原因已經有人在github的issues上給出解釋了：當你使用swagger-core版本大於等於1.5.11,並且swagger-models版本小於1.5.11就會有異常發生。所以我們顯式的引入這兩個jar，替換掉swagger2默認引入的這兩個jar。

二、生成adoc格式文件

下面的代碼是通過編碼方式實現的生成adoc格式文件的方式

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
public class DemoApplicationTests {
    @Test
    public void generateAsciiDocs() throws Exception {
        //    輸出Ascii格式
        Swagger2MarkupConfig config = new Swagger2MarkupConfigBuilder()
                .withMarkupLanguage(MarkupLanguage.ASCIIDOC) //設置生成格式
                .withOutputLanguage(Language.ZH)  //設置語言中文還是其他語言
                .withPathsGroupedBy(GroupBy.TAGS)
                .withGeneratedExamples()
                .withoutInlineSchema()
                .build();

        Swagger2MarkupConverter.from(new URL("http://localhost:8888/v2/api-docs"))
                .withConfig(config)
                .build()
                .toFile(Paths.get("src/main/resources/docs/asciidoc"));
    }
}

• 使用RunWith註解和SpringBootTest註解，啟動應用服務容器。 SpringBootTest.WebEnvironment.DEFINED_PORT表示使用application.yml定義的端口，而不是隨機使用一個端口進行測試，這很重要。
• Swagger2MarkupConfig 是輸出文件的配置，如文件的格式和文件中的自然語言等
• Swagger2MarkupConverter的from表示哪一個HTTP服務作為資源導出的源頭(JSON格式)，可以自己訪問試一下這個鏈接。8888是我的服務端口，需要根據你自己的應用配置修改。
• toFile表示將導出文件存放的位置，不用加後綴名。也可以使用toFolder表示文件導出存放的路徑。二者區別在於使用toFolder導出為文件目錄下按標籤TAGS分類的多個文件，使用toFile是導出一個文件（toFolder多個文件的合集）。

@Test
public void generateMarkdownDocsToFile() throws Exception {
    //    輸出Markdown到單文件
    Swagger2MarkupConfig config = new Swagger2MarkupConfigBuilder()
            .withMarkupLanguage(MarkupLanguage.MARKDOWN)
            .withOutputLanguage(Language.ZH)
            .withPathsGroupedBy(GroupBy.TAGS)
            .withGeneratedExamples()
            .withoutInlineSchema()
            .build();

    Swagger2MarkupConverter.from(new URL("http://localhost:8888/v2/api-docs"))
            .withConfig(config)
            .build()
            .toFile(Paths.get("src/main/resources/docs/markdown"));
}

上面的這一段代碼是生成markdown格式接口文件的代碼。執行上面的2段單元測試代碼，就可以生產對應格式的接口文件。

還有一種方式是通過maven插件的方式，生成adoc和markdown格式的接口文件。筆者不常使用這種方式，沒有使用代碼生成的方式配置靈活，很多配置都放到pom.xml感覺很臃腫。但還是介紹一下,首先配置maven插件swagger2markup-maven-plugin。

<plugin>
    <groupId>io.github.swagger2markup</groupId>
    <artifactId>swagger2markup-maven-plugin</artifactId>
    <version>1.3.1</version>
    <configuration>
        <swaggerInput>http://localhost:8888/v2/api-docs</swaggerInput><!---swagger-api-json路徑-->
        <outputDir>src/main/resources/docs/asciidoc</outputDir><!---生成路徑-->
        <config>
            <swagger2markup.markupLanguage>ASCIIDOC</swagger2markup.markupLanguage><!--生成格式-->
        </config>
    </configuration>
</plugin>

然後運行插件就可以了，如下圖：

三、通過maven插件生成HTML文檔

<plugin>
    <groupId>org.asciidoctor</groupId>
    <artifactId>asciidoctor-maven-plugin</artifactId>
    <version>1.5.6</version>
    <configuration>
         <!--asciidoc文件目錄-->
        <sourceDirectory>src/main/resources/docs</sourceDirectory>
        <!---生成html的路徑-->
        <outputDirectory>src/main/resources/html</outputDirectory>
        <backend>html</backend>
        <sourceHighlighter>coderay</sourceHighlighter>
        <attributes>
            <!--導航欄在左-->
            <toc>left</toc>
            <!--显示層級數-->
            <!--<toclevels>3</toclevels>-->
            <!--自動打数字序號-->
            <sectnums>true</sectnums>
        </attributes>
    </configuration>
</plugin>

adoc的sourceDirectory路徑必須和第三小節中生成的adoc文件路徑一致。然後按照下圖方式運行插件。

HTMl接口文檔显示的效果如下，有了HTML接口文檔你想轉成其他各種格式的文檔就太方便了，有很多工具可以使用。這裏就不一一介紹了。

期待您的關注

• 向您推薦博主的系列文檔：
• 本文轉載註明出處（必須帶連接，不能只轉文字）：。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

介紹了高效科技組織的特點及管理經驗，指出科技團隊的定位和使命在於支持業務、賦能業務、最終引領業務，同時，還介紹了面向未來的科技組織的特點及對管理者提出的能力要求。

內容來源 | LeaTech全球CTO領導力峰會宜信公司CTO 高級副總裁向江旭分享《大型科技團隊的管理》

主講人 | 宜信公司CTO 高級副總裁向江旭

實錄整理 | 宜信技術學院成芳

引言：11月16日，由51CTO旗下CTO訓練營品牌精心打造的LeaTech全球CTO領導力峰會在北京粵財JW萬豪酒店拉開序幕。作為CTO、技術VP、技術總監等技術管理者人群的高端社交圈，本屆峰會現場聚集了CTO訓練營歷屆校友、CTO導師，以及行業中的資深技術管理者。600多位與會嘉賓在現場充分交流了有關技術性視野、技術領導力、技術團隊組織建設等精彩話題的觀點與思考，藉助峰會這個線下平台，技術管理者們积極探索了更多商業可能，開拓管理視野，令自身領導力再上新台階。

本次峰會邀請到宜信公司CTO 高級副總裁向江旭先生，帶來主題為《大型科技團隊的管理》的分享，向江旭先生在分享中提到科技團隊的定位和使命在於支持業務、賦能業務、最終引領業務，同時，他還介紹了面向未來的科技組織的特點及對管理者提出的能力要求。

以下為本次演講的分享實錄。

各位朋友下午好，今天我分享的主題是《大型科技團隊的管理》，非常高興能跟大家分享一些關於大型科技團隊管理的經驗和觀察。

去年12月，我在另一個峰會的分享中提到經濟寒冬、金融寒冬，今年這個時候也開始進入冬季，但是不管外面的大環境如何，技術人/技術圈都非常幸運，在科技賦能的市場中，技術人總是被需要的，技術圈也總是熱情高漲的。我認為，關於技術團隊的管理經驗非常值得與大家一起分享和交流。

一、大型科技團隊的特點及定位

大型科技團隊一般都有以下幾個特點：

• 一定規模。顧名思義，談到大型科技團隊首先想到的特點肯定是團隊成員眾多。
• 地域分佈廣。科技團隊的成員可能分佈在不同的地方。
• 團隊背景多元。這種多元包括兩個層面的含義：一是角色背景多元，有的成員甚至不是做技術（這裏特指軟件研發）的，而是從其他行業轉過來的，比如行業數據分析師等角色。二是團隊成員的種族、國家背景多元。

一定規模、團隊背景多元化、分佈在不同地域等特點，使得大型科技團隊在管理上面臨着非常大的挑戰。

1.1 大型科技公司的科技團隊組織架構

上圖是一些大家耳熟能詳的漫畫，介紹了幾種典型的科技公司的科技團隊架構形式。

• “Google式團隊架構”：最上面是兩個創始人和一個CEO，下面是部門負責人，其特點是下一級有多個彙報對象，這是谷歌的內部架構和管理方式，從一個層面代表了某個歷史階段科技組織的團隊構造。
• “Amazon式團隊架構”：典型的層級式管理，逐級彙報。
• “Oracle式團隊架構”：特點是有一個專門的法務團隊，因為Oracle收購了很多公司，某種程度上它是一個法務或銷售導向的公司。
• “Facebook式團隊架構”：Facebook的團隊結構和它的業務一樣，是社交網絡形式的網狀結構。
• “Apple式團隊架構”：（這是一張比較早期的圖）其特點是最核心的靈魂人物的觸角深入到公司的方方面面，事無巨細都是他在親自主導。
• “Microsoft式團隊結構”：最後這張也是之前的圖，代表的是我服務過的公司-微軟。三大版塊部門沒有交集，很深的部門牆，互相之間存在比較激烈的內部鬥爭。

由此可見，大型科技公司的文化基因決定了其科技團隊的組織架構形式，而科技組織架構的設計和管理很大程度上決定了組織的效能。

1.2 科技團隊的使命和定位

在討論科技團隊的管理之前，有一個很重要的前提，要知道科技團隊的使命和定位是什麼。

很多非科技驅動的公司，比如銀行、保險公司、地產公司等，都在計劃或嘗試轉型成為一家科技公司。我認為在這樣的背景下，科技從業人員反而應該更清楚自己的定位。

1）支持業務

首先我們是支持業務的，要把業務放在最核心的位置，如果公司是靠產品和服務生存的，業務沒做好，科技也不一定能做好。當然公司類型不一樣，不能一概而論。比如純科技公司微軟，產品和服務本身就是技術，技術的好壞決定了公司業務的好壞。

2）賦能業務

科技團隊通過開發一些工具，幫助業務、銷售等部門實現業務流程信息化、智能化，使得業務流轉的過程更為暢通和友好。甚至更進一步，開發一個好的技術平台，使得生態圈、合作夥伴、第三方公司能夠在平台上發展業務、共建生態，這時科技團隊就起到了賦能業務的作用。

3）引領業務

科技團隊還可以通過技術創新、產品創新來拓展新的業務領域，催生新的業務模式，增加新的收入來源，成為引領業務的力量。

這三點是相輔相成的，支持業務、賦能業務，最後引領業務，不論是在技術驅動的公司，還是在業務驅動的公司，技術團隊的使命和定位都是如此。

1.3 技術和業務的關係

定位決定地位，業務發展是終極目標，當面臨整體技術戰略與商業戰略衝突、技術實施節點選擇、技術與業務路徑匹配等問題時，技術管理者可以從以下方面進行思考。

1）解決技術戰略與商業戰略的衝突。

技術和業務除了相輔相成的關係外，還存在一定的衝突。其實技術和業務的衝突在大家平時的工作中經常見到，比如業務同事着急上線一個功能來做活動；而技術覺得要達到同樣的目的，我們可能需要好的設計和架構，而不是簡單做一個臨時補丁式的功能，這就是很常見的技術和業務的衝突。

雖然技術負責人要對未來中長期的戰略布局保持持續思考，但這取決於公司的大小、規模和階段。如果是初創公司，首要任務是生存，那麼業務需求是最高優先級，首先要考慮解決系統不穩定、安全或其他問題；但對於有一定體量的企業，公司業務已經發展到一定階段，技術團隊也有一定的規模，當存在業務需求和技術戰略的衝突時，在滿足最緊迫的業務需求的同時，將一定精力投入到基礎技術研發中去，必須要做中長期的項目預研，做一些底層、甚至風險較高的研究。

2）技術變革的實施節點

我們永遠在高速公路上奔跑，一邊行駛一邊換輪子或換部件的事情一直在發生。技術重構、技術變革、技術債務償還的時機和節點和對業務產生的影響，是我們面臨的又一個挑戰，也是需要我們長期考慮的問題。

什麼節點選擇什麼樣的技術？技術負責人具體把握新技術引入節點，其實難度很大。如果新技術距離實現商用價值僅有一年時間，那麼必須要進行布局，申請預算，建立團隊推進；如果新技術商業化已經迫在眉睫，競爭對手已經在布局了，那麼採取的措施就不是從頭做起，更好的應對方式可能是進行併購或資本運作。

3）技術與業務路徑匹配

新技術來臨，對業務的影響和衝擊會分短、中、長期，有的技術在短期內對新的業態有幫助，有的技術需要一個比較長的周期才會對業務產生影響。這時技術領導人要評估技術本身，並將其與業務的戰略路徑進行匹配，如果對長期業務有幫助，新技術仍然要引入，只是選擇時間點、投入範圍等可能會不一樣。

同時，在引入新技術進行技術創新時，還要注意新技術對當前業務產生的影響。舉個例子，宜信是一家金融科技公司，有自己的催收部門，我們一方面通過規範催收人員的行為來進行催收，一方面自己研發催收機器人。催收機器人的出現意味着部分催收人員的工作將會被替代，同時，機器人的特點之一是沒有情緒，它會按照程序設定禮貌地和用戶溝通，這就會在一定程度對催收效果和業務產生影響。因此我們還要考慮在哪個環節使用機器人這個技術，帶來的效果會更好。這也是技術和業務相輔相成又存在矛盾衝突關係的體現。

1.4 科技戰略

對於大型科技團隊而言，科技戰略思維也非常重要。舉幾個例子。

舉例提到的是我工作過的幾家公司，它們在不同的時間點做了一些不同的戰略調整。

很多年以前，在中國90%的Windows都不是正版的，於是微軟內部提出一個計劃，希望Windows在中國免費。這個計劃現在看起來非常簡單，也很容易理解，正版Windows免費帶來的好處顯而易見：它是一個非常好的終端用戶觸點、可以獲取更多用戶，可以基於這些用戶數據做數據分析和精準營銷等。但是當時微軟Windows的老大極力反對這件事，他認為這會影響Windows的營收，因此這個計劃最終沒有實行。這就是戰略思維的問題，如果不是在中國本地親身體驗這個環境和市場，可能做出的戰略決策就不一定是準確的，帶來的結果可能是痛失更大的發展機會。

微軟後來的雲戰略轉型是成功的，而移動轉型卻是失敗的。移動轉型時期，微軟收購諾基亞旗下的大部分手機業務，並基於Windows自研出一個Windows操作系統放在手機硬件上。因為微軟覺得沒有推出自己的手機這是一個缺憾，還認為佔領移動終端必須基於Windows，因此作出這樣的決策，結果以失敗告終。Windows使微軟一度成為桌面系統垄斷的贏家，也使得微軟在移動轉型失敗，成為下一步發展的絆腳石，可謂是成也Windows，敗也Windows。

再看蘇寧，蘇寧受到阿里巴巴、京東等電商的衝擊，痛下決心必須做電商，它採取的戰略是結合自己線下門店的優勢做O2O智慧門戶，實行線上下單、線下體驗、送貨到家，這種全渠道、全觸點的用戶交互與服務形式，使得蘇寧成為“中國傳統行業数字化轉型互聯網”為數不多的成功案例之一。

現在金融行業正處於嚴監管的環境，金融公司該何去何從？在這裏分享一些我的看法。

縱觀改革開放以來的歷史，每個行業在開始初期都是開放的，任何人都可以參与進來，魚龍混雜，一旦行業出現亂象，就會有監管介入，那些能力不強的、不合規的會被淘汰，等到監管后再開放、行業再成熟時，最後存活下來的才能健康發展。金融行業也是如此。

科技同仁不僅要埋頭做好自己的工作，也要抬頭看看歷史和未來，思考我們身處的這個行業，我們從事的科技工作對公司、對行業的作用是什麼？行業未來的發展趨勢是什麼，這對我們未來的職業發展也有幫助。

二、大型科技團隊的管理實踐

2.1 成功科技組織的特點

無論是前面提到的國際科技巨頭，還是國內優秀的互聯網公司，成功的科技組織都具備一些共同的特點。

1）高效、敏捷

優秀的科技團隊一定是一個高效、敏捷的團隊，能夠快速響應用戶和市場的需求變化，快速上線產品、得到反饋、不斷迭代更新，滿足或超過用戶的預期。

2）商業思維

科技團隊一定要有商業頭腦、商業思維，因為無論是搭建系統，還是做APP，一定有用戶，我們需要真正洞察到用戶的痛點和問題，幫他們及時解決問題，給他們創造價值。

3）數據驅動

如果科技團隊的工作只圍繞產品經理提出的需求，未來的產品路線圖還不夠，一定要基於用戶反饋、市場反饋、日活、月活、留存、轉化等數據來驅動產品走向、技術走向。

4）變革創新

很多公司都在強調變革和創新，我認為這是科技團隊必須要打造的環境和氛圍，方式很多，比如組織各種各樣的黑客馬拉松、團隊之間的交流分享等。我們公司也在組織黑客馬拉松，每個月業務部門都有比較棘手或緊迫的項目，技術團隊與業務團隊聯合把這些業務痛點解決，成果馬上應用到業務環境中去。

2.2 績效管理

績效管理的機制在微軟實行了很多年，其強制5%、10%的淘汰機制，一直被人詬病，因為它使得很多團隊互相指責、互相拆台，有的員工為了績效“寧當雞頭不當鳳尾”，組織內形成不好的文化和結果。後來改成了5%、10%的獎勵，從懲罰後進者變成鼓勵先進者，取得的效果好很多。

2.3 溝通

溝通每天都在進行，比如各種大大小小的會議。我發現不管開多少會，高層領導的想法、思路並不一定能被所有同事理解，因為溝通方式、溝通渠道的原因，信息不能觸達到所有人，需要重複很多遍。

我們可以利用一些非正式場合或社交媒體來進行交流。比如我們公司有每月的CTO午餐會，抽籤的方式，各團隊都有機會可以坐下來跟我一起吃午餐，通過這種面對面的交流，我會向團隊成員提出一些我的看法，或者推薦一本書，他們會向我反饋當時的痛點、想法等。我覺得這是一種非常好的機制，大家能夠在一個寬鬆的環境下面對面地交流。記得以前在思科工作的時候，也有CEO早餐會，每個月過生日的員工可以跟CEO吃早餐，也是一種很好的溝通互動的方式。

2.4 團隊文化

團隊文化包含很多要素，我對這幾點的認同感比較深：主人翁意識、緊迫感、同理心。

2.5 技術決策

商業世界充滿了選擇和決策。作為一個技術決策人，有時很難做決定，難做決定就意味着拖延，這一點對於大型科技團隊的決策人來說是比較忌諱的，很多時候不在於你做的決策是對還是錯，而是在於你做不做決定。如果你是一個不敢做決定的人，帶領的團隊就會缺乏方向感，你做的決定就會受到團隊的質疑和挑戰。

有時候即使你做出了錯誤的決定，但大家一起努力執行的時候，可能可以逐步調整轉變從而達成正確的結果。技術管理人一定要懂得取捨，通過大腦計算，明確目標，了解關聯方，分析可選項和利弊，基於目標、數據和對未來的預期做出決策。

2.6 執行力

執行的時候要不忘初心，一步步地大處着眼、小處着手，小步快跑，快速迭代，及時反饋，及時調整，朝着設定的目標專心致志、心無旁騖。

2.7 終極目標

最後的終極目標是，希望科技團隊能夠做到比業務更懂業務，比用戶更懂用戶，讓技術本身變成公司核心的業務。我認為如果能做到科技團隊在公司起到了核心的作用，這才是永生的價值。

三、面向未來的科技組織

3.1 回顧微軟的3任CEO和3種組織

就微軟而言，同一家公司，文化和技術氛圍在不同階段是不一樣的。

第一個階段，CEO比爾蓋茨，技術為王，認為技術改變世界，代碼可以改變千萬人的生活。這一階段造就了好的產品，同時也存在一些負面影響，比如垄斷等。

第二個階段，CEO鮑爾默，業績為王，要求所有設備，包括手機、電視、車等，都跑到Windows上，還和海爾合作推出基於Windows的智能電視，當時所有跟Windows衝突的想法和計劃基本都被扼殺在搖籃里，這也導致公司錯失了很多發展的機會。

第三個階段，CEO納德拉，公司文化變得更加包容、強調同理心、開放透明，包括技術開源、提供雲服務、和蘋果/谷歌等競爭對手合作。這種開放包容的氛圍，讓微軟得以浴火重生。

3.2 面向未來的科技組織

面向未來的科技組織應該是什麼樣的？

1）跨界融合

不同背景的人，融合到一個跨界的氛圍和組織中，一起發力。

2）数字化轉型

傳統公司也在做数字化轉型，這時會引入很多科技人才，利用科技幫助公司完成轉型，並在行業實現快速發展。

3）全球共享、全球分佈

團隊組成全球化，來自不同國家、不同種族的人組成一個全球化的團隊。東南亞有一個集美團、滴滴、螞蟻金服為一體的公司，解決出行、支付、快遞的問題，這家公司有一個幾千人的科技團隊，團隊成員來自50多個種族，分佈在美國、新加坡、北京、印尼、印度等國家和地區。

4）技術引領

未來科技團隊不僅要做到賦能業務，還要能實現引領業務。

這樣的科技團隊需要跨界、全球化、複合型的領導人，只有既懂科技、又懂管理；既要懂業務，又懂行業的管理人才才能適合於領導面向未來的科技組織。

以上就是今天跟大家分享的全部內容，時間很短，主要介紹了一些在不同公司不同行業的大型科技團隊的管理經驗，希望大家在面向未來的科技組織中找到自己的定位，成為一個優秀的領導者。謝謝大家。

本文根據向江旭老師在LeaTech全球CTO領導力峰會上的分享內容整理所得，轉載請聯繫授權。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

本文由葡萄城技術團隊編撰並首發

轉載請註明出處：，葡萄城為開發者提供專業的開發工具、解決方案和服務，賦能開發者。

引言

在Web應用發展的初期，那時關係型數據庫受到了較為廣泛的關注和應用，原因是因為那時候Web站點基本上訪問和併發不高、交互也較少。而在後來，隨着訪問量的提升，使用關係型數據庫的Web站點多多少少都開始在性能上出現了一些瓶頸，而瓶頸的源頭一般是在磁盤的I/O上。而隨着互聯網技術的進一步發展，各種類型的應用層出不窮，這導致在當今雲計算、大數據盛行的時代，對性能有了更多的需求，主要體現在以下四個方面：

1. 低延遲的讀寫速度：應用快速地反應能極大地提升用戶的滿意度
2. 支撐海量的數據和流量：對於搜索這樣大型應用而言，需要利用PB級別的數據和能應對百萬級的流量
3. 大規模集群的管理：系統管理員希望分佈式應用能更簡單的部署和管理
4. 龐大運營成本的考量：IT部門希望在硬件成本、軟件成本和人力成本能夠有大幅度地降低

為了克服這一問題，NoSQL應運而生，它同時具備了高性能、可擴展性強、高可用等優點，受到廣泛開發人員和倉庫管理人員的青睞。

Redis是什麼

Redis是現在最受歡迎的NoSQL數據庫之一，Redis是一個使用ANSI C編寫的開源、包含多種數據結構、支持網絡、基於內存、可選持久性的鍵值對存儲數據庫，其具備如下特性：

• 基於內存運行，性能高效
• 支持分佈式，理論上可以無限擴展
• key-value存儲系統
• 開源的使用ANSI C語言編寫、遵守BSD協議、支持網絡、可基於內存亦可持久化的日誌型、Key-Value數據庫，並提供多種語言的API

相比於其他數據庫類型，Redis具備的特點是：

• C/S通訊模型
• 單進程單線程模型
• 豐富的數據類型
• 操作具有原子性
• 持久化
• 高併發讀寫
• 支持lua腳本

哪些大廠在使用Redis？

• github
• twitter
• 微博
• Stack Overflow
• 阿里巴巴
• 百度
• 美團
• 搜狐

Redis的應用場景有哪些？

Redis 的應用場景包括：緩存系統（“熱點”數據：高頻讀、低頻寫）、計數器、消息隊列系統、排行榜、社交網絡和實時系統。

 

Redis的數據類型及主要特性

Redis提供的數據類型主要分為5種自有類型和一種自定義類型，這5種自有類型包括：String類型、哈希類型、列表類型、集合類型和順序集合類型。

String類型：

它是一個二進制安全的字符串，意味着它不僅能夠存儲字符串、還能存儲圖片、視頻等多種類型, 最大長度支持512M。

對每種數據類型，Redis都提供了豐富的操作命令，如：

• GET/MGET
• SET/SETEX/MSET/MSETNX
• INCR/DECR
• GETSET
• DEL

哈希類型：

該類型是由field和關聯的value組成的map。其中，field和value都是字符串類型的。

Hash的操作命令如下：

• HGET/HMGET/HGETALL
• HSET/HMSET/HSETNX
• HEXISTS/HLEN
• HKEYS/HDEL
• HVALS

列表類型：

該類型是一個插入順序排序的字符串元素集合, 基於雙鏈表實現。

List的操作命令如下：

• LPUSH/LPUSHX/LPOP/RPUSH/RPUSHX/RPOP/LINSERT/LSET
• LINDEX/LRANGE
• LLEN/LTRIM

集合類型：

Set類型是一種無順序集合, 它和List類型最大的區別是：集合中的元素沒有順序, 且元素是唯一的。

Set類型的底層是通過哈希表實現的，其操作命令為：

• SADD/SPOP/SMOVE/SCARD
• SINTER/SDIFF/SDIFFSTORE/SUNION

Set類型主要應用於：在某些場景，如社交場景中，通過交集、並集和差集運算，通過Set類型可以非常方便地查找共同好友、共同關注和共同偏好等社交關係。

順序集合類型：

ZSet是一種有序集合類型，每個元素都會關聯一個double類型的分數權值，通過這個權值來為集合中的成員進行從小到大的排序。與Set類型一樣，其底層也是通過哈希表實現的。

ZSet命令：

• ZADD/ZPOP/ZMOVE/ZCARD/ZCOUNT
• ZINTER/ZDIFF/ZDIFFSTORE/ZUNION

Redis的數據結構

Redis的數據結構如下圖所示：

關於上表中的部分釋義：

1. 壓縮列表是列表鍵和哈希鍵的底層實現之一。當一個列表鍵只包含少量列表項，並且每個列表項要麼就是小整數，要麼就是長度比較短的字符串，Redis就會使用壓縮列表來做列表鍵的底層實現
2. 整數集合是集合鍵的底層實現之一，當一個集合只包含整數值元素，並且這個集合的元素數量不多時，Redis就會使用整數集合作為集合鍵的底層實現

如下是定義一個Struct數據結構的例子：

 

簡單動態字符串SDS (Simple Dynamic String)

基於C語言中傳統字符串的缺陷，Redis自己構建了一種名為簡單動態字符串的抽象類型，簡稱SDS，其結構如下：

SDS幾乎貫穿了Redis的所有數據結構，應用十分廣泛。

SDS的特點

和C字符串相比，SDS的特點如下：

　　1. 常數複雜度獲取字符串長度

　　　　Redis中利用SDS字符串的len屬性可以直接獲取到所保存的字符串的長
　　　　度，直接將獲取字符串長度所需的複雜度從C字符串的O(N)降低到了O(1)。

　　2. 減少修改字符串時導致的內存重新分配次數

　　　　通過C字符串的特性，我們知道對於一個包含了N個字符的C字符串來說，其底層實現總是N+1個字符長的數組（額外一個空字符結尾）

　　　　那麼如果這個時候需要對字符串進行修改，程序就需要提前對這個C字符串數組進行一次內存重分配（可能是擴展或者釋放）　

　　　　而內存重分配就意味着是一個耗時的操作。

Redis巧妙的使用了SDS避免了C字符串的缺陷。在SDS中，buf數組的長度不一定就是字符串的字符數量加一，buf數組裡面可以包含未使用的字節，而這些未使用的字節由free屬性記錄。

與此同時，SDS採用了空間預分配的策略，避免C字符串每一次修改時都需要進行內存重分配的耗時操作，將內存重分配從原來的每修改N次就分配N次——>降低到了修改N次最多分配N次。

如下是Redis對SDS的簡單定義：

  

Redis特性1：事務

• 命令序列化，按順序執行
• 原子性
• 三階段: 開始事務 – 命令入隊 – 執行事務
• 命令：MULTI/EXEC/DISCARD

Redis特性2：發布訂閱(Pub/Sub)

• Pub/sub是一種消息通訊模式
• Pub發送消息, Sub接受消息
• Redis客戶端可以訂閱任意數量的頻道
• “fire and forgot”, 發送即遺忘
• 命令：Publish/Subscribe/Psubscribe/UnSub

　　

Redis特性3：Stream

• Redis 5.0新增
• 等待消費
• 消費組(組內競爭)
• 消費歷史數據
• FIFO

 

 

 

以上就是Redis的基本概念，下面我們將介紹在開發過程中可能會踩到的“坑”。

Redis常見問題解析：擊穿

概念：在Redis獲取某一key時, 由於key不存在, 而必須向DB發起一次請求的行為, 稱為“Redis擊穿”。

引發擊穿的原因：

• 第一次訪問
• 惡意訪問不存在的key
• Key過期

合理的規避方案：

• 服務器啟動時, 提前寫入
• 規範key的命名, 通過中間件攔截
• 對某些高頻訪問的Key，設置合理的TTL或永不過期

Redis常見問題解析：雪崩

概念：Redis緩存層由於某種原因宕機后，所有的請求會湧向存儲層，短時間內的高併發請求可能會導致存儲層掛機，稱之為“Redis雪崩”。

合理的規避方案：

• 使用Redis集群
• 限流

Redis在產品開發中的應用實踐

為此，我很高興的為大家介紹，葡萄城架構師Jim將在2019-11-27 14：00 為大家帶來一場公開課，其中 Jim除了為大家講解Redis的基礎，同時也會實際演示他所在的項目組使用Redis時碰到的問題以及解決方案，對於剛接觸Redis的同學來說，更具參考意義和學習價值，歡迎大家屆時參加，公開課地址：。

• 後端採用nodeJS
• 使用Azure的Redis服務
• Redis的使用場景

　　　　-  token緩存, 用於令牌驗證

　　　　-  IP白名單

碰到的問題

• “網絡抖動”或者Redis服務異常導致Redis訪問超時
• Redis客戶端驅動穩定性問題

　　　　-  連接池 “Broken connection” 問題

　　　　-  JS的Promise引出的Redis重置問題

下面我們來簡單了解一下Redis的進階知識。

進階之Redis協議簡介

Redis客戶端通訊協議：RESP(Redis Serialization Protocol)，其特點是：

• 簡單
• 解析速度快
• 可讀性好

Redis集群內部通訊協議：RECP(Redis Cluster Protocol ) ，其特點是：

• 每一個node兩個tcp 連接
• 一個負責client-server通訊(P: 6379)
• 一個負責node之間通訊(P: 10000 + 6379)

 

Redis協議支持的數據類型：

• 簡單字符(首字節: “+”)

      　　“+OK\r\n”

• 錯誤(首字節: “-”)

      　　“-error msg\r\n”

• 数字(首字節: “:”)

      　　“:123\r\n”

• 批量字符(首字節: “$”)

      　　“&hello\r\nWhoa re you\r\n”

• 數組(首字節: “*”)

      　　“*0\r\n”

      　　“*-1\r\n”

除了Redis，還有什麼NoSQL型數據庫

市面上類似於Redis，同樣是NoSQL型的數據庫有很多，如下圖所示，除了Redis，還有MemCache、Cassadra和Mongo。下面，我們就分別對這幾個數據庫做一下簡要的介紹：

 

 

Memcache：這是一個和Redis非常相似的數據庫，但是它的數據類型沒有Redis豐富。Memcache由LiveJournal的Brad Fitzpatrick開發，作為一套分佈式的高速緩存系統，被許多網站使用以提升網站的訪問速度，對於一些大型的、需要頻繁訪問數據庫的網站訪問速度的提升效果十分顯著。

Apache Cassandra：（社區內一般簡稱為C*）這是一套開源分佈式NoSQL數據庫系統。它最初由Facebook開發，用於儲存收件箱等簡單格式數據，集Google BigTable的數據模型與Amazon Dynamo的完全分佈式架構於一身。Facebook於2008將 Cassandra 開源，由於其良好的可擴展性和性能，被 Apple、Comcast、Instagram、Spotify、eBay、Rackspace、Netflix等知名網站所採用，成為了一種流行的分佈式結構化數據存儲方案。

MongoDB：是一個基於分佈式文件存儲、面向文檔的NoSQL數據庫，由C++編寫，旨在為WEB應用提供可擴展的高性能數據存儲解決方案。MongoDB是一個介於關係數據庫和非關係數據庫之間的產品，是非關係數據庫當中功能最豐富，最像關係型數據庫的，它支持的數據結構非常鬆散，是一種類似json的BSON格式。

總結

以上就是Redis入門介紹教程，如果各位還想了解更多，歡迎通過評論和私信的方式告訴我。

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

一、背景

地圖App的功能可以簡單概括為定位，搜索，導航三部分，分別解決在哪裡，去哪裡，和怎麼去的問題。高德地圖的搜索場景下，輸入的是，地理相關的檢索query，用戶位置，App圖面等信息，輸出的是，用戶想要的POI。如何能夠更加精準地找到用戶想要的POI，提高滿意度，是評價搜索效果的最關鍵指標。

一個搜索引擎通常可以拆分成query分析、召回、排序三個部分，query分析主要是嘗試理解query表達的含義，為召回和排序給予指導。

地圖搜索的query分析不僅包括通用搜索下的分詞，成分分析，同義詞，糾錯等通用NLP技術，還包括城市分析，wherewhat分析，路徑規劃分析等特定的意圖理解方式。

常見的一些地圖場景下的query意圖表達如下：

query分析是搜索引擎中策略密集的場景，通常會應用NLP領域的各種技術。地圖場景下的query分析，只需要處理地理相關的文本，多樣性不如網頁搜索，看起來會簡單一些。但是，地理文本通常比較短，並且用戶大部分的需求是唯一少量結果，要求精準度非常高，如何能夠做好地圖場景下的文本分析，並提升搜索結果的質量，是充滿挑戰的。

二、整體技術架構

搜索架構

類似於通用檢索的架構，地圖的檢索架構包括query分析，召回，排序三個主要部分。先驗的，用戶的輸入信息可以理解為多種意圖的表達，同時下發請求嘗試獲取檢索結果。后驗的，拿到每種意圖的檢索結果時，進行綜合判斷，選擇效果最好的那個。

query分析流程

具體的意圖理解可分為基礎query分析和應用query分析兩部分，基礎query分析主要是使用一些通用的NLP技術對query進行理解，包括分析，成分分析，省略，同義詞，糾錯等。應用query分析主要是針對地圖場景里的特定問題，包括分析用戶目標城市，是否是where+what表達，是否是從A到B的路徑規劃需求表達等。

整體技術演進

在地里文本處理上整體的技術演進經歷了規則為主，到逐步引入機器學習，到機器學習全面應用的過程。由於搜索模塊是一個高併發的線上服務，對於深度模型的引入有比較苛刻的條件，但隨着性能問題逐漸被解決，我們從各個子方向逐步引入深度學習的技術，進行新一輪的效果提升。

NLP領域技術在最近幾年取得了日新月異的發展，bert，XLNet等模型相繼霸榜，我們逐步統一化各個query分析子任務，使用統一的向量表示對進行用戶需求進行表達，同時進行seq2seq的多任務學習，在效果進一步提升的基礎上，也能夠保證系統不會過於臃腫。

本文就高德地圖搜索的地理文本處理，介紹相關的技術在過去幾年的演進。我們將選取一些點分上下兩篇進行介紹，上篇主要介紹搜索引擎中一些通用的query分析技術，包括糾錯，改寫和省略。下篇着重介紹地圖場景中特有query分析技術，包括城市分析，wherewhat分析，路徑規劃。

三、通用query分析技術演進

3.1 糾錯

在搜索引擎中，用戶輸入的檢索詞（query）經常會出現拼寫錯誤。如果直接對錯誤的query進行檢索，往往不會得到用戶想要的結果。因此不管是通用搜索引擎還是垂直搜索引擎，都會對用戶的query進行糾錯，最大概率獲得用戶想搜的query。

在目前的地圖搜索中，約有6%-10%的用戶請求會輸入錯誤，所以query糾錯在地圖搜索中是一個很重要的模塊，能夠極大的提升用戶搜索體驗。

在搜索引擎中，低頻和中長尾問題往往比較難解決，也是糾錯模塊面臨的主要問題。另外，地圖搜索和通用搜索，存在一個明顯的差異，地圖搜索query結構化比較突出，query中的片段往往包含一定的位置信息，如何利用好query中的結構化信息，更好地識別用戶意圖，是地圖糾錯獨有的挑戰。

常見錯誤分類

(1) 拼音相同或者相近，例如: 盤橋物流園-潘橋物流園
(2) 字形相近，例如: 河北冒黎-河北昌黎
(3) 多字或者漏字，例如: 泉州州頂街-泉州頂街

糾錯現狀

原始糾錯模塊包括多種召回方式，如：

拼音糾錯：主要解決短query的拼音糾錯問題，拼音完全相同或者模糊音作為糾錯候選。
拼寫糾錯：也叫形近字糾錯，通過遍歷替換形近字，用query熱度過濾，加入候選。
組合糾錯：通過翻譯模型進行糾錯替換，資源主要是通過query對齊挖掘的各種替換資源。

組合糾錯翻譯模型計算公式：

其中p(f)是語言模型，p(f|e)是替換模型。

問題1：召回方式存在缺陷。目前query糾錯模塊主要召回策略包括拼音召回、形近字召回，以及替換資源召回。對於低頻case，解決能力有限。

問題2：排序方式不合理。糾錯按照召回方式分為幾個獨立的模塊，分別完成相應的召回和排序，不合理。

技術改造

改造1：基於空間關係的實體糾錯
原始的糾錯主要是基於用戶session挖掘片段替換資源，所以對於低頻問題解決能力有限。但是長尾問題往往集中在低頻，所以低頻問題是當前的痛點。

地圖搜索與通用搜索引擎有個很大的區別在於，地圖搜索query比較結構化，例如北京市朝陽區阜榮街10號首開廣場。我們可以對query進行結構化切分（也就是地圖中成分分析的工作），得到這樣一種帶有類別的結構化描述，北京市【城市】朝陽區【區縣】阜榮街【道路】10號【門址後綴】首開廣場【通用實體】。

同時，我們擁有權威的地理知識數據，利用權威化的地理實體庫進行前綴樹+後綴樹的索引建庫，提取疑似糾錯的部分在索引庫中進行拉鏈召回，同時利用實體庫中的邏輯隸屬關係對糾錯結果進行過濾。實踐表明，這種方式對低頻的區劃或者實體的錯誤有着明顯的作用。

基於字根的字形相似度計算

上文提到的排序策略裏面通過字形的編輯距離作為排序的重要特徵，這裏我們開發了一個基於字根的字形相似度計算策略，對於編輯距離的計算更為細化和準確。漢字信息有漢字的字根拆分詞表和漢字的筆畫數。

將一個漢字拆分成多個字根，尋找兩個字的公共字根，根據公共字根筆畫數來計算連個字的相似度。

改造2：排序策略重構

原始的策略召回和排序策略耦合，導致不同的召回鏈路，存在顧此失彼的情況。為了能夠充分發揮各種召回方式的優勢，急需要對召回和排序進行解耦並進行全局排序優化。為此我們增加了排序模塊，將流程分為召回和排序兩階段。

模型選擇

對於這個排序問題，這裏我們參考業界的實踐，使用了基於pair-wise的gbrank進行模型訓練。

樣本建設

通過線上輸出結合人工review的方式構造樣本。

特徵建設
(1) 語義特徵。如統計語言模型。
(2) 熱度特徵。pv，點擊等。
(3) 基礎特徵。編輯距離，切詞和成分特徵，累積分佈特徵等。

這裏解決了糾錯模塊兩個痛點問題，一個是在地圖場景下的大部分低頻糾錯問題。另一個是重構了模塊流程，將召回和排序解耦，充分發揮各個召回鏈路的作用，召回方式更新后只需要重訓排序模型即可，使得模塊更加合理，為後面的深度模型升級打下良好的基礎。後面在這個框架下，我們通過深度模型進行seq2seq的糾錯召回，取得了進一步的收益。

3.2 改寫

糾錯作為query變換的一種方式的召回策略存在諸多限制，對於一些非典型的query變換表達，存在策略的空白。比如query=永城市新農合辦，目標POI是永城市新農合服務大廳。用戶的低頻query，往往得不到較好搜索效果，但其實用戶描述的語義與主poi的高頻query是相似的。

這裏我們提出一種query改寫的思路，可以將低頻query改寫成語義相似的高頻query，以更好地滿足用戶需求多樣性的表達。

這是一個從無到有的實現。用戶表達的query是多樣的，使用規則表達顯然是難以窮盡的，直觀的思路是通過向量的方式召回，但是向量召回的方式很可能出現泛化過多，不適應地圖場景的檢索的問題，這些都是要在實踐過程中需要考慮的問題。

方案

整體看，方案包括召回，排序，過濾，三個階段。

召回階段

我們調研了句子向量表示的幾種方法，選擇了算法簡單，效果和性能可以和CNN，RNN媲美的SIF（Smooth Inverse Frequency）。向量召回可以使用開源的Faiss向量搜索引擎，這裏我們使用了阿里內部的性能更好的的向量檢索引擎。

排序階段
樣本構建
原query與高頻query候選集合，計算語義相似度，選取語義相似度的TOPK，人工標註的訓練樣本。

特徵建設

1.基礎文本特徵
2.編輯距離
3.組合特徵

模型選擇

使用xgboost進行分數回歸

過濾階段
通過向量召回的query過度泛化非常嚴重，為了能夠在地圖場景下進行應用，增加了對齊模型。使用了兩種統計對齊模型giza和fastalign，實驗證明二者效果幾乎一致，但fastalign在性能上好於giza，所以選擇fastalign。

通過對齊概率和非對齊概率，對召回的結果進行進一步過濾，得到精度比較高的結果。

query改寫填補了原始query分析模塊中一些低頻表達無法滿足的空白，區別於同義詞或者糾錯的顯式query變換表達，句子的向量表示是相似query的一種隱式的表達，有其相應的優勢。

向量表示和召回也是深度學習模型逐步開始應用的嘗試。同義詞，改寫，糾錯，作為地圖中query變換主要的三種方式，以往在地圖模塊里比較分散，各司其職，也會有互相重疊的部分。在後續的迭代升級中，我們引入了統一的query變換模型進行改造，在取得收益的同時，也擺脫掉了過去很多規則以及模型耦合造成的歷史包袱。

3.2 省略

在地圖搜索場景里，有很多query包含無效詞，如果用全部query嘗試去召回很可能不能召回有效結果。如廈門市搜”湖裡區縣后高新技術園新捷創運營中心11樓1101室 縣后brt站”。這就需要一種檢索意圖，在不明顯轉義下，使用核心term進行召回目標poi候選集合，當搜索結果無果或者召回較差時起到補充召回的作用。

在省略判斷的過程中存在先驗后驗平衡的問題。省略意圖是一個先驗的判斷，但是期望的結果是能夠進行POI有效召回，和POI的召回字段的現狀密切相關。如何能夠在策略設計的過程中保持先驗的一致性，同時能夠在後驗POI中拿到相對好的效果，是做好省略模塊比較困難的地方。

原始的省略模塊主要是基於規則進行的，規則依賴的主要特徵是上游的成分分析特徵。由於基於規則擬合，模型效果存在比較大的優化空間。另外，由於強依賴成分分析，模型的魯棒性並不好。

技術改造

省略模塊的改造主要完成了規則到crf模型的升級，其中也離線應用了深度學習模型輔助樣本生成。

模型選擇

識別出來query哪些部分是核心哪些部分是可以省略的，是一個序列標註問題。在淺層模型的選型中，顯而易見地，我們使用了crf模型。

特徵建設

term特徵。使用了賦權特徵，詞性，先驗詞典特徵等。
成分特徵。仍然使用成分分析的特徵。
統計特徵。統計片段的左右邊界熵，城市分佈熵等，通過分箱進行離散化。

樣本建設

項目一期我們使用了使用線上策略粗標，外包細標的方式，構造了萬級的樣本供crf模型訓練。

但是省略query的多樣性很高，使用萬級的樣本是不夠的，在線上模型無法快速應用深度模型的情況下，我們使用了boostraping的方式，藉助深度模型的泛化能力，離線構造了大量樣本。

使用了這種方式，樣本從萬級很容易擴充到百萬級，我們仍然使用crf模型進行訓練和線上應用。

在省略模塊，我們完成了規則到機器學習的升級，引入了成分以外的其他特徵，提升了模型的魯棒性。同時並且利用離線深度學習的方式進行樣本構造的循環，提升了樣本的多樣性，使得模型能夠更加接近crf的天花板。

在後續深度模型的建模中，我們逐步擺脫了對成分分析特徵的依賴，對query到命中poi核心直接進行建模，構建大量樣本，取得了進一步的收益。

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

偽類選擇器介紹

• 偽類選擇器就是用來給超級鏈接設置不同的狀態樣式。
• 超級鏈接分為4種狀態如：正常狀態、訪問過後狀態、鼠標放上狀態、激活狀態。

偽類選擇器說明表

選擇器	描述
：link	向未被訪問的超級鏈接添加樣式,正常狀態。
：visited	向已經被訪問的超級鏈接添加樣式，訪問過後狀態。
：hover	當鼠標懸浮在超級鏈接上方時，向超級鏈接添加樣式，鼠標放上狀態。
：active	鼠標放在超級鏈接上並且點擊的一瞬間，向超級鏈接添加樣式，激活狀態。

偽類選擇器實踐

• 讓我們進入偽類選擇器實踐，實踐內容將超級鏈接4種狀態進行演示，演示效果如：將向未被訪問的超級鏈接文本顏色設置為紅色、已經被訪問的超級鏈接文本顏色設置為綠色、當鼠標懸浮在超級鏈接上文本顏色設置為紫色、用鼠標點擊超級鏈接的一瞬間文本顏色設置為藍色。

• 代碼塊

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>偽類選擇器</title>
    <style>
        a:link{
            color:red;
        }
        a:visited{
            color: lime;
        }
        a:hover{
            color: purple;
        }
        a:active{
            color: blue;
        }
    </style>
</head>
  
<body>
    <a href="https://www.cnblogs.com/lq000122/">微笑是最初的信仰</a>
</body>
</html>

• 正常狀態結果圖

• 鼠標放上狀態結果圖

• 激活狀態結果圖

• 訪問過後狀態

總結

• 超級鏈接的不同狀態他其實是由順序，也就是說偽類選擇器設置其實是順序的，如果按照偽類選擇器的順序，那麼設置的樣式就不會被渲染。
• 順序：link、visited、hover 、active。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

JMM與問題引入

為啥先說JMM,因為CAS的實現類中維護的變量都被volatile修飾, 這個volatile 是遵循JMM規範(不是百分百遵循,下文會說)實現的保證多線程併發訪問某個變量實現線程安全的手段

一連串的知識點慢慢縷

首先說什麼是JMM, JMM就是大家所說的java的內存模型, 它是人們在邏輯上做出的劃分, 或者可以將JMM當成是一種規範, 有哪些規範呢? 如下

1. 可見性: 某一個線程對內存中的變量做出改動后,要求其他的線程在第一事件內馬上馬得到通知,在CAS的實現中, 可見性其實是通過不斷的while循環讀取而得到的通知, 而不是被動的得到通知
2. 原子性: 線程在執行某個操作的時,要麼一起成功,要麼就一起失敗
3. 有序性: 為了提高性能, 編譯器處理器會進行指令的重排序, 源碼-> 編譯器優化重排 -> 處理器優化重排 -> 內存系統重排 -> 最終執行的命令

JVM運行的實體是線程, 每一個線程在創建之後JVM都會為其創建一個工作空間, 這個工作空間是每一個線程之間的私有空間, 並且任何兩條線程之間的都不能直接訪問到對方的工作空間, 線程之間的通信,必須通過共享空間來中轉完成

JMM規定所有的變量全部存在主內存中,主內存是一塊共享空間,那麼如果某個線程相對主內存中共享變量做出修改怎麼辦呢? 像下面這樣:

1. 將共享變量的副本拷貝到工作空間中
2. 對變量進行賦值修改
3. 將工作空間中的變量寫回到內存中

JMM還規定如下:

• 任何線程在解鎖前必須將工作空間的共享變量立即刷新進內存中
• 線程在加鎖前必須讀取主內存中的值更新到自己的工作空間中
• 加鎖和解鎖是同一把鎖

問題引入

這時候如果多個線程併發按照上面的三步走去訪問主內存中的共享變量的話就會出現線程安全性的問題, 比如說 現在主內存中的共享變量是c=1, 有AB兩個線程去併發訪問這個c變量, 都想進行c++, 現在A將c拷貝到自己的工作空間進行c++, 於是c=2 , 於此同時線程B也進行c++, c在B的工作空間中=2, AB線程將結果寫回工作空間最終的結果就是2, 而不是我們預期的3

相信怎麼解決大家都知道, 就是使用JUC,中的原子類就能規避這個問題

而原子類的底層實現使用的就是CAS技術

什麼是CAS

CAS(compare and swap) 顧名思義: 比較和交換,在JUC中原子類的底層使用的都是CAS無鎖實現線程安全,是一門很炫的技術

如下面兩行代碼, 先比較再交換, 即: 如果從主內存中讀取到的值為4就將它更新為2019

        AtomicInteger atomicInteger = new AtomicInteger(4);
        atomicInteger.compareAndSet(4,2019);

跟進AtomicInteger的源碼如下, 底層維護着一個int 類型的 變量, (當然是因為我選擇的原來類是AtomicInteger類型), 並且這個int類型的值被 volatile 修飾

    private volatile int value;

    /**
     * Creates a new AtomicInteger with the given initial value.
     *
     * @param initialValue the initial value
     */
    public AtomicInteger(int initialValue) {
        value = initialValue;
    }

什麼是volatile

volatile是JVM提供的輕量的同步機制, 為什麼是輕量界別呢? , 剛才在上面說了JMM規範中提到了三條特性, 而JVM提供的volatile僅僅滿足上面的規範中的 2/3, 如下:

1. 保證可見性
2. 不保證原子性
3. 禁止指令重排序

單獨的volatile是不能滿足原子性的,即如下代碼在多線程併發訪問的情況下依然會出現線程安全性問題

private volatile int value;
 
public void add(){
  value++;   
}

那麼JUC的原子類是如何實現的 可以滿足原子性呢? 於是就不得不說本片博文的主角, CAS

CAS源碼跟進

我們跟進AtomicInteger中的先遞增再獲取的方法 incrementAndGet()

    public final int incrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
    }

通過代碼我們看到調用了Unsafe類來實現

什麼是Unsafe類?

進入Unsafe類,可以看到他裏面存在大量的 native方法,這些native方法全部是空方法,

這個unsafe類其實相當於一個後門,他是java去訪問調用系統上 C C++ 函數類庫的方法 如下圖

繼續跟進這個方法incrementAndGet() 於是我們就來到了我們的主角方法, 關於這個方法倒是不難理解,主要是搞清楚方法中的var12345到底代表什麼就行, 如下代碼+註釋

var1: 上一個方法傳遞進來的: this,即當前對象
var2: 上一個方法傳遞進來的valueOffset, 就是內存地址偏移量
      通過這個內存地址偏移量我能精確的找到要操作的變量在內存中的地址
      
var4: 上一個方法傳遞進來的1, 就是每次增長的值
var5: 通過this和內存地址偏移量讀取出來的當前內存中的目標值
public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

        return var5;
    }

注意它用的是while循環, 相對if(flag){} 這種寫法會多一次判斷, 整體的思路就是 在進行修改之前先進行一次比較,如果讀取到的當前值和預期值是相同的,就自增,否則的話就繼續輪詢修改

小總結

通過上面的過程, 其實就能總結出CAS的底層實現原理

• volatile
• 自旋鎖
• unsafe類

補充: CAS通過Native方法的底層實現,本質上是操作系統層面上的CPU的併發原語,JVM會直接實現出彙編層面的指令,依賴於硬件去實現, 此外, 對於CPU的原語來說, 有兩條特性1,必定連續, 2.不被中斷

CAS的優缺點

優點:

它的底層我們看到了通過do-while 實現的自旋鎖來實現, 就省去了在多個線程之間進行切換所帶來的額外的上下文切換的開銷

缺點:

1. 通過while循環不斷的嘗試獲取, 省去了上下文切換的開銷,但是佔用cpu的資源
2. CAS只能保證一個共享變量的原子性, 如果存在多個共享變量的話不得不加鎖實現
3. 存在ABA問題

ABA問題

什麼是ABA問題

我們這樣玩, 還是AB兩個線程, 給AtomicInteger賦初始值0

A線程中的代碼如下:

        Thread.sleep(3000);
        atomicInteger.compareAndSet(0,2019);

B線程中的代碼如下:

        atomicInteger.compareAndSet(0,1);
        atomicInteger.compareAndSet(1,0);

AB線程同時啟動, 雖然最終的結果A線程能成果的將值修改成2019,,但是它不能感知到在他睡眠過程中B線程對數據進行過改變, 換句話說就是A線程被B線程欺騙了

ABA問題的解決— AtomicStampedRefernce.java

帶時間戳的原子引用, 實現的機制就是通過 原子引用+版本號來完成, 每次對指定值的修改相應的版本號會加1, 實例如下

        // 0表示初始化, 1表示初始版本號
        AtomicStampedReference<Integer> reference = new AtomicStampedReference<>(0, 1);
        reference.getStamp(); // 獲取版本號
        reference.attemptStamp(1,2); // 期待是1, 如果是1就更新為2

原子引用

JUC中我們可以找到像AtomicInteger這樣已經定義好了實現類, 但是JUC沒有給我們提供類似這樣 AtomicUser或者 AtomicProduct 這樣自定義類型的原子引用類型啊, 不過java仍然是提供了後門就是 原子引用類型

使用實例:

        User user  = getUserById(1);
        AtomicReference<User> userAtomicReference = new AtomicReference<User>();
        user.setUsername("張三");
        userAtomicReference.compareAndSet(user,user);

歡迎關注我, 會繼續更新筆記

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

　　之前寫過一篇:本來是用來閑來分享一下自己的思維方式，時至今日發現居然有些人正在使用了，本着對代碼負責人的態度，對代碼部分已知bug進行了修改，並增加了若干功能，如立即啟動，實時停止等功能，新增加的功能會在這一篇做詳細的說明。

　　提到分佈式任務調度，市面上本身已經有一些框架工具可以使用，但是個人覺得功能做的都太豐富，架構都過於複雜，所以才有了我重複造輪子。個人喜歡把複雜的問題簡單化，利用有限的資源實現竟可能多的功能。因為有幾個朋友問部署方式，這裏再次強調下：我的這個服務可以直接打成jar放在自己本地倉庫，然後依賴進去，或者直接copy代碼過去，當成自己項目的一部分就可以了。也就是說跟隨你們自己的項目啟動，所以我這裏也沒有寫界面。下面先談談怎麼基於上次的代碼實現任務立即啟動吧!

　　調度和自己服務整合後部署圖抽象成如下:

　　

 

 

 　　用戶在前端點擊立即請求按鈕，通過各種負載均衡軟件或者設備，到達某台機器的某個帶有本調度框架的服務，然後進行具體的執行，也就是說這個立即啟動就是一個最常見最簡單的請求，沒有過多複雜的問題（比如多節點會不會重複執行這些）。最簡單的辦法，當用戶請求過來直接用一個線程或者線程池執行用戶點的那個任務的邏輯代碼就行了，當然我這裏沒有那麼粗暴，現有的調度代碼資源如下:

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 創建任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 可以明確知道最多只會運行2個線程，直接使用系統自帶工具就可以了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工作線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池，初始線程數量corePoolSize，線程池等待隊列大小queueSize，當初始線程都有任務，並且等待隊列滿后
         * 線程數量會自動擴充最大線程數maxSize，當新擴充的線程空閑60s后自動回收.自定義線程池是因為Executors那幾個線程工具
         * 各有各的弊端，不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)才開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不該自己拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態，如果更新成功，其他節點就不會更新成功。如果其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務，version必然和查出來時候的version不一樣了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是因為上面樂觀鎖已經更新了版本，會導致後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就可以從延時隊列拿出任務對象,然後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了，這裏沒必要了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完后刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務，如果父任務失敗了，子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改成執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態，不然這裏可能和恢複線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下，避免上面update修改后version不同步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務后再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

   
}

　　上面主要就是三組線程，Loader負責加載將要執行的任務放入本地的任務隊列，Boss線程負責取出任務隊列的任務，然後分配Worker線程池的一個線程去執行。由上面的代碼可以看到如果要立即執行，其實只需要把一個延時為0的任務放入任務隊列，等着Boss線程去取然後分配給worker執行就可以實現了，代碼如下:

    /**
     * 立即執行任務，就是設置一下延時為0加入任務隊列就好了，這個可以外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

　　啟動不用再多說，下面介紹一下停止任務，根據面向對象的思維，用戶要想停止一個任務，最終執行停止任務的就是正在執行任務的那個節點。停止任務有兩種情況，第一種任務沒有正在運行如何停止，第二種是任務正在運行如何停止。第一種其實直接改變一下任務對象的狀態為停止就行了，不必多說。下面主要考慮如何停止正在運行的任務，細心的朋友可能已經發現上面代碼和之前那一篇代碼有點區別，之前用的Runnble作為線程實現接口，這個用了Callable，其實在java中停止線程池中正在運行的線程最常用的就是直接調用future的cancel方法了，要想獲取到這個future對象就需要將以前實現Runnbale改成實現Callable，然後提交到線程池由execute改成submit就可以了，然後每次提交到線程池得到的future對象使用taskId一起保存在一個map中，方便根據taskId隨時找到。當然任務執行完后要及時刪除這個map里的任務，以免常駐其中導致內存溢出。停止任務的請求流程如下

　　

 

 

　　圖還是原來的圖，但是這時候情況不一樣了，因為停止任務的時候假如當前正在執行這個任務的節點處於服務1，負載均衡是不知道要去把你引到服務1的，他可能會引入到服務2，那就悲劇了，所以通用的做法就是停止請求過來不管落到哪個節點上，那個節點就往一個公用的mq上發一個帶有停止任務業務含義的消息，各個節點訂閱這個消息，然後判斷都判斷任務在不在自己這裏執行，如果在就執行停止操作。但是這樣勢必讓我們的調度服務又要依賴一個外部的消息隊列服務，就算很方便的就可以引入一個外部的消息隊列，但是你真的可以駕馭的了嗎，消息丟了咋辦，重複發送了咋辦，消息服務掛了咋辦，網絡斷了咋辦，又引入了一大堆問題，那我是不是又要寫n篇文章來分別解決這些問題。往往現實卻是就是這麼殘酷，你解決了一個問題，引入了更多的問題，這就是為什麼bug永遠改不完的道理了。當然這不是我的風格，我的風格是利用有限的資源做盡可能多的事情（可能是由於我工作的企業都是那種資源貧瘠的，養成了我這種習慣，土豪公司的程序員請繞道，哈哈）。

　　簡化一下問題:目前的問題就是如何讓正在執行任務的節點知道，然後停止正在執行的這個任務，其實就是這個停止通知如何實現。這不免讓我想起了12306網站上買票，其實我們作為老百姓多麼希望12306可以在有票的時候發個短信通知一下我們，然後我們上去搶，但是現實卻是，你要麼使用軟件一直刷，要麼是自己隔一段時間上去瞄一下有沒有票。如果把有票了給我們發短信通知定義為異步通知，那麼這種我們要隔一段時間自己去瞄一下的方式就是同步輪訓。這兩種方式都能達到告知的目的，關鍵的區別在於你到底有沒有時間去一直去瞄，不過相比於可以回家，這些時間都是值得的。個人認為軟件的設計其實就是一個權衡是否值得的過程。如果約定了不使用外部消息隊列這種異步通知的方式，那麼我們只能使用同步輪訓的方式了。不過正好我們的任務調度本身已經有一個心跳機制，沒隔一段時間就去更新一下節點狀態，如果我們把用戶的停止請求作為命令信息更新到每個節點的上，然後隨着心跳獲取到這個節點的信息，然後判斷這個命令，做相應的處理是不是就可以完美解決這個問題。值得嗎？很明顯是值得的，我們只是在心跳邏輯上加一個小小的副作用就實現了通知功能了。代碼如下

package com.rdpaas.task.common;

/**
 * @author rongdi
 * @date 2019/11/26
 */
public enum NotifyCmd {

    //沒有通知，默認狀態
    NO_NOTIFY(0),
    //開啟任務(Task)
    START_TASK(1),
    //修改任務(Task)
    EDIT_TASK(2),
    //停止任務(Task)
    STOP_TASK(3);

    int id;

    NotifyCmd(int id) {
        this.id = id;
    }

    public int getId() {
        return id;
    }

    public static NotifyCmd valueOf(int id) {
        switch (id) {
            case 1:
                return START_TASK;
            case 2:
                return EDIT_TASK;
            case 3:
                return STOP_TASK;
            default:
                return NO_NOTIFY;
        }
    }

}

package com.rdpaas.task.handles;

import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.utils.SpringContextUtil;

/**
 * @author: rongdi
 * @date:
 */
public interface NotifyHandler<T> {

    static NotifyHandler chooseHandler(NotifyCmd notifyCmd) {
        return SpringContextUtil.getByTypeAndName(NotifyHandler.class,notifyCmd.toString());
    }

    public void update(T t);

}

package com.rdpaas.task.handles;

import com.rdpaas.task.scheduler.TaskExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author: rongdi
 * @date:
 */
@Component("STOP_TASK")
public class StopTaskHandler implements NotifyHandler<Long> {

    @Autowired
    private TaskExecutor taskExecutor;

    @Override
    public void update(Long taskId) {
        taskExecutor.stop(taskId);
    }

}

class HeartBeat implements Runnable {
        @Override
        public void run() {
            for(;;) {
                try {
                    /**
                     * 時間到了就可以從延時隊列拿出節點對象，然後更新時間和序號，
                     * 最後再新建一個超時時間為心跳時間的節點對象放入延時隊列，形成循環的心跳
                     */
                    DelayItem<Node> item = heartBeatQueue.take();
                    if(item != null && item.getItem() != null) {
                        Node node = item.getItem();
                        handHeartBeat(node);
                    }
                    heartBeatQueue.offer(new DelayItem<>(config.getHeartBeatSeconds() * 1000,new Node(config.getNodeId())));
                } catch (Exception e) {
                    logger.error("task heart beat error,cause by:{} ",e);
                }
            }
        }
    }

    /**
     * 處理節點心跳
     * @param node
     */
    private void handHeartBeat(Node node) {
        if(node == null) {
            return;
        }
        /**
         * 先看看數據庫是否存在這個節點
         * 如果不存在：先查找下一個序號，然後設置到node對象中，最後插入
         * 如果存在：直接根據nodeId更新當前節點的序號和時間
         */
        Node currNode= nodeRepository.getByNodeId(node.getNodeId());
        if(currNode == null) {
            node.setRownum(nodeRepository.getNextRownum());
            nodeRepository.insert(node);
        } else  {
            nodeRepository.updateHeartBeat(node.getNodeId());
            NotifyCmd cmd = currNode.getNotifyCmd();
            String notifyValue = currNode.getNotifyValue();
            if(cmd != null && cmd != NotifyCmd.NO_NOTIFY) {
                /**
                 * 藉助心跳做一下通知的事情，比如及時停止正在執行的任務
                 * 根據指令名稱查找Handler
                 */
                NotifyHandler handler = NotifyHandler.chooseHandler(currNode.getNotifyCmd());
                if(handler == null || StringUtils.isEmpty(notifyValue)) {
                    return;
                }
                /**
                 * 執行操作
                 */
                handler.update(Long.valueOf(notifyValue));
            }
            
        }


    }

　　最終的任務調度代碼如下:

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 創建任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 可以明確知道最多只會運行2個線程，直接使用系統自帶工具就可以了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工作線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池，初始線程數量corePoolSize，線程池等待隊列大小queueSize，當初始線程都有任務，並且等待隊列滿后
         * 線程數量會自動擴充最大線程數maxSize，當新擴充的線程空閑60s后自動回收.自定義線程池是因為Executors那幾個線程工具
         * 各有各的弊端，不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)才開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不該自己拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態，如果更新成功，其他節點就不會更新成功。如果其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務，version必然和查出來時候的version不一樣了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是因為上面樂觀鎖已經更新了版本，會導致後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就可以從延時隊列拿出任務對象,然後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了，這裏沒必要了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完后刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務，如果父任務失敗了，子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改成執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態，不然這裏可能和恢複線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下，避免上面update修改后version不同步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務后再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

    /**
     * 立即執行任務，就是設置一下延時為0加入任務隊列就好了，這個可以外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

    /**
     * 立即停止正在執行的任務，留給外部調用的方法
     * @param taskId
     * @return
     */
    public boolean stopNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        if(task == null) {
            return false;
        }
        /**
         * 該任務不是正在執行，直接修改task狀態為已完成即可
         */
        if(task.getStatus() != TaskStatus.DOING) {
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
            return true;
        }
        /**
         * 該任務正在執行，使用節點配合心跳發布停用通知
         */
        int n = nodeRepository.updateNotifyInfo(NotifyCmd.STOP_TASK,String.valueOf(taskId));
        return n > 0;
    }

    /**
     * 立即停止正在執行的任務，這個不需要自己調用，是給心跳線程調用
     * @param taskId
     * @return
     */
    public boolean stop(Long taskId) {
        Task task = taskRepository.get(taskId);
        /**
         * 不是自己節點的任務，本節點不能執行停用
         */
        if(task == null || !config.getNodeId().equals(task.getNodeId())) {
            return false;
        }
        /**
         * 拿到正在執行任務的future，然後強制停用，並刪除doingFutures的任務
         */
        Future future = doingFutures.get(taskId);
        boolean flag =  future.cancel(true);
        if(flag) {
            doingFutures.remove(taskId);
            /**
             * 修改狀態為已停用
             */
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
        }
        /**
         * 重置通知信息，避免重複執行停用通知
         */
        nodeRepository.resetNotifyInfo(NotifyCmd.STOP_TASK);
        return flag;
    }
}

　　好吧，其實實現很簡單，關鍵在於思路，不BB了，詳細代碼見： 在下告辭！

　　

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

下面分享關於位操作的一些筆記：

一、位操作簡單介紹

首先，以下是按位運算符：

在嵌入式編程中，常常需要對一些寄存器進行配置，有的情況下需要改變一個字節中的某一位或者幾位，但是又不想改變其它位原有的值，這時就可以使用按位運算符進行操作。下面進行舉例說明，假如有一個8位的TEST寄存器：

當我們要設置第0位bit0的值為1時，可能會這樣進行設置：

TEST = 0x01;

但是，這樣設置是不夠準確的，因為這時候已經同時操作到了高7位：bit1~bit7，如果這高7位沒有用到的話，這麼設置沒有什麼影響；但是，如果這7位正在被使用，結果就不是我們想要的了。

在這種情況下，我們就可以借用按位操作運算符進行配置。

對於二進制位操作來說，不管該位原來的值是0還是1，它跟0進行&運算，得到的結果都是0，而跟1進行&運算，將保持原來的值不變；不管該位原來的值是0還是1，它跟1進行|運算，得到的結果都是1，而跟0進行|運算，將保持原來的值不變。

所以，此時可以設置為：

TEST = TEST | 0x01;

其意義為：TEST寄存器的高7位均不變，最低位變成1了。在實際編程中，常改寫為：

TEST |= 0x01;

這種寫法可以一定程度上簡化代碼，是 C 語言常用的一種編程風格。設置寄存器的某一位還有另一種操作方法，以上的等價方法如：

TEST |= (0x01 << 0);

第幾位要置1就左移幾位。

同樣的，要給TEST的低4位清0，高4位保持不變，可以進行如下配置：

TEST &= 0xF0;

二、嵌入式中位操作一些常見用法

1、一個32bit數據的位、字節讀取操作

（1）獲取單字節：

#define GET_LOW_BYTE0(x)    ((x >>  0) & 0x000000ff)    /* 獲取第0個字節 */
#define GET_LOW_BYTE1(x)    ((x >>  8) & 0x000000ff)    /* 獲取第1個字節 */
#define GET_LOW_BYTE2(x)    ((x >> 16) & 0x000000ff)    /* 獲取第2個字節 */
#define GET_LOW_BYTE3(x)    ((x >> 24) & 0x000000ff)    /* 獲取第3個字節 */

示例：

（2）獲取某一位：

#define GET_BIT(x, bit) ((x & (1 << bit)) >> bit)   /* 獲取第bit位 */

示例：

2、一個32bit數據的位、字節清零操作

（1）清零某個字節：

#define CLEAR_LOW_BYTE0(x)  (x &= 0xffffff00)   /* 清零第0個字節 */
#define CLEAR_LOW_BYTE1(x)  (x &= 0xffff00ff)   /* 清零第1個字節 */
#define CLEAR_LOW_BYTE2(x)  (x &= 0xff00ffff)   /* 清零第2個字節 */
#define CLEAR_LOW_BYTE3(x)  (x &= 0x00ffffff)   /* 清零第3個字節 */

示例：

（2）清零某一位：

#define CLEAR_BIT(x, bit)   (x &= ~(1 << bit))  /* 清零第bit位 */

示例：

3、一個32bit數據的位、字節置1操作

（1）置某個字節為1：

#define SET_LOW_BYTE0(x)    (x |= 0x000000ff)   /* 第0個字節置1 */   
#define SET_LOW_BYTE1(x)    (x |= 0x0000ff00)   /* 第1個字節置1 */   
#define SET_LOW_BYTE2(x)    (x |= 0x00ff0000)   /* 第2個字節置1 */   
#define SET_LOW_BYTE3(x)    (x |= 0xff000000)   /* 第3個字節置1 */

示例：

（2）置位某一位：

#define SET_BIT(x, bit) (x |= (1 << bit))   /* 置位第bit位 */

4、判斷某一位或某幾位連續位的值

（1）判斷某一位的值

舉例說明：判斷0x68第3位的值。

也就是說，要判斷第幾位的值，if里就左移幾位（當然別過頭了）。在嵌入式編程中，可通過這樣的方式來判斷寄存器的狀態位是否被置位。

（2）判斷某幾位連續位的值

/* 獲取第[n:m]位的值 */
#define BIT_M_TO_N(x, m, n)  ((unsigned int)(x << (31-(n))) >> ((31 - (n)) + (m)))

示例：

這是一個查詢連續狀態位的例子，因為有些情況不止有0、1兩種狀態，可能會有多種狀態，這種情況下就可以用這種方法來取出狀態位，再去執行相應操作。

以上是對32bit數據的一些操作進行總結，其它位數的數據類似，可根據需要進行修改。

三、STM32寄存器配置

STM32有幾套固件庫，這些固件庫函數以函數的形式進行1層或者多層封裝（軟件開發中很重要的思想之一：分層思想），但是到了最裏面的一層就是對寄存器的配置。我們平時都比較喜歡固件庫來開發，大概是因為固件庫用起來比較簡單，用固件庫寫出來的代碼比較容易閱讀。最近一段時間一直在配置寄存器，越發地發現使用寄存器來進行一些外設的配置也是很容易懂的。使用寄存器的方式編程無非就是往寄存器的某些位置1、清零以及對寄存器一些狀態位進行判斷、讀取寄存器的內容等。

這些基本操作在上面的例子中已經有介紹，我們依舊以實例來鞏固上面的知識點（以STM32F1xx為例）：

（1）寄存器配置

看一下GPIO功能的端口輸出數據寄存器 (GPIOx_ODR) (x=A..E) ：

假設我們要讓PA10引腳輸出高、輸出低，可以這麼做：

方法一：

GPIOA->ODR |= 1 << 10;      /* PA10輸出高（置1操作） */
GPIOA->ODR &= ~（1 << 10）;  /* PA10輸出低（清0操作） */

也可用我們上面的置位、清零的宏定義：

SET_BIT(GPIOA->ODR, 10);    /* PA10輸出高（置1操作） */
CLEAR_BIT(GPIOA->ODR, 10);  /* PA10輸出低（清0操作） */

方法二：

GPIOA->ODR |= (uint16_t)0x0400;   /* PA10輸出高（置1操作） */
GPIOA->ODR &= ~(uint16_t)0x0400;  /* PA10輸出低（清0操作） */

貌似第二種方法更麻煩？還得去細心地去構造一個數據。

但是，其實第二種方法其實是ST推薦我們用的方法，為什麼這麼說呢？因為ST官方已經把這些我們要用到的值給我們配好了，在stm32f10x.h中：

這個頭文件中存放的就是外設寄存器的一些位配置。

所以我們的方法二等價於：

GPIOA->ODR |= GPIO_ODR_ODR10;   /* PA10輸出高（置1操作） */
GPIOA->ODR &= ~GPIO_ODR_ODR10;  /* PA10輸出低（清0操作） */

兩種方法都是很好的方法，但方法一似乎更好理解。

配置連續幾位的方法也是一樣的，就不介紹了。簡單介紹配置不連續位的方法，以TIM1的CR1寄存器為例：

設置CEN位為1、設置CMS[1:0]位為01、設置CKD[1:0]位為10：

TIM1->CR1 |= （0x1 << 1)| (0x1 << 5) |(0x2 << 8);

這是組合的寫法。當然，像上面一樣拆開來寫也是可以的。

（2）判斷標誌位

以狀態寄存器(USART_SR) 為例：

判斷RXNE是否被置位：

/* 數據寄存器非空，RXNE標誌置位 */
if (USART1->SR & (1 << 5))
{
    /* 其它代碼 */
    
    USART1->SR &= ~(1 << 5);  /* 清零RXNE標誌 */
}

或者：

/* 數據寄存器非空，RXNE標誌置位 */
if (USART1->SR & USART_SR_RXNE)
{
    /* 其它代碼 */
    
    USART1->SR &= ~USART_SR_RXNE;  /* 清零RXNE標誌 */
}

四、總結

以上就是本次關於位操作的一點總結筆記，有必要掌握。雖然說在用STM32的時候有庫函數可以用，但是最接近芯片內部原理的還是寄存器。有可能之後有用到其它芯片沒有像ST這樣把寄存器相關配置封裝得那麼好，那就不得不直接操控寄存器了。

此外，使用庫函數的方式代碼佔用空間大，用寄存器的話，代碼佔用空間小。之前有個需求，我們能用的Flash的空間大小隻有4KB，遇到類似這樣的情況就不能那麼隨性的用庫函數了。

最後，應用的時候當然是怎麼簡單就怎麼用。學從“難”處學，用從易處用，與君共勉~

END：以上筆記中如有錯誤，歡迎指出！謝謝

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※3c收購,鏡頭 收購有可能以全新價回收嗎?

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?