分類: 3C資訊

  • HashMap7淺析

    HashMap7淺析

    一、概述

      HashMap,基於哈希結構的Map接口的一個實現,無序,允許null鍵值對,線程不安全的。可以使用集合工具類Collections中的synchronizedMap方法,去創建一個線程安全的集合map。

      在jdk1.7中,HashMap主要是基於 數組+鏈表 的結構實現的。鏈表的存在主要是解決 hash 衝突而存在的。插入數據的時候,計算key的hash值,取得存儲的數組下標,如果衝突已有元素,則會在衝突地址上生成個鏈表,再通過key的比較,鏈表是否已存在,存在則覆蓋,不存在則鏈表上添加。這種方式,如果存在大量衝突的時候,會導致鏈表過長,那麼直接導致的就是犧牲了查詢和添加的效率。所以在jdk1.8版本之後,使用的就是 數組 + 鏈表 + 紅黑樹,當鏈表長度超過 8(實際加上初始的節點,整個有效長度是 9) 的時候,轉為紅黑樹存儲。

      本文中內容,主要基於jdk1.7版本,單線程環境下使用的HahsMap沒有啥問題,但是當在多線程下使用的時候,則可能會出現併發異常,具體表象是CPU會直線上升100%。下面是主要介紹相關的存取以及為什麼會出現線程安全性問題。

    二、結構

      

      HashMap默認初始化size=16的哈希數組,然後通過計算待存儲的key的hash值,去計算得到哈希數組的下標值,然後放入鏈表中(新增節點或更新)。鏈表的存在即是解決hash衝突的。

    三、源碼實現分析

      1、存儲具體數據的table數組:

          

        Entry為HashMap中的靜態內部類,其具體結構如下圖

          

        key、value屬性就是存儲鍵值對的,next則是指向鏈表的下一個元素節點。

         2、 默認初始化方法:

        

        默認構造方法,不對table進行初始化new(真正初始化動作放在put中,後面會看到),只是設置參數的默認值,hashmap長度和table長度初始化成DEFAULT_INITIAL_CAPACITY(16),加載因子loadFactor默認DEFAULT_LOAD_FACTOR(0.75f,至於為什麼是0.75,這個可以參見 )。

        加載因子:默認情況下,16*0.75=12,也就是在存儲第13個元素的時候,就會進行擴容(jdk1.7的threshold真正計算放在第一次初始化中,後面會再提及)。此元素的設置,直接影響到的是key的hash衝突問題。

      3、put方法

     public V put(K key, V value) {
       
    if (table == EMPTY_TABLE) { inflateTable(threshold); } if (key == null) return putForNullKey(value); int hash = hash(key); int i = indexFor(hash, table.length); for (Entry<K,V> e = table[i]; e != null; e = e.next) { Object k; if (e.hash == hash && ((k = e.key) == key || key.equals(k))) { V oldValue = e.value; e.value = value; e.recordAccess(this); return oldValue; } } modCount++; addEntry(hash, key, value, i); return null; }

      3.1、EMPTY_TABLE是HashMap中的一個靜態的空的Entry數組,table也是HashMap的一個屬性,默認就是EMPTY_TABLE(這兩句可參見上面源碼),table就是我們真正數據存儲使用的。
      3.2、前面提及,無參構造的時候,並未真正完成對HashMap的初始化new操作,而僅僅只是設置幾個常量,所以在第一次put數據的時候,table是空的。則會進入下面的初始化table方法中。

    if (table == EMPTY_TABLE) {
        inflateTable(threshold);
    }
    
    private void inflateTable(int toSize) {
        // Find a power of 2 >= toSize
        int capacity = roundUpToPowerOf2(toSize);
    
        threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1); //計算加載因子,默認情況下結果為12
        table = new Entry[capacity];  //真正的初始化table數組
        initHashSeedAsNeeded(capacity);
    }

      3.3、key的null判斷

    if (key == null)
        return putForNullKey(value);
    
    private V putForNullKey(V value) {
        for (Entry<K,V> e = table[0]; e != null; e = e.next) {
            if (e.key == null) {
                V oldValue = e.value;
                e.value = value;
                e.recordAccess(this);
                return oldValue;
            }
        }
        modCount++;
        addEntry(0, null, value, 0);
        return null;
    }
    
    void addEntry(int hash, K key, V value, int bucketIndex) {
        if ((size >= threshold) && (null != table[bucketIndex])) {
            resize(2 * table.length);
            hash = (null != key) ? hash(key) : 0;
            bucketIndex = indexFor(hash, table.length);
        }
    
        createEntry(hash, key, value, bucketIndex);
    }
    
    void createEntry(int hash, K key, V value, int bucketIndex) {
        Entry<K,V> e = table[bucketIndex];
        table[bucketIndex] = new Entry<>(hash, key, value, e);
        size++;
    }

      具體步驟解析:

        1、key為null,取出table[0]的鏈表結構Enrty,如果取出的元素不為null,則對其進行循環遍歷,查找其中是否存在key為null的節點元素。

           2、如果存在key == null的節點,則使用新的value去更新節點的oldValue,並且將oldValue返回。

        3、如果不存在key == null的元素,則執行新增元素addEntry方法:

          (1)判斷是否需要擴容,size為當前數組table中,已存放的Entry鏈表個數,更直接點說,就是map.size()方法的返回值。threshold上面的真正初始化HashMap的時候已經提到,默認情況下,計算得到 threshold=12。若同時滿足  (size >= threshold) && (null != table[bucketIndex]) ,則對map進行2倍的擴容,然後對key進行重新計算hash值和新的數組下標。

          (2)創建新的節點原色createEntry方法,首先獲取table數組中下標為bucketIndex的鏈表的表頭元素,然後新建個Entry作為新的表頭,並且新表頭其中的next指向老的表頭數據。

      3.4、key不為null的存儲  
        原理以及過程上通key==null的大體相同,只不過,key==null的時候,固定是獲取table[0]的鏈表進行操作,而在不為key != null的時候,下標位置是通過
      int hash = hash(key); int i = indexFor(hash, table.length); 計算得到的

      static int indexFor(int h, int length) {
            // assert Integer.bitCount(length) == 1 : "length must be a non-zero power of 2";
            return h & (length-1);
        }

      很清晰的就能看明白,先計算key的hash,然後與當前table的長度進行相與,這樣計算得到待存放數據的下標。得到下標后,過程就與key==null一致了,遍歷是否存在,存在則更新並返回oldVlaue,不存在則新建Entry。

      4、get方法

     public V get(Object key) {
            if (key == null)
                return getForNullKey();
            Entry<K,V> entry = getEntry(key);
    
            return null == entry ? null : entry.getValue();
        }
        如果key == null,則調用getForNullKey方法,遍歷table[0]處的鏈表。
    private V getForNullKey() {
            if (size == 0) {
                return null;
            }
            for (Entry<K,V> e = table[0]; e != null; e = e.next) {
                if (e.key == null)
                    return e.value;
            }
            return null;
        }

      如果key != null,則調用getEntry,根據key計算得到在table數組中的下標,獲取鏈表Entry,然後遍歷查找元素,key相等,則返回該節點元素。

     final Entry<K,V> getEntry(Object key) {
            if (size == 0) {
                return null;
            }
    
            int hash = (key == null) ? 0 : hash(key);
            for (Entry<K,V> e = table[indexFor(hash, table.length)];
                 e != null;
                 e = e.next) {
                Object k;
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    return e;
            }
            return null;
        }

    四、線程不安全分析

      上述,主要淺析了下HashMap的存取過程,HashMap的線程安全性問題主要也就是在上述的擴容resize方法上,下面來看看在高併發下,擴容后,是如何引起100%問題的。

      1、在進行新元素 put 的時候,這在上面中的3.3的代碼片段中可以查看,addEntry 添加新節點的時候,會計算是否需要擴容處理:(size >= threshold) && (null != table[bucketIndex]) 。

      2、如果擴容的話,會接下來調用 resize 方法

     void resize(int newCapacity) {
            Entry[] oldTable = table;
            int oldCapacity = oldTable.length;
            if (oldCapacity == MAXIMUM_CAPACITY) {
                threshold = Integer.MAX_VALUE;
                return;
            }
    
            Entry[] newTable = new Entry[newCapacity];
            //關鍵性代碼,構建新hashmap並將老的數據移動過來
            transfer(newTable, initHashSeedAsNeeded(newCapacity));
            table = newTable;
            threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
        }

      3、其中,出現100%問題的關鍵就是上面的 transfer 方法,新建hashmap移動複製老數據

     1  void transfer(Entry[] newTable, boolean rehash) {
     2         int newCapacity = newTable.length;
     3         for (Entry<K,V> e : table) {
     4             // 遍歷老的HashMap,當遇到不為空的節點的是,進入移動方法
     5             while(null != e) {
     6                 // 首先創建個Entry節點 指向該節點所在鏈表的下一個節點數據
     7                 Entry<K,V> next = e.next;
     8                 if (rehash) {
     9                     e.hash = null == e.key ? 0 : hash(e.key);
    10                 }
    11               // 計算老的數據在新Hashmap中的下標位置
    12                 int i = indexFor(e.hash, newCapacity);
    13              // 將新HashMap中相應位置的元素,掛載到老數據的後面(不管有無數據)
    14                 e.next = newTable[i];
    15                 // 將新HashMap中相應位置指向上面已經成功掛載新數據的老數據
    16              newTable[i] = e;
    17              // 移動到鏈表節點中的下一個數據,繼續複製節點
    18                 e = next;
    19             }
    20         }
    21     }    

      問題的關鍵就在上述的14、15行上,這兩行的動作,在高併發下可能就會造成循環鏈表,循環鏈表在等待下一個嘗試 get 獲取數據的時候,就悲劇了。下面舉例模擬說說這個過程:

      (1)假設目前某個位置的鏈表存儲結構為 A -> B -> C,有兩個線程同時進行擴容操作

      (2)線程1執行到第7行 Entry<K,V> next = e.next; 的時候被掛起了,此時,線程1的 e 指向 A , next 指向的是 B

      (3)線程2執行完成了整個的擴容過程,那麼此時的鏈表結構應該是變為了 C -> B -> A

      (4)線程1喚醒繼續執行,而需要操作的鏈表實際就變成了了上述線程2完成后的 C ->B -> A,下面分為幾步去完成整個操作:

          第一次循環:

            (i)執行 e.next = newTable[i] ,將 A 的 next 指向線程1的新的HashMap,由於此時無數據,所以 e.next = null

            (ii)執行 newTable[i] = e,將線程1的新的HashMap的第一個元素指向 A 

            (iii)執行e = next,移動到鏈表中的下一個元素,也就是上面的(2)中的 線程掛起的時候的 B

          第二次循環:

            (i)執行 Entry<K,V> next = e.next,此時的 e 指向 B,next指向 A

            (ii)執行 e.next = newTable[i] ,將 B 的 next 指向線程1的新的HashMap,由於此時有數據A,所以 e.next = A

            (iii)執行 newTable[i] = e,將線程1的新的HashMap的第一個元素指向 B,此時線程1的新Hashmap鏈表結構為B -> A

            (iiii)執行e = next,移動到鏈表中的下一個元素 A

          第三次循環:

            (i)執行 Entry<K,V> next = e.next,此時的 e 指向 A,next指向 null

            (ii)執行 e.next = newTable[i] ,將 A 的 next 指向線程1的新的HashMap,由於此時有數據B,所以 e.next = B

            (iii)執行 newTable[i] = e,將線程1的新的HashMap的第一個元素指向 A ,此時線程1的新Hashmap鏈表結構為 A -> B -> A

            (iiii)執行e = next,移動到鏈表中的下一個元素,已移動到鏈表結尾,結束 while 循環,完成鏈表的轉移。

      (5)上述過程中,很顯然的,最終的鏈表結構中,出現了 A -> B -> A 的循環結構。擴容完成了,剩下的等待的是get獲取的時候, getEntry 方法中 for循環e = e.next中就永遠出不來了。

      注意:擴容過程中,newTable是每個擴容線程獨有的,共享的只是每個Entry節點數據,最終的擴容是會調用 table = newTable 賦值操作完成。

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

    ※帶您來了解什麼是 USB CONNECTOR  ?

    ※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

    ※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站,提高曝光率!!

    ※綠能、環保無空污,成為電動車最新代名詞,目前市場使用率逐漸普及化

    ※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

  • 9012年了,再不會Https就老了

    9012年了,再不會Https就老了

    前言  

           合格的web後端程序員,在開發之外,必須給IIS、Nginx、各種原生web服務器上配置Https,不然你就僅僅是個碼農,本博最近將專題記錄

    • 如何為IIS,Nginx配置Https

    • 如何申請適用於生產的免費SSL證書

    本博客小試牛刀,先實操在Nginx for Docker上添加自簽名SSL證書

    為啥先倒騰自簽名SSL證書,申請公網SSL證書需要公網可識別的域名或者公網IP;

    如果有實際SSL證書, 按照本文替換即可。

    手繪Https原理

    長話短說:  目前常見的Http請求明文傳輸, 報文可被截取並篡改,請求可被偽造; 

    因此基於常見HTTP(HTTP-TCP-IP)協議棧引入SSL/TSL(Transport Secure Layer) ,HTTPS在進行加密傳輸之前會進行一次握手,確定傳輸密鑰。

    流程解讀:

    ① 傳輸密鑰是對稱密鑰,用於雙方對傳輸數據的加解密

    ② 怎麼在傳輸之前確立傳輸密鑰呢? 針對普遍的多客戶端訪問受信web服務器的場景, 提出非對稱密鑰(公鑰存於客戶端,私鑰存於web服務器),雙方能互相加解密,說明中間數據(傳輸密鑰)沒被篡改。

    ③ 再拋出疑問,怎麼認定下發的公鑰是這個web服務器匹配的密鑰?怎麼確定這個公鑰下發過程沒被截取篡改? 這就是追溯到握手階段的下發證書過程,瀏覽器內置的CA機構認定該證書是其有效下發,並通過簽名認定該證書沒被篡改,最終認定該 證書下發的公鑰是受信web服務器準確下發。

    ④ 如果面向面試記憶Https原理,恐怕有些難度,所以個人用一種 【雞生蛋還是蛋生雞】的方式向上追溯流程, 方便大家知其然更知其所以然。

    前置準備

     >   CentOS機器上安裝Docker、 Docker-Compose

     >   常規操作構建 Nginx for Docker網站, 項目結構如下: 

    ssl-docker-nginx
        ├── docker-compose.yml
        ├── nginx   
        │      └── nginx.conf
        └── site
               └── index.html

        該項目將會使用  nginx/nginx.conf、site/index.html替換Nginx鏡像默認配置文件和默認啟動頁,docker-compose.yml 如下:

    version: '2'
    services:
      server:
        image: nginx:latest
        volumes:
          - ./nginx/nginx.conf:/etc/nginx/nginx.conf
          - ./site:/usr/share/nginx/html
        ports:
        - "8080:80"

        docker-compose  up -d 啟動Nginx容器,還是那樣熟悉的味道: 【chrome默認將http連接認定為不安全

    添加SSL自簽名證書

    很明顯: web服務器需要存儲證書(內置了公鑰)和私鑰

     ① 創建自簽名證書 (什麼叫自簽名,就是自己給自己頒發 SSL證書)

    [nodotnet@gs-server-5809 ssl-docker-nginx]$ openssl req -newkey rsa:2048 -nodes -keyout nginx/my-site.com.key -x509 -days 365out nginx/my-site.com.crt

    req是證書請求的子命令,-newkey rsa:2048 -keyout nginx/my-site.com.key表示生成私鑰(PKCS8格式),

              -nodes 表示私鑰不加密,

               -x509表示輸出證書,-days365 為有效期,此後根據提示輸入證書擁有者信息;

           之後會在nginx目錄下生產2個文件, 分別是私鑰、證書

     ② 將證書和密鑰掛載到Nginx Image, 修改docker-compose.yml

    version: '2'
    services:
      server:
        image: nginx:latest
        volumes:
          - ./nginx/nginx.conf:/etc/nginx/nginx.conf
          - ./site:/usr/share/nginx/html
          - ./nginx/my-site.com.crt:/etc/nginx/my-site.com.crt    #新行 - ./nginx/my-site.com.key:/etc/nginx/my-site.com.key    #新行
        ports:
        - "8080:80"
        - "443:443"        // 容器開啟HTTPS默認的443端口

     ③  修改nginx/nginx.conf,接受Https請求

    events {
      worker_connections  4096;  ## Default: 1024
    }
    
    http {
        server {
            listen 80;
            root         /usr/share/nginx/html/;
        }
    
        server {            # 新Server接受來自443端口的Https請求
            listen              443 ssl;
            ssl_certificate     /etc/nginx/my-site.com.crt;
            ssl_certificate_key /etc/nginx/my-site.com.key;
            root        /usr/share/nginx/html;
        }
    }

    執行docker-compose down && docker-compose up -d 發起https://10.201.80.126:443請求,當前自簽名證書頒發機構不在瀏覽器內置的CA機構,所以該證書目前被瀏覽器認為是無效。

    理論上將 該自簽名證書導出,之後在 【chrome瀏覽器】-【高級設置】-【管理證書】中導入該證書,即可讓 chrome接受自簽名SSL證書。

    That‘s All,  Https作為以後web的主流配置,碼農進階資深必須掌握;後續會記錄Https & HSTS, 申請免費SSL證書,盡請關注。

     

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    ※為什麼 USB CONNECTOR 是電子產業重要的元件?

    網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司,幫您輕鬆架站!

    ※想要讓你的商品成為最夯、最多人討論的話題?網頁設計公司讓你強力曝光

    ※想知道最厲害的台北網頁設計公司推薦台中網頁設計公司推薦專業設計師”嚨底家”!!

  • 如何高效的學習技術

    如何高效的學習技術

      我們相信努力學習一定會有收穫,但是方法不當,既讓人身心疲憊,也沒有切實的回報。高中時代,我的同桌是個漂亮女同學。她的物理成績很差,雖然她非常勤奮的學習,但成績總是不理想。為了鞏固純潔的同學關係,我親密無間地輔導她的物理,發現她不知道題目考什麼。我們的教科書與試題都圍繞着考試大綱展開,看到一道題,應該先想想它在考哪些定理和公式的運用。
      不少朋友每天都閱讀技術文章,但是第二天就忘乾淨了。工作中領導和同事都認可你的溝通和技術能力,但是跳槽面試卻屢屢碰壁。面試官問技術方案,明明心裏清楚,用嘴說出來卻前言不搭后語。面試官再問底層算法,你說看過但是忘記了。他不在乎你看沒看過,答不上就是零分。正如男女相親,男方談吐瀟洒才能吸引姑娘。可是男方緊張了,平時挺能說,關鍵時候卻支支吾吾,姑娘必然認為他不行。人生充滿了許多考試,有形的和無形的,每次考試的機會只有一次。
      工作五年十年後,別人成了架構師,自己還在基層打滾,原因是什麼?職場上無法成功升遷的原因有很多,沒有持續學習、學習效果不好、無法通過心儀公司的的面試,一定是很重要的原因。
      把自己當成一台計算機,既有輸入,也要有輸出,用輸出倒逼輸入

      近些年誕生了許多新技術,比如最時髦的AI(目前還在智障階段),數學基礎是初中就接觸過的概率統計。萬丈高樓從地起,不要被新工具或者中間件迷住雙眼,一味地追新求快。基礎知識是所有技術的基石,在未來很長的時間都不會變化,應該花費足夠的時間鞏固基礎。
      以數據結構和算法為例,大家閱讀一下Java的BitSet的源碼,裏面有大量的移位操作,移位運算掌握的好,看這份源碼就沒問題。Java同步工具類AQS用到了雙向鏈表,鏈表知識不過關,肯定搞不懂它的原理。互聯網大廠都喜歡考算法,為了通過面試也要精通算法。
      以Java工程師應該掌握的知識為例,按重要程度排出六個梯度:

    • 第一梯度:計算機組成原理、數據結構和算法、網絡通信原理、操作系統原理;
    • 第二梯度:Java基礎、JVM內存模型和GC算法、JVM性能調優、JDK工具、設計模式;
    • 第三梯度:Spring系列、Mybatis、Dubbo等主流框架的運用和原理;
    • 第四梯度:MySQL(含SQL編程)、Redis、RabbitMQ/RocketMQ/Kafka、ZooKeeper等數據庫或者中間件的運用和原理;
    • 第五梯度:CAP理論、BASE理論、Paxos和Raft算法等其他分佈式理論;
    • 第六梯度:容器化、大數據、AI、區塊鏈等等前沿技術理論;

    有同學認為第五梯度應該在移到第一梯度。其實很多小公司的日活犹如古天樂一樣平平無奇,離大型分佈式架構還遠得很。學習框架和中間件的時候,順手掌握分佈式理論,效果更好。

      許多公司的招聘JD沒有設定技術人員年齡門檻,但是會加上一句“具備與年齡相當的知識的廣度與深度”。多廣才算廣,多深才算深?這是很主觀的話題,這裏不展開討論。
      如何變得更廣更深呢?突破收入上升的瓶頸,發掘自己真正的興趣
      大多數人只是公司的普通職員,收入上升的瓶頸就是升職加薪。許多IT公司會對技術人員有個評級,如果你的評級不高,那就依照晉級章程努力升級。如果你在一個小公司,收入一般,發展前景不明,準備大廠的面試就是最好的學習過程。在這些過程中,你必然學習更多知識,變得更廣更深。
      個人興趣是前進的動力之一,許多知名開源項目都源於作者的興趣。個人興趣並不局限技術領域,可以是其他學科。我有個朋友喜歡玩山地自行車,還給一些做自行車話題的自媒體投稿。久而久之,居然能夠寫一手好文章了,我相信他也能寫好技術文檔。

      哲學不是故作高深的學科,它的現實意義就是解決問題。年輕小伙是怎麼泡妞的?三天兩頭花不斷,大庭廣眾跪求愛。這類套路為什麼總是能成功呢?禮物滿足女人的物慾,當眾求愛滿足女人的虛榮心,投其所好。食堂大媽打菜的手越來越抖,辣子雞丁變成辣子辣丁,為什麼呢?食堂要控製成本,直接提價會惹眾怒。
      科學上的哲學,一般指研究事物發展的規律,歸納終極的解決方案。軟件行業充滿哲學味道的作品非常多,比如。舉個例子,當軟件系統遇到性能問題,嘗試下面兩種哲學思想提升性能:

    • 空間換時間:比如引入緩存,消耗額外的存儲提高響應速度。
    • 時間換空間:比如大文件的分片處理,分段處理后再匯總結果。

    設計穩健高可用的系統,嘗試從三個方面考慮問題:

    • 存儲:數據會丟失嗎,數據一致性怎麼解決。
    • 計算:計算怎麼擴容,應用允許任意增加節點嗎。
    • 傳輸:網絡中斷或擁塞怎麼辦。

    從無數的失敗或者成功的經驗中,總結出高度概括性的方案,讓我們下一步做的更好。

      英語是極為重要的基礎,學好英語與掌握編程語言一樣重要。且不說外企對英語的要求,許多知名博客就是把英文翻譯成中文,充當知識的搬運工。如果英語足夠好,直接閱讀一手英語資料,避免他人翻譯存在的謬誤。

      體系化的知識比零散的更容易記憶和理解,這正如一部好的電視劇,劇情環環相扣才能吸引觀眾。建議大家使用思維導圖羅列知識點,構建體繫結構,如下圖所示:

      高中是我們知識的巔峰時刻,每周小考每月大考,教輔資料堆成山,地獄式的反覆操練強化記憶。複習是對抗遺忘的唯一辦法。大腦的遺忘是有規律的,先快后慢。一天後,學到的知識只剩下原來的25%,甚至更低。隨着時間的推移,遺忘的速度減慢,遺忘的數量也就減少。

    時間間隔 記憶量
    剛看完 100%
    20分鐘后 60%
    1小時后 40%
    1天後 30%
    2天後 27%

    每個人的遺忘程度都不一樣,建議第二天複習前一天的內容,七天後複習這段時間的所有內容。

      不少朋友利用碎片時間學習,比如在公交上看公眾號的推送。其實我們都高估了自己的抗干擾能力,如果處在嘈雜的環境,注意力容易被打斷,記憶留存度也很低。碎片時間適合學習簡單孤立的知識點,比如鏈表的定義與實現。
      學習複雜的知識,需要大段的連續時間。圖書館是個好地方,安靜氛圍好。手機放一邊,不要理會QQ微信,最好閱讀紙質書,泡上一整天。有些城市出現了付費自習室,提供格子間、茶水等等,也是非常好的選擇。

      從下面這張圖我們可以看到,教授他人是知識留存率最高的方式。

      準備PPT和演講內容,給同事來一場技術分享。不光複習知識,還鍛煉口才。曾經有個同事說話又快又急,口頭禪也多,比如”對吧、是不是”,別人經常聽不清,但是他本人不以為然。領導讓他做了幾次技術分享,聽眾的反應可想而知,他才徹底認清缺點。
      堅持寫技術博客,別在意你寫的東西在網上已經重複千百遍。當自己動手的時候,才會意識到眼高手低。讓文章讀起來流暢清晰,需要嘔心瀝血的刪改。寫作是對大腦的長期考驗,想不到肯定寫不出,想不清楚肯定寫不清楚。

    我們經常說不要重複造輪子。為了開發效率,可以不造輪子,但是必須具備造輪子的能力。建議造一個簡單的MQ,你能用到通信協議、設計模式、隊列等許多知識。在造輪子的過程中,你會頻繁的翻閱各種手冊或者博客,這就是用輸出倒逼輸入

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

    ※評比前十大台北網頁設計台北網站設計公司知名案例作品心得分享

    ※智慧手機時代的來臨,RWD網頁設計已成為網頁設計推薦首選

    ※評比南投搬家公司費用收費行情懶人包大公開

  • 計算機硬件—調度與死鎖

    計算機硬件—調度與死鎖

    進程調度原因及調度切換時機,進程調度方式與實現及各種調度算法的個人總結:

    1.一般調度概念 1)什麼是調度 就是選出待分派的作業或進程。 操作系統管理了系統的有限資源,當有多個進程(或作業)發出請求要使用這些資源時,因為資源的有限性,必須按照一定的原則選擇進程(或作業)來佔用資源。這就是調度。

    2)調度目的 1.控制資源使用者的數量 2.選取資源使用者 3.許可哪些使用者佔用資源 4.讓使用者直接佔用資源。

    二、調度類型 1、高級調度(長程調度)。 即作業調度,選取輸入井中的作業,生成根進程。目的是控制使用系統資源的進程數。

    2、中級調度(中程調度)。 指選取進程佔用內存或有資格佔用內存,又稱進程滾入滾出,中級調度將在存儲管理章節中介紹,有頁式調度、段式調度、段頁式調度等。

    3、低級調度(短程調度) 指選取進程佔用處理機,又稱進程調度。 4、I/O調度 選取進程佔用I/O設備。

    三、調度和狀態轉換 在許多系統中,調度被分為三種:長程、中程和短程。 1)調度和進程狀態轉換

     

    從狀態轉換的觀點: 1、長程調度(作業調度)就是將一個或一批作業從後備狀態變為運行狀態。一個作業一旦被高級調度選中,便可獲得所需要的基本內存和設備資源,並被裝入內存,此後就以進程形式參与併發執行,與其它進程競爭CPU。

    從狀態轉換的觀點: 2、中程調度就是將進程從活動態變為靜止的掛起態,或者將進程從掛起態變為就緒或阻塞態。 3、短程調度就是將某個進程從就緒態變為(在CPU上運行的)執行態。

    2)調度的層次(調度作用的嵌套關係)

     

     

     

    3)三級調度示意圖 調度從根本上講,是要使隊列延遲的時間最小,並優化系統的執行效率

     

     

     

     

     

     

     

    4.1.2作業調度的功能

    ①記錄系統中各個作業的情況。 ②按照某種調度算法從後備作業隊列中挑選作業,即決定接納多少個作業進入內存和挑選哪些作業進入內存。

    ③為選中的作業分配內存和外設等資源。 ④為選中的作業建立相應的進程,並把該進程放入就緒隊列中。 ⑤作業結束後進行善後處理工作。如輸出必要的信息,收回該作業所佔用的全部資源,撤消與該作業相關的全部進程和該作業的JCB 。

     

    4.1.3進程調度的功能與調度時機 一、進程調度的功能 (1)保存現場(2)挑選進程(3)恢復現場

     

     

     

    二、進程調度的時機 一般在下列事件發生后要執行進程調度。 (1)創建進程。 當進程創建時,要決定是運行父進程還是子進程。 (2)進程終止。 (3)等待事件。 (4)中斷髮生。 (5)運行到時。

    4.1.4進程調度的基本方式 1)非剝奪調度(非搶佔) 只有當處理機上的進程主動放棄處理機時,才重新調度。 2)剝奪調度(搶佔) 當進程運行時可以被系統以某種原則為由剝奪其處理機。

    例:只有一部電話機,小李正在通電話,此時小張有急事也要打電話,此時小張的調度方式有兩種: 一種是非搶佔,即等待小李打完電話,自己再打電話。 另一種是搶佔,不等小李通完電話,搶過話筒就撥打電話。

    3)進程調度在核心態進行。 CPU狀態有兩種,目態和管態。 管態——也稱核心態或系統態,系統程序在CPU上運行的狀態。 目態——用戶程序在CPU上運行的狀態。

     

     

     

    4.2 調度算法 4.2.1 常用的調度算法

    1. FCFS 誰先到就緒隊列就將處理機分給誰。 2. 短作業優先 取一個下次所需運行時間最短的作業(該算法能使平均等待時間最短)。

    3. 優先級調度 選優先級最高的進程佔用處理機(優先級可動態改變)。 4.輪轉調度法 以先來後到的次序+ 時間片輪轉。

    5.多隊列調度法 按屬性將就緒進程分類,不同類進程可有不同的調度算法。 6.多級反饋隊列調度法 設置多條就緒隊列,進程被調度執行后,在被剝奪或放棄處理機后而在就緒時,可以改變其就緒隊列(見下圖)。

     

     

    又一個多級反饋隊列調度算法的例子:使用優先權實現調度。做法如下:

    (1)以優先級設置多隊列。 (2)各隊列的調度算法採用FCFS+時間片輪轉. (3)進程優先級升降原則是:等待過久升,輸入/輸出完成時升,運行完一個完整時間片降……

    (4)進程最初進入就緒隊列以用戶初置優先級為參數。 (5)開始調度時,先從最高優先權的就緒隊列(RQ0)開始選取一個進程,如果RQ0空,則檢查RQ1,如此下去。見下圖示:

     

     

    4.2.2 調度策略的討論

    一、調度性能評價準則 1、CPU利用率。 2、吞吐率。即每單位時間所完成的作業數目。 3、周轉時間。即從作業提交直至完成這一作業的時間間隔。

    4、等待時間。即作業在就緒隊列中等待所花的時間。 5、響應時間。即從作業提交到首次產生回答信息之間的時間。

    二、主要的調度算法 下面以表所示的進程集作為範例討論不同的調度策略。 到達時間:進程或作業提交時間。 服務時間:進程要求服務的時間,或完成作業所需的時間。

     

    一、FCFS調度策略(或先進先出)

     

     

     

     

     

     

     

    FCFS策略更適合於長進程。例:

     

     

    注:周轉時間=完成時間—到達時間 tq/ts :帶權周轉時間=周轉時間/服務時間

    (1)進程C的等待標準化時間是不能容忍的。它位於系統的總時間是它所需服務時間的100倍。無論何時,一個長進程到后,一個短進程就會出現較長的等待。 (2)進程D的標準化等待時間尚可容忍,小於2.0。進程D的輪轉時間差不多是C的兩倍。

    總結: FCFS(First Come First Served)即先來先服務,故它的本質是非搶佔的。它簡單易行,但調度性能較差,有可能使短的、重要的或緊迫的作業等進程長期等待,其實現過程容易,可採用FIFO隊列管理。

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

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

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

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

    南投搬家費用,距離,噸數怎麼算?達人教你簡易估價知識!

  • Java性能分析神器–VisualVM Launcher[1]

    Java性能分析神器–VisualVM Launcher[1]

    Java性能分析神器1–VisualVM Launcher

    VisualVM

    當你日復一日敲代碼的時候,當你把各種各樣的框架集成到一起的時候,看着大功告成成功運行的日誌,有沒有那麼一絲絲迷茫和惆悵:這TM起的是什麼玩意?每一行日誌背後代表的是什麼東西??他為什麼就能跑起來了呢????

    這種時候不要慌,給大家推薦一款功能強大的插件:VisualVM Launcher。(eclipse就叫 )。這個插件需要和客戶端配合使用 。

    VisualVM是集成了命令行JDK工具和輕量級分析功能的可視化工具。JVM提供了一些常用的jdk命令行工具:

    • jstat(JVM Statistics Monitoring Tool):用於收集Hotspot虛擬機各方面的運行數據(查看虛擬機各雲心狀態信息),显示本地或遠程虛擬機進程中的類裝載,內存,垃圾收集, JIT編譯等運行數據。
    • jps(JVM Process Status Tool):显示指定系統內所有的HotSpot虛擬機進程(查看虛擬機進程信息),可用於查詢正在運行的虛擬機進程, 同時可選擇性的显示虛擬機執行主類, 即執行main函數的類, 以及進程的本地虛擬機
      ID(Local Virtual Machine Identifier 簡稱LVMID)(對於本地虛擬機進程來說, 進程的本地虛擬機ID與操作系統的進程ID是一致的)
    • jinfo(Configuration Info for Java):显示虛擬機配置信息(查看虛擬機配置參數信息),可用於查看和調整虛擬機的配置參數.
    • jmap(JVM Memory Map):生成虛擬機的內存轉儲快照, 生成heapdump文件(生成虛擬機內存轉儲快照),可用於獲取heapdump文件, 且可以查詢finalize執行隊列, Java堆與永久代的一些信息。
    • jhat(JVM Heap Dump Browser):用於分析heapdump文件, 它會建立一個HTTP/HTML服務器, 讓用戶在瀏覽器上查看分析結果(分析虛擬機轉儲快照信息),jhat命令與jmap命令搭配使用, 用於分析jmap生成的堆轉儲快照, jhat內置了一個微型的HTTP/HTML服務器, 生成dump文件的分析結果后, 可以在瀏覽器中查看。
    • jstack(JVM Stack Trace):显示虛擬機的線程快照(虛擬機堆棧跟蹤),用於生成虛擬機當前時刻的線程快照。 線程快照指的是當前虛擬機內的每一條線程正在執行的方法堆棧的集合, 生成線程快照的作用是, 可用於定位線程出現長時間停頓的原因, 如線程間死鎖, 死循環, 請求外部資源導致的長時間等待等問題, 當線程出現停頓時 就可以用jstack各個線程調用的堆棧情況

    這些工具功能強大,可以很方便的查看jvm內存分配,內存大小,裝載類總數,線程總數等。有了這些信息,就可以很快的進程診斷,性能調優辣。

    安裝VisualVM和VisualVM Launcher

    1. Idea安裝VisualVM Launcher插件

    ​ Preferences –> Plugins –> 搜索VisualVM Launcher,安裝重啟即可

    2. 配置Idea VisualVM Launcher插件

    ​ Preferences –> other settings -> VisualVM Launcher –> 輸入VisualVM executable 和 JDK home即可

    3. 配置完之後的idea頁面

    4. 安裝VisualVM客戶端

    ​ –> 選擇對應的系統安裝包 –> 對應安裝,安裝完成后打開是這樣的頁面:

    VisualVM和java命令行工具

    1. jmap+jhat內存快照與分析:Heap Dump
    1. HeapDump又叫做堆存儲文件,指一個Java進程在某個時間點的內存快照。Heap Dump在觸發內存快照的時候會保存此刻的java對象和類的信息。通常在寫heap Dump文件前會觸發一次FullGC,所以heap dump文件里保存的都是FullCG后留下的對象信息。

    2. jmap進行內存快照方式:

      jmap -dump:format=b,file=<filename.hprof> <pid>

    3. jhat進行內存快照分析:

      • jhat <heap dump file>
      • 使用了jhat命令,就啟動了一個http服務,端口是7000,即http://localhost:7000/,就可以在瀏覽器里分析
    4. VisualVM進行內存快照方式:

      • 在“應用程序”窗口中右鍵單擊應用程序節點,然後選擇“堆 Dump”。
      • 在“應用程序”窗口中雙擊應用程序節點以打開應用程序標籤,然後在“監視”標籤中單擊“堆 Dump”。
    5. VisualVM快照頁面,也可以右鍵保存此時的快照:

    6. 想要打開保存好的java快照:

      • 單擊“堆 Dump”工具欄中的“類”,以查看活動類和對應實例的列表。
      • 雙擊某個類名打開“實例”視圖,以查看實例列表。
      • 從列表中選擇某個實例,以查看對該實例的引用。
    2. jinfo:显示虛擬機配置信息(查看虛擬機配置參數信息)
    1. 虛擬機配置信息:JVM的啟動參數

    2. jinfo進行查看虛擬機配置信息查詢(jinfo -help查看更多)

      jinfo <pid>

    3. Visual VM查看虛擬機配置信息,直接在應用程序打開,就可以看到JVM參數 和 系統屬性:

    4. 一些常見的虛擬機配置參數:

      • -Xms:初始堆大小。如:-Xms256m
      • -Xmx:最大堆大小。如:-Xmx512m
      • -Xmn:新生代大小。通常為 Xmx 的 1/3 或 1/4。
      • -Xss:為每個線程分配的內存大小,JDK1.5+ 每個線程堆棧大小為 1M,一般來說如果棧不是很深的話, 1M 是絕對夠用了的。
      • -XX:NewRatio:新生代與老年代的比例,如 –XX:NewRatio=2,則新生代占整個堆空間的1/3,老年代佔2/3
      • -XX:SurvivorRatio:新生代中 Eden 與 Survivor 的比值。默認值為 8。即 Eden 佔新生代空間的 8/10,另外兩個 Survivor 各占 1/10
      • -XX:PermSize:永久代(方法區)的初始大小
        • PermSize永久代的概念在jdk1.8中已經不存在了,取而代之的是metaspace元空間,當認為執行永久代的初始大小以及最大值是jvm會給出如此下提示:
          • Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=30m; support was removed in 8.0
          • Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=30m; support was removed in 8.0
      • -XX:MaxPermSize:永久代(方法區)的最大值
      • -XX:+PrintGCDetails:打印 GC 信息
      • -XX:+HeapDumpOnOutOfMemoryError:讓虛擬機在發生內存溢出時 Dump 出當前的內存堆轉儲快照,以便分析用
    3. jps查看虛擬機進程信息
    1. 用來查詢正在運行的虛擬機進程

    2. jps命令,:

      • jps
    3. VisualVM查看正在運行的虛擬機進程:

    4. jstack显示虛擬機的線程快照
    1. 生成虛擬機當前時刻的線程快照,用來查找運行時死鎖,死循環的原因

    2. jstack命令,

      • jstack <pid>
    3. VisualVM生成虛擬機線程快照方式:

      • 在“應用程序”窗口中右鍵單擊應用程序節點,然後選擇“線程 Dump”。
      • 在“應用程序”窗口中雙擊應用程序節點以打開應用程序標籤,然後在“線程”標籤中單擊“線程 Dump”。
    4. VisualVM線程快照頁面,也可以右鍵保存快照:

    5. jstat收集Hotspot虛擬機各方面的運行數據
    1. 運行數據:對Java應用程序的資源和性能進行實時監控,主要包括GC情況和Heap Size資源使用情況。

    2. jstat進行資源與性能監控,:

      • jstat -gc <pid>
    3. VisualVM進行程序資源的實時監控:

    VisualVM也提供了一些其他功能

    此外,VisualVM也提供很多插件,有各樣的功能,我就不多介紹了

    這篇文章,介紹了VisualVM的作用和用法,下面會寫一篇姊妹篇 帶上代碼,去分析當系統出現死鎖或者循環等異常時,內存、線程和CPU在做什麼。

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    ※想知道網站建置網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計後台網頁設計

    ※不管是台北網頁設計公司台中網頁設計公司,全省皆有專員為您服務

    ※Google地圖已可更新顯示潭子電動車充電站設置地點!!

    ※帶您來看台北網站建置台北網頁設計,各種案例分享

  • 深入理解@LoadBalanced註解的實現原理與客戶端負載均衡

    前提

    在閱讀這篇博客之前,希望你對SpringCloud套件熟悉和理解,更希望關注下

    概述

    在使用springcloud ribbon客戶端負載均衡的時候,可以給RestTemplate bean 加一個@LoadBalanced註解,就能讓這個RestTemplate在請求時擁有客戶端負載均衡的能力,先前有細嚼過但是沒有做過筆記,剛好處理此類問題記錄下

    @LoadBalanced

    /**
     * 註釋將RestTemplate bean標記為配置為使用LoadBalancerClient。
     */
    @Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD })
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Qualifier
    public @interface LoadBalanced {
    }

    通過源碼可以發現這是一個LoadBalanced標記註解並且標記了@Qualifier(基於Spring Boot的自動配置機制),我們可以溯源到LoadBalancerAutoConfiguration

    LoadBalancerAutoConfiguration

    /**
     * 功能區的自動配置(客戶端負載平衡)
     */
    @Configuration
    @ConditionalOnClass(RestTemplate.class)
    @ConditionalOnBean(LoadBalancerClient.class)
    @EnableConfigurationProperties(LoadBalancerRetryProperties.class)
    public class LoadBalancerAutoConfiguration {
    
        @LoadBalanced
        @Autowired(required = false)
        private List<RestTemplate> restTemplates = Collections.emptyList();   //這裏持有@LoadBalanced標記的RestTemplate實例
    
        @Autowired(required = false)
        private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();
    
        @Bean
        public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(
                final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
            return () -> restTemplateCustomizers.ifAvailable(customizers -> {
                for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
                    for (RestTemplateCustomizer customizer : customizers) {
              //為restTemplate添加定製
                        customizer.customize(restTemplate);
                    }
                }
            });
        }
    
       // ... 
    
        /**
         * 以下針對classpath存在RetryTemplate.class的情況配置,先忽略
         */
        @Configuration
        @ConditionalOnClass(RetryTemplate.class)
        public static class RetryAutoConfiguration {
    
            @Bean
            @ConditionalOnMissingBean
            public LoadBalancedRetryFactory loadBalancedRetryFactory() {
                return new LoadBalancedRetryFactory() {
                };
            }
        }
    
      // ... 
    }

    @LoadBalanced@Autowried結合使用,意思就是這裏注入的RestTempate Bean是所有加有@LoadBalanced註解標記的(持有@LoadBalanced標記的RestTemplate實例)

    這段自動裝配的代碼的含義不難理解,就是利用了RestTempllate的攔截器,使用RestTemplateCustomizer對所有標註了@LoadBalanced的RestTemplate Bean添加了一個LoadBalancerInterceptor攔截器,而這個攔截器的作用就是對請求的URI進行轉換獲取到具體應該請求哪個服務實例ServiceInstance。

    關鍵問下自己:為什麼?

    • RestTemplate實例是怎麼被收集的?
    • 怎樣通過負載均衡規則獲取具體的具體的server?

    繼續扒看源碼>
    上面可以看出,會LoadBalancerAutoConfiguration類對我們加上@LoadBalanced註解的bean 添加loadBalancerInterceptor攔截器

    LoadBalancerInterceptor

    /**
    * 功能區的自動配置(客戶端負載平衡)。
    */
    public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
    
        private LoadBalancerClient loadBalancer;
    
        private LoadBalancerRequestFactory requestFactory;
    
        public LoadBalancerInterceptor(LoadBalancerClient loadBalancer,
                LoadBalancerRequestFactory requestFactory) {
            this.loadBalancer = loadBalancer;
            this.requestFactory = requestFactory;
        }
    
        public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
            // for backwards compatibility
            this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
        }
    
        @Override
        public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
                final ClientHttpRequestExecution execution) throws IOException {
            final URI originalUri = request.getURI();
            String serviceName = originalUri.getHost();
            Assert.state(serviceName != null,
                    "Request URI does not contain a valid hostname: " + originalUri);
            return this.loadBalancer.execute(serviceName,
                    this.requestFactory.createRequest(request, body, execution));
        }
    
    }

    重點看intercept方法 當我們restTemplate執行請求操作時,就會被攔截器攔截進入intercept方法,而loadBalancer是LoadBalancerClient的具體實現

    RibbonLoadBalancerClient

        public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
                throws IOException {
            ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
            Server server = getServer(loadBalancer, hint);
            if (server == null) {
                throw new IllegalStateException("No instances available for " + serviceId);
            }
            RibbonServer ribbonServer = new RibbonServer(serviceId, server,
                    isSecure(server, serviceId),
                    serverIntrospector(serviceId).getMetadata(server));
    
            return execute(serviceId, ribbonServer, request);
        }

    看到這裏相信都遇到過類似的錯誤,恍然大悟

    No instances available for  xxxxx

    總結

    • 1.根據serviceId 獲取對應的loadBalancer
    • 2.根據loadBalancer獲取具體的server(這裏根據負載均衡規則,獲取到具體的服務實例)
    • 3.創建RibbonServer
    • 4.執行具體請求

    這裏

    注意: @LoadBalanced 標記註解獲取到最後通過負載均衡規則獲取具體的具體的server來發起請求

    案例

    /**
     * 服務註冊中心配置
     *
     * @author <a href="mailto:shangzhi.ibyte@gmail.com">iByte</a>
     * @since 1.0.1
     */
    @Configuration
    @EnableConfigurationProperties(ModuleMappingHelper.class)
    public class DiscoveryConfig {
        @Autowired
        Environment environment;
    
        /**
         * DiscoveryHeaderHelper默認bean
         * @return
         */
        @Bean
        public DiscoveryHeaderHelper discoveryHeaderHelper() {
            DiscoveryHeaderHelper discoveryHeaderHelper = new DiscoveryHeaderHelper(environment);
            DiscoveryHeaderHelper.INSTANCE = discoveryHeaderHelper;
            return discoveryHeaderHelper;
        }
    
        /**
         * resttemplate構建
         */
        @Resource
        private RestTemplateBuilder restTemplateBuilder;
    
        /**
         * resttemplate請求bean,更改系統本身的builder
         * @return
         */
        @Bean
        @LoadBalanced
        public RestTemplate restTemplate() {
            RestTemplate restTemplate = restTemplateBuilder.configure(new RestTemplate());
            //RestTemplate interceptors 遠程調用請求增加頭部信息處理
            restTemplate.getInterceptors().add(new RestApiHeaderInterceptor());
            //RestTemplate Set the error handler 錯誤處理
            restTemplate.setErrorHandler(new RestResponseErrorHandler());
            return  restTemplate;
        }
    
        @Bean
        public DiscoveryClient.DiscoveryClientOptionalArgs discoveryClientOptionalArgs() {
            DiscoveryClient.DiscoveryClientOptionalArgs discoveryClientOptionalArgs = new DiscoveryClient.DiscoveryClientOptionalArgs();
            discoveryClientOptionalArgs.setAdditionalFilters(Collections.singletonList(new DiscoveryHeaderClientFilter()));
            discoveryClientOptionalArgs.setEventListeners(Collections.singleton(new EurekaClientEventListener()));
            return discoveryClientOptionalArgs;
        }
    }

    源碼地址 >

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

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

    ※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

    ※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

    南投搬家前需注意的眉眉角角,別等搬了再說!

  • OTA升級詳解(三)

    OTA升級詳解(三)

    君子知夫不全不粹之不足以為美也, 

    故誦數以貫之,

    思索以通之,

    為其人以處之,

    除其害者以持養之;

              出自荀子《勸學篇》

    終於OTA的升級過程的詳解來了,之前的兩篇文章與主要是鋪墊,

    OTA升級的一些基礎知識,那這邊文章就開始揭開OTA-recovery模式升級過程的神秘面紗,需要說明的是

    以下重點梳理了本人認為的關鍵、核心的流程,其他如ui部分、簽名校驗部分我並未花筆墨去描述,主要

    還是講升級的核心,其他都是枝枝恭弘=叶 恭弘恭弘=叶 恭弘。Android 10 recovery源碼分析,代碼來源路徑:

    https://www.androidos.net.cn/android/10.0.0_r6/xref

    本文所講的流程代碼路徑為:bootable/recovery/

    首先從文件層面說下升級功能的調用流程,說明如下:

    recovery-main.cpp      升級的主入口

    recovery.cpp                開始recovery升級的處理流程

    install/install.cpp         執行升級的處理流程(調用updater)

    updater/updater.cpp  完成升級的核心流程

     

     

     

    1 主入口代碼為:recovery-main.cpp,main入口

    1.1 日誌相關的工作準備

     

     1 // We don't have logcat yet under recovery; so we'll print error on screen and log to stdout
     2 // (which is redirected to recovery.log) as we used to do.
     3 android::base::InitLogging(argv, &UiLogger);
     4 
     5 // Take last pmsg contents and rewrite it to the current pmsg session.
     6 static constexpr const char filter[] = "recovery/";
     7 // Do we need to rotate?
     8 bool do_rotate = false;
     9 
    10 __android_log_pmsg_file_read(LOG_ID_SYSTEM, ANDROID_LOG_INFO, filter, logbasename, &do_rotate);
    11 // Take action to refresh pmsg contents
    12 __android_log_pmsg_file_read(LOG_ID_SYSTEM, ANDROID_LOG_INFO, filter, logrotate, &do_rotate);
    13 
    14 time_t start = time(nullptr);
    15 
    16 // redirect_stdio should be called only in non-sideload mode. Otherwise we may have two logger
    17 // instances with different timestamps.
    18 redirect_stdio(Paths::Get().temporary_log_file().c_str());

    1.2 load_volume_table(); 加載系統分區信息,注意這裏並明白掛載分區

    .mount_point = “/tmp”, .fs_type = “ramdisk”, .blk_device = “ramdisk”, .length = 0 

    mount_point — 掛載點    fs_type — 分區類型 

    blk_device     — 設備塊名 length  — 分區大小

    1.3 掛載/cache分區,我們的升級命令都放在這個分區下

    1 has_cache = volume_for_mount_point(CACHE_ROOT) != nullptr;

    1.4 獲取升級的參數並寫BCB塊信息

    std::vector<std::string> args = get_args(argc, argv);
    
    if (!update_bootloader_message(options, &err)) {
        LOG(ERROR) << "Failed to set BCB message: " << err;
    }

    a、讀取misc分區分區,並將recovery模式升級的標記寫到misc分區中,這樣做的目的是斷電續升,

    升級中掉電之後,如果下次開機重啟,在bootloader中會讀取此標記,並重新進入到recovery模式中

    update_bootloader_message函數完成此功能。

    b、從/cache/recovery/command 中讀取升級參數,這裏recovery啟動進程是未帶入參數時,command

    文件的接口其實有很詳細的解釋

     * The arguments which may be supplied in the recovery.command file:
     *   --update_package=path - verify install an OTA package file
     *   --wipe_data - erase user data (and cache), then reboot
     *   --prompt_and_wipe_data - prompt the user that data is corrupt, with their consent erase user
     *       data (and cache), then reboot
     *   --wipe_cache - wipe cache (but not user data), then reboot
     *   --show_text - show the recovery text menu, used by some bootloader (e.g. http://b/36872519).
     *   --set_encrypted_filesystem=on|off - enables / diasables encrypted fs
     *   --just_exit - do nothing; exit and reboot

    1.5 加載recovery_ui_ext.so,完成升級中與屏幕信息的显示,升級進度,升級結果等。這裏就不多說了。

    static constexpr const char* kDefaultLibRecoveryUIExt = "librecovery_ui_ext.so";
      // Intentionally not calling dlclose(3) to avoid potential gotchas (e.g. `make_device` may have
      // handed out pointers to code or static [or thread-local] data and doesn't collect them all back
      // in on dlclose).
      void* librecovery_ui_ext = dlopen(kDefaultLibRecoveryUIExt, RTLD_NOW);
    
      using MakeDeviceType = decltype(&make_device);
      MakeDeviceType make_device_func = nullptr;
      if (librecovery_ui_ext == nullptr) {
        printf("Failed to dlopen %s: %s\n", kDefaultLibRecoveryUIExt, dlerror());
      } else {
        reinterpret_cast<void*&>(make_device_func) = dlsym(librecovery_ui_ext, "make_device");
        if (make_device_func == nullptr) {
          printf("Failed to dlsym make_device: %s\n", dlerror());
        }
      }

    1.6 非fastboot模式升級就開始了recovery模式升級,start_recovery

    ret = fastboot ? StartFastboot(device, args) : start_recovery(device, args);

    2 進入 recovery.cpp 

    2.1 參數解析,這些參數其實就是來源於/cache/recovery/command, 上面已經通過get_arg,

    讀取到了args中

    2.2 界面的各種ui信息显示,點事電量的檢查等待輔助動作。

    2.3 函數名為安裝升級包,其實還未真正開始進行升級包的安裝

    1 status = install_package(update_package, should_wipe_cache, true, retry_count, ui);

    2.4 安裝結束之後由finish_recovery()完成收尾工作,保存日誌、清除BCB中的標記,設備重啟。

     1 static void finish_recovery() {
     2   std::string locale = ui->GetLocale();
     3   // Save the locale to cache, so if recovery is next started up without a '--locale' argument
     4   // (e.g., directly from the bootloader) it will use the last-known locale.
     5   if (!locale.empty() && has_cache) {
     6     LOG(INFO) << "Saving locale \"" << locale << "\"";
     7     if (ensure_path_mounted(LOCALE_FILE) != 0) {
     8       LOG(ERROR) << "Failed to mount " << LOCALE_FILE;
     9     } else if (!android::base::WriteStringToFile(locale, LOCALE_FILE)) {
    10       PLOG(ERROR) << "Failed to save locale to " << LOCALE_FILE;
    11     }
    12   }
    13 
    14   copy_logs(save_current_log, has_cache, sehandle);
    15 
    16   // Reset to normal system boot so recovery won't cycle indefinitely.
    17   std::string err;
    18   if (!clear_bootloader_message(&err)) {
    19     LOG(ERROR) << "Failed to clear BCB message: " << err;
    20   }
    21 
    22   // Remove the command file, so recovery won't repeat indefinitely.
    23   if (has_cache) {
    24     if (ensure_path_mounted(COMMAND_FILE) != 0 || (unlink(COMMAND_FILE) && errno != ENOENT)) {
    25       LOG(WARNING) << "Can't unlink " << COMMAND_FILE;
    26     }
    27     ensure_path_unmounted(CACHE_ROOT);
    28   }
    29 
    30   sync();  // For good measure.
    31 }

    3 install/install.cpp

    3.1 install.cpp其實就進入了安裝升級包的準備動作,剛上的install_package,是假的,這裏才是

    really_install_package

    1 really_install_package(path, &updater_wipe_cache, needs_mount, &log_buffer,
    2                                     retry_count, &max_temperature, ui);

    3.2 really_install_package 關鍵地方已加註釋

     1 static int really_install_package(const std::string& path, bool* wipe_cache, bool needs_mount,
     2                                   std::vector<std::string>* log_buffer, int retry_count,
     3                                   int* max_temperature, RecoveryUI* ui) {
     4   ui->SetBackground(RecoveryUI::INSTALLING_UPDATE);
     5   ui->Print("Finding update package...\n");
     6   // Give verification half the progress bar...
     7   ui->SetProgressType(RecoveryUI::DETERMINATE);
     8   ui->ShowProgress(VERIFICATION_PROGRESS_FRACTION, VERIFICATION_PROGRESS_TIME);
     9   LOG(INFO) << "Update location: " << path;
    10 
    11   // Map the update package into memory.
    12   ui->Print("Opening update package...\n");
    13 
    14   if (needs_mount) {
    15     if (path[0] == '@') {
    16       ensure_path_mounted(path.substr(1));
    17     } else {
    18       ensure_path_mounted(path);
    19     }
    20   }
    21 
    22   /* 將zip映射到內存中 */
    23   auto package = Package::CreateMemoryPackage(
    24       path, std::bind(&RecoveryUI::SetProgress, ui, std::placeholders::_1));
    25   if (!package) {
    26     log_buffer->push_back(android::base::StringPrintf("error: %d", kMapFileFailure));
    27     return INSTALL_CORRUPT;
    28   }
    29 
    30   // Verify package.進行zip包進行簽名校驗
    31   if (!verify_package(package.get(), ui)) {
    32     log_buffer->push_back(android::base::StringPrintf("error: %d", kZipVerificationFailure));
    33     return INSTALL_CORRUPT;
    34   }
    35 
    36   // Try to open the package.打開zip包
    37   ZipArchiveHandle zip = package->GetZipArchiveHandle();
    38   if (!zip) {
    39     log_buffer->push_back(android::base::StringPrintf("error: %d", kZipOpenFailure));
    40     return INSTALL_CORRUPT;
    41   }
    42 
    43   // Additionally verify the compatibility of the package if it's a fresh install.
    44   if (retry_count == 0 && !verify_package_compatibility(zip)) {
    45     log_buffer->push_back(android::base::StringPrintf("error: %d", kPackageCompatibilityFailure));
    46     return INSTALL_CORRUPT;
    47   }
    48 
    49   // Verify and install the contents of the package.
    50   ui->Print("Installing update...\n");
    51   if (retry_count > 0) {
    52     ui->Print("Retry attempt: %d\n", retry_count);
    53   }
    54   ui->SetEnableReboot(false);
    55   int result =
    56       /* 執行升級updater進程進行升級 */
    57       try_update_binary(path, zip, wipe_cache, log_buffer, retry_count, max_temperature, ui);
    58   ui->SetEnableReboot(true);
    59   ui->Print("\n");
    60 
    61   return result;
    62 }

    3.3 try_update_binary

    從升級包中讀取元數據信息

    1 ReadMetadataFromPackage(zip, &metadata)

    3.4 從升級包中讀取updater進程

     1 int SetUpNonAbUpdateCommands(const std::string& package, ZipArchiveHandle zip, int retry_count,
     2                              int status_fd, std::vector<std::string>* cmd) {
     3   CHECK(cmd != nullptr);
     4 
     5   // In non-A/B updates we extract the update binary from the package.
     6   static constexpr const char* UPDATE_BINARY_NAME = "META-INF/com/google/android/update-binary";
     7   ZipString binary_name(UPDATE_BINARY_NAME);
     8   ZipEntry binary_entry;
     9   if (FindEntry(zip, binary_name, &binary_entry) != 0) {
    10     LOG(ERROR) << "Failed to find update binary " << UPDATE_BINARY_NAME;
    11     return INSTALL_CORRUPT;
    12   }
    13 
    14   const std::string binary_path = Paths::Get().temporary_update_binary();
    15   unlink(binary_path.c_str());
    16   android::base::unique_fd fd(
    17       open(binary_path.c_str(), O_CREAT | O_WRONLY | O_TRUNC | O_CLOEXEC, 0755));
    18   if (fd == -1) {
    19     PLOG(ERROR) << "Failed to create " << binary_path;
    20     return INSTALL_ERROR;
    21   }
    22 
    23   int32_t error = ExtractEntryToFile(zip, &binary_entry, fd);
    24   if (error != 0) {
    25     LOG(ERROR) << "Failed to extract " << UPDATE_BINARY_NAME << ": " << ErrorCodeString(error);
    26     return INSTALL_ERROR;
    27   }
    28 
    29   // When executing the update binary contained in the package, the arguments passed are:
    30   //   - the version number for this interface
    31   //   - an FD to which the program can write in order to update the progress bar.
    32   //   - the name of the package zip file.
    33   //   - an optional argument "retry" if this update is a retry of a failed update attempt.
    34   *cmd = {
    35     binary_path,
    36     std::to_string(kRecoveryApiVersion),
    37     std::to_string(status_fd),
    38     package,
    39   };
    40   if (retry_count > 0) {
    41     cmd->push_back("retry");
    42   }
    43   return 0;
    44 }

    3.5 創建管道,這裏子進程關閉了讀端,父進程關閉了寫端,這樣就是保證從單向的信息通信,從

    子進程傳入信息到父進程中。

    1 android::base::Pipe(&pipe_read, &pipe_write, 0)

    3.6 創建子進程,在子進程中運行update-binary進程

     1 if (pid == 0) {
     2     umask(022);
     3     pipe_read.reset();
     4 
     5     // Convert the std::string vector to a NULL-terminated char* vector suitable for execv.
     6     auto chr_args = StringVectorToNullTerminatedArray(args);
     7     /* chr_args[0] 其實就是升級包中的 META-INF/com/google/android/update-binary */
     8     execv(chr_args[0], chr_args.data());
     9     // We shouldn't use LOG/PLOG in the forked process, since they may cause the child process to
    10     // hang. This deadlock results from an improperly copied mutex in the ui functions.
    11     // (Bug: 34769056)
    12     fprintf(stdout, "E:Can't run %s (%s)\n", chr_args[0], strerror(errno));
    13     _exit(EXIT_FAILURE);
    14   }

    3.7 recovery獲取子進程的信息並显示,進度、ui_print 等等。

    1 FILE* from_child = android::base::Fdopen(std::move(pipe_read), "r");
    2 while (fgets(buffer, sizeof(buffer), from_child) != nullptr)

    4 execv執行升級進程之後,工作在updater/updater.cpp中完成。

    4.1 這裏的主要核心就是構造腳本解析器對updater-script中的命令進行執行,至於這個腳本解析器

    是如何構造的,如何執行的, 其實我也搞的不是很清楚。

    4.2 安裝升級包的核心程序就是Configure edify’s functions. 中的那些註冊回調函數

      1 int main(int argc, char** argv) {
      2 // Various things log information to stdout or stderr more or less
      3 // at random (though we've tried to standardize on stdout).  The
      4 // log file makes more sense if buffering is turned off so things
      5 // appear in the right order.
      6   setbuf(stdout, nullptr);
      7   setbuf(stderr, nullptr);
      8 // We don't have logcat yet under recovery. Update logs will always be written to stdout
      9 // (which is redirected to recovery.log).
     10   android::base::InitLogging(argv, &UpdaterLogger);
     11 if (argc != 4 && argc != 5) {
     12     LOG(ERROR) << "unexpected number of arguments: " << argc;
     13 return 1;
     14   }
     15 /* 支持的版本檢查 */
     16 char* version = argv[1];
     17 if ((version[0] != '1' && version[0] != '2' && version[0] != '3') || version[1] != '\0') {
     18 // We support version 1, 2, or 3.
     19     LOG(ERROR) << "wrong updater binary API; expected 1, 2, or 3; got " << argv[1];
     20 return 2;
     21   }
     22 // Set up the pipe for sending commands back to the parent process.
     23 int fd = atoi(argv[2]);
     24   FILE* cmd_pipe = fdopen(fd, "wb");
     25   setlinebuf(cmd_pipe);
     26 // Extract the script from the package.
     27 /* 從包中提取腳本 */
     28 const char* package_filename = argv[3];
     29   MemMapping map;
     30 if (!map.MapFile(package_filename)) {
     31     LOG(ERROR) << "failed to map package " << argv[3];
     32 return 3;
     33   }
     34   ZipArchiveHandle za;
     35 int open_err = OpenArchiveFromMemory(map.addr, map.length, argv[3], &za);
     36 if (open_err != 0) {
     37     LOG(ERROR) << "failed to open package " << argv[3] << ": " << ErrorCodeString(open_err);
     38     CloseArchive(za);
     39 return 3;
     40   }
     41 ZipString script_name(SCRIPT_NAME);
     42   ZipEntry script_entry;
     43 int find_err = FindEntry(za, script_name, &script_entry);
     44 if (find_err != 0) {
     45     LOG(ERROR) << "failed to find " << SCRIPT_NAME << " in " << package_filename << ": "
     46                << ErrorCodeString(find_err);
     47     CloseArchive(za);
     48 return 4;
     49   }
     50 std::string script;
     51   script.resize(script_entry.uncompressed_length);
     52 int extract_err = ExtractToMemory(za, &script_entry, reinterpret_cast<uint8_t*>(&script[0]),
     53                                     script_entry.uncompressed_length);
     54 if (extract_err != 0) {
     55     LOG(ERROR) << "failed to read script from package: " << ErrorCodeString(extract_err);
     56     CloseArchive(za);
     57 return 5;
     58   }
     59 // Configure edify's functions.
     60 /* 註冊updater-script中的回調函數 這裏主要是一些斷言函數 abort assert*/
     61   RegisterBuiltins();
     62 /* 這裏主要是一些安裝升級包的函數 主要是對有文件系統的分區來說*/
     63   RegisterInstallFunctions();
     64 /* 這裏主要註冊對裸分區進行升級的函數 */
     65   RegisterBlockImageFunctions();
     66   RegisterDynamicPartitionsFunctions();
     67   RegisterDeviceExtensions();
     68 // Parse the script.
     69 std::unique_ptr<Expr> root;
     70 int error_count = 0;
     71 int error = ParseString(script, &root, &error_count);
     72 if (error != 0 || error_count > 0) {
     73     LOG(ERROR) << error_count << " parse errors";
     74     CloseArchive(za);
     75 return 6;
     76   }
     77   sehandle = selinux_android_file_context_handle();
     78   selinux_android_set_sehandle(sehandle);
     79 if (!sehandle) {
     80 fprintf(cmd_pipe, "ui_print Warning: No file_contexts\n");
     81   }
     82 // Evaluate the parsed script.
     83   UpdaterInfo updater_info;
     84   updater_info.cmd_pipe = cmd_pipe;
     85   updater_info.package_zip = za;
     86   updater_info.version = atoi(version);
     87   updater_info.package_zip_addr = map.addr;
     88   updater_info.package_zip_len = map.length;
     89 State state(script, &updater_info);
     90 if (argc == 5) {
     91 if (strcmp(argv[4], "retry") == 0) {
     92       state.is_retry = true;
     93     } else {
     94 printf("unexpected argument: %s", argv[4]);
     95     }
     96   }
     97 std::string result;
     98 bool status = Evaluate(&state, root, &result);
     99 if (!status) {
    100 if (state.errmsg.empty()) {
    101       LOG(ERROR) << "script aborted (no error message)";
    102 fprintf(cmd_pipe, "ui_print script aborted (no error message)\n");
    103     } else {
    104       LOG(ERROR) << "script aborted: " << state.errmsg;
    105 const std::vector<std::string> lines = android::base::Split(state.errmsg, "\n");
    106 for (const std::string& line : lines) {
    107 // Parse the error code in abort message.
    108 // Example: "E30: This package is for bullhead devices."
    109 if (!line.empty() && line[0] == 'E') {
    110 if (sscanf(line.c_str(), "E%d: ", &state.error_code) != 1) {
    111             LOG(ERROR) << "Failed to parse error code: [" << line << "]";
    112           }
    113         }
    114 fprintf(cmd_pipe, "ui_print %s\n", line.c_str());
    115       }
    116     }
    117 // Installation has been aborted. Set the error code to kScriptExecutionFailure unless
    118 // a more specific code has been set in errmsg.
    119 if (state.error_code == kNoError) {
    120       state.error_code = kScriptExecutionFailure;
    121     }
    122 fprintf(cmd_pipe, "log error: %d\n", state.error_code);
    123 // Cause code should provide additional information about the abort.
    124 if (state.cause_code != kNoCause) {
    125 fprintf(cmd_pipe, "log cause: %d\n", state.cause_code);
    126 if (state.cause_code == kPatchApplicationFailure) {
    127         LOG(INFO) << "Patch application failed, retry update.";
    128 fprintf(cmd_pipe, "retry_update\n");
    129       } else if (state.cause_code == kEioFailure) {
    130         LOG(INFO) << "Update failed due to EIO, retry update.";
    131 fprintf(cmd_pipe, "retry_update\n");
    132       }
    133     }
    134 if (updater_info.package_zip) {
    135       CloseArchive(updater_info.package_zip);
    136     }
    137 return 7;
    138   } else {
    139 fprintf(cmd_pipe, "ui_print script succeeded: result was [%s]\n", result.c_str());
    140   }
    141 if (updater_info.package_zip) {
    142     CloseArchive(updater_info.package_zip);
    143   }
    144 return 0;
    145 }

    以上就是基於Android的OTARecovery模式升級流程。我這裏主要是梳理整個升級流程的主要,

    很多地方還是寫的不夠細,望讀者理解,我認為比較核心與關鍵的地方有以下幾點吧

    • 主系統與recovery升級系統,升級消息的傳遞通過cache;
    • BCB塊中寫信息來保證斷電續升;
    • 主系統中fork子進程進行升級進程的執行,並通過pipe管道進行信息交互;
    • updater中使用命令與執行的分離,命令在updater-script中,執行在update-binary中;
      • 升級程序通過升級包帶入的,那麼核心升級流程是每次都有機會變更或者優化的,
      • 這樣就比那些將升級流程預置在系統中的要靈活的很多;

     

                                         

     

     

            長按二維碼關注【嵌入式C部落】,獲取更多編程資料及精華文章

     

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    ※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站,提高曝光率!!

    網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司,幫您輕鬆架站!

    ※想知道最厲害的台北網頁設計公司推薦台中網頁設計公司推薦專業設計師”嚨底家”!!

  • 5種常見Bean映射工具的性能比對

    5種常見Bean映射工具的性能比對

    本文由 JavaGuide 翻譯自 https://www.baeldung.com/java-performance-mapping-frameworks 。轉載請註明原文地址以及翻譯作者。

    1. 介紹

    創建由多個層組成的大型 Java 應用程序需要使用多種領域模型,如持久化模型、領域模型或者所謂的 DTO。為不同的應用程序層使用多個模型將要求我們提供 bean 之間的映射方法。手動執行此操作可以快速創建大量樣板代碼並消耗大量時間。幸運的是,Java 有多個對象映射框架。在本教程中,我們將比較最流行的 Java 映射框架的性能。

    綜合日常使用情況和相關測試數據,個人感覺 MapStruct、ModelMapper 這兩個 Bean 映射框架是最佳選擇。

    2. 常見 Bean 映射框架概覽

    2.1. Dozer

    Dozer 是一個映射框架,它使用遞歸將數據從一個對象複製到另一個對象。框架不僅能夠在 bean 之間複製屬性,還能夠在不同類型之間自動轉換。

    要使用 Dozer 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
        <groupId>net.sf.dozer</groupId>
        <artifactId>dozer</artifactId>
        <version>5.5.1</version>
    </dependency>

    更多關於 Dozer 的內容可以在官方文檔中找到: http://dozer.sourceforge.net/documentation/gettingstarted.html ,或者你也可以閱讀這篇文章:https://www.baeldung.com/dozer 。

    2.2. Orika

    Orika 是一個 bean 到 bean 的映射框架,它遞歸地將數據從一個對象複製到另一個對象。

    Orika 的工作原理與 Dozer 相似。兩者之間的主要區別是 Orika 使用字節碼生成。這允許以最小的開銷生成更快的映射器。

    要使用 Orika 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
        <groupId>ma.glasnost.orika</groupId>
        <artifactId>orika-core</artifactId>
        <version>1.5.2</version>
    </dependency>

    更多關於 Orika 的內容可以在官方文檔中找到:https://orika-mapper.github.io/orika-docs/,或者你也可以閱讀這篇文章:https://www.baeldung.com/orika-mapping。

    2.3. MapStruct

    MapStruct 是一個自動生成 bean mapper 類的代碼生成器。MapStruct 還能夠在不同的數據類型之間進行轉換。Github 地址:https://github.com/mapstruct/mapstruct。

    要使用 MapStruct 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
        <groupId>org.mapstruct</groupId>
        <artifactId>mapstruct-processor</artifactId>
        <version>1.2.0.Final</version>
    </dependency>

    更多關於 MapStruct 的內容可以在官方文檔中找到:https://mapstruct.org/,或者你也可以閱讀這篇文章:https://www.baeldung.com/mapstruct。

    要使用 MapStruct 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
        <groupId>org.mapstruct</groupId>
        <artifactId>mapstruct-processor</artifactId>
        <version>1.2.0.Final</version>
    </dependency>

    2.4. ModelMapper

    ModelMapper 是一個旨在簡化對象映射的框架,它根據約定確定對象之間的映射方式。它提供了類型安全的和重構安全的 API。

    更多關於 ModelMapper 的內容可以在官方文檔中找到:http://modelmapper.org/ 。

    要使用 ModelMapper 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
      <groupId>org.modelmapper</groupId>
      <artifactId>modelmapper</artifactId>
      <version>1.1.0</version>
    </dependency>

    2.5. JMapper

    JMapper 是一個映射框架,旨在提供易於使用的、高性能的 Java bean 之間的映射。該框架旨在使用註釋和關係映射應用 DRY 原則。該框架允許不同的配置方式:基於註釋、XML 或基於 api。

    更多關於 JMapper 的內容可以在官方文檔中找到:https://github.com/jmapper-framework/jmapper-core/wiki。

    要使用 JMapper 框架,我們需要添加這樣的依賴到我們的項目:

    <dependency>
        <groupId>com.googlecode.jmapper-framework</groupId>
        <artifactId>jmapper-core</artifactId>
        <version>1.6.0.1</version>
    </dependency>
    

    3.測試模型

    為了能夠正確地測試映射,我們需要有一個源和目標模型。我們已經創建了兩個測試模型。

    第一個是一個只有一個字符串字段的簡單 POJO,它允許我們在更簡單的情況下比較框架,並檢查如果我們使用更複雜的 bean 是否會發生任何變化。

    簡單的源模型如下:

    public class SourceCode {
        String code;
        // getter and setter
    }
    

    它的目標也很相似:

    public class DestinationCode {
        String code;
        // getter and setter
    }

    源 bean 的實際示例如下:

    public class SourceOrder {
        private String orderFinishDate;
        private PaymentType paymentType;
        private Discount discount;
        private DeliveryData deliveryData;
        private User orderingUser;
        private List<Product> orderedProducts;
        private Shop offeringShop;
        private int orderId;
        private OrderStatus status;
        private LocalDate orderDate;
        // standard getters and setters
    }

    目標類如下圖所示:

    public class Order {
        private User orderingUser;
        private List<Product> orderedProducts;
        private OrderStatus orderStatus;
        private LocalDate orderDate;
        private LocalDate orderFinishDate;
        private PaymentType paymentType;
        private Discount discount;
        private int shopId;
        private DeliveryData deliveryData;
        private Shop offeringShop;
        // standard getters and setters
    }

    整個模型結構可以在這裏找到:https://github.com/eugenp/tutorials/tree/master/performance-tests/src/main/java/com/baeldung/performancetests/model/source。

    4. 轉換器

    為了簡化測試設置的設計,我們創建了如下所示的轉換器接口:

    public interface Converter {
        Order convert(SourceOrder sourceOrder);
        DestinationCode convert(SourceCode sourceCode);
    }

    我們所有的自定義映射器都將實現這個接口。

    4.1. OrikaConverter

    Orika 支持完整的 API 實現,這大大簡化了 mapper 的創建:

    public class OrikaConverter implements Converter{
        private MapperFacade mapperFacade;
    
        public OrikaConverter() {
            MapperFactory mapperFactory = new DefaultMapperFactory
              .Builder().build();
    
            mapperFactory.classMap(Order.class, SourceOrder.class)
              .field("orderStatus", "status").byDefault().register();
            mapperFacade = mapperFactory.getMapperFacade();
        }
    
        @Override
        public Order convert(SourceOrder sourceOrder) {
            return mapperFacade.map(sourceOrder, Order.class);
        }
    
        @Override
        public DestinationCode convert(SourceCode sourceCode) {
            return mapperFacade.map(sourceCode, DestinationCode.class);
        }
    }

    4.2. DozerConverter

    Dozer 需要 XML 映射文件,有以下幾個部分:

    <mappings xmlns="http://dozer.sourceforge.net"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://dozer.sourceforge.net
      http://dozer.sourceforge.net/schema/beanmapping.xsd">
    
        <mapping>
            <class-a>com.baeldung.performancetests.model.source.SourceOrder</class-a>
            <class-b>com.baeldung.performancetests.model.destination.Order</class-b>
            <field>
                <a>status</a>
                <b>orderStatus</b>
            </field>
        </mapping>
        <mapping>
            <class-a>com.baeldung.performancetests.model.source.SourceCode</class-a>
            <class-b>com.baeldung.performancetests.model.destination.DestinationCode</class-b>
        </mapping>
    </mappings>

    定義了 XML 映射后,我們可以從代碼中使用它:

    public class DozerConverter implements Converter {
        private final Mapper mapper;
    
        public DozerConverter() {
            DozerBeanMapper mapper = new DozerBeanMapper();
            mapper.addMapping(
              DozerConverter.class.getResourceAsStream("/dozer-mapping.xml"));
            this.mapper = mapper;
        }
    
        @Override
        public Order convert(SourceOrder sourceOrder) {
            return mapper.map(sourceOrder,Order.class);
        }
    
        @Override
        public DestinationCode convert(SourceCode sourceCode) {
            return mapper.map(sourceCode, DestinationCode.class);
        }
    }

    4.3. MapStructConverter

    Map 結構的定義非常簡單,因為它完全基於代碼生成:

    @Mapper
    public interface MapStructConverter extends Converter {
        MapStructConverter MAPPER = Mappers.getMapper(MapStructConverter.class);
    
        @Mapping(source = "status", target = "orderStatus")
        @Override
        Order convert(SourceOrder sourceOrder);
    
        @Override
        DestinationCode convert(SourceCode sourceCode);
    }

    4.4. JMapperConverter

    JMapperConverter 需要做更多的工作。接口實現后:

    public class JMapperConverter implements Converter {
        JMapper realLifeMapper;
        JMapper simpleMapper;
    
        public JMapperConverter() {
            JMapperAPI api = new JMapperAPI()
              .add(JMapperAPI.mappedClass(Order.class));
            realLifeMapper = new JMapper(Order.class, SourceOrder.class, api);
            JMapperAPI simpleApi = new JMapperAPI()
              .add(JMapperAPI.mappedClass(DestinationCode.class));
            simpleMapper = new JMapper(
              DestinationCode.class, SourceCode.class, simpleApi);
        }
    
        @Override
        public Order convert(SourceOrder sourceOrder) {
            return (Order) realLifeMapper.getDestination(sourceOrder);
        }
    
        @Override
        public DestinationCode convert(SourceCode sourceCode) {
            return (DestinationCode) simpleMapper.getDestination(sourceCode);
        }
    }

    我們還需要向目標類的每個字段添加@JMap註釋。此外,JMapper 不能在 enum 類型之間轉換,它需要我們創建自定義映射函數:

    @JMapConversion(from = "paymentType", to = "paymentType")
    public PaymentType conversion(com.baeldung.performancetests.model.source.PaymentType type) {
        PaymentType paymentType = null;
        switch(type) {
            case CARD:
                paymentType = PaymentType.CARD;
                break;
    
            case CASH:
                paymentType = PaymentType.CASH;
                break;
    
            case TRANSFER:
                paymentType = PaymentType.TRANSFER;
                break;
        }
        return paymentType;
    }

    4.5. ModelMapperConverter

    ModelMapperConverter 只需要提供我們想要映射的類:

    public class ModelMapperConverter implements Converter {
        private ModelMapper modelMapper;
    
        public ModelMapperConverter() {
            modelMapper = new ModelMapper();
        }
    
        @Override
        public Order convert(SourceOrder sourceOrder) {
           return modelMapper.map(sourceOrder, Order.class);
        }
    
        @Override
        public DestinationCode convert(SourceCode sourceCode) {
            return modelMapper.map(sourceCode, DestinationCode.class);
        }
    }
    

    5. 簡單的模型測試

    對於性能測試,我們可以使用 Java Microbenchmark Harness,關於如何使用它的更多信息可以在 這篇文章:https://www.baeldung.com/java-microbenchmark-harness 中找到。

    我們為每個轉換器創建了一個單獨的基準測試,並將基準測試模式指定為 Mode.All。

    5.1. 平均時間

    對於平均運行時間,JMH 返回以下結果(越少越好):

    這個基準測試清楚地表明,MapStruct 和 JMapper 都有最佳的平均工作時間。

    5.2. 吞吐量

    在這種模式下,基準測試返回每秒的操作數。我們收到以下結果(越多越好):

    在吞吐量模式中,MapStruct 是測試框架中最快的,JMapper 緊隨其後。

    5.3. SingleShotTime

    這種模式允許測量單個操作從開始到結束的時間。基準給出了以下結果(越少越好):

    這裏,我們看到 JMapper 返回的結果比 MapStruct 好得多。

    5.4. 採樣時間

    這種模式允許對每個操作的時間進行採樣。三個不同百分位數的結果如下:

    所有的基準測試都表明,根據場景的不同,MapStruct 和 JMapper 都是不錯的選擇,儘管 MapStruct 對 SingleShotTime 給出的結果要差得多。

    6. 真實模型測試

    對於性能測試,我們可以使用 Java Microbenchmark Harness,關於如何使用它的更多信息可以在 這篇文章:https://www.baeldung.com/java-microbenchmark-harness 中找到。

    我們為每個轉換器創建了一個單獨的基準測試,並將基準測試模式指定為 Mode.All。

    6.1. 平均時間

    JMH 返回以下平均運行時間結果(越少越好):

    該基準清楚地表明,MapStruct 和 JMapper 均具有最佳的平均工作時間。

    6.2. 吞吐量

    在這種模式下,基準測試返回每秒的操作數。我們收到以下結果(越多越好):

    在吞吐量模式中,MapStruct 是測試框架中最快的,JMapper 緊隨其後。

    6.3. SingleShotTime

    這種模式允許測量單個操作從開始到結束的時間。基準給出了以下結果(越少越好):

    6.4. 採樣時間

    這種模式允許對每個操作的時間進行採樣。三個不同百分位數的結果如下:

    儘管簡單示例和實際示例的確切結果明顯不同,但是它們的趨勢相同。在哪種算法最快和哪種算法最慢方面,兩個示例都給出了相似的結果。

    6.5. 結論

    根據我們在本節中執行的真實模型測試,我們可以看出,最佳性能顯然屬於 MapStruct。在相同的測試中,我們看到 Dozer 始終位於結果表的底部。

    7. 總結

    在這篇文章中,我們已經進行了五個流行的 Java Bean 映射框架性能測試:ModelMapper MapStruct Orika ,Dozer, JMapper。

    示例代碼地址:https://github.com/eugenp/tutorials/tree/master/performance-tests。

    開源項目推薦

    作者的其他開源項目推薦:

    1. :【Java學習+面試指南】 一份涵蓋大部分Java程序員所需要掌握的核心知識。
    2. : 適合新手入門以及有經驗的開發人員查閱的 Spring Boot 教程(業餘時間維護中,歡迎一起維護)。
    3. : 我覺得技術人員應該有的一些好習慣!
    4. :從零入門 !Spring Security With JWT(含權限驗證)後端部分代碼。

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

    ※帶您來了解什麼是 USB CONNECTOR  ?

    ※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

    ※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站,提高曝光率!!

    ※綠能、環保無空污,成為電動車最新代名詞,目前市場使用率逐漸普及化

    ※廣告預算用在刀口上,網站設計公司幫您達到更多曝光效益

  • RocketMQ ACL使用指南

    RocketMQ ACL使用指南

    目錄

    @(本節目錄)

    1、什麼是ACL?

    ACL是access control list的簡稱,俗稱訪問控制列表。訪問控制,基本上會涉及到用戶、資源、權限、角色等概念,那在RocketMQ中上述會對應哪些對象呢?

    • 用戶
      用戶是訪問控制的基礎要素,也不難理解,RocketMQ ACL必然也會引入用戶的概念,即支持用戶名、密碼。
    • 資源
      資源,需要保護的對象,在RocketMQ中,消息發送涉及的Topic、消息消費涉及的消費組,應該進行保護,故可以抽象成資源。
    • 權限
      針對資源,能進行的操作,
    • 角色
      RocketMQ中,只定義兩種角色:是否是管理員。

    另外,RocketMQ還支持按照客戶端IP進行白名單設置。

    2、ACL基本流程圖

    在講解如何使用ACL之前,我們先簡單看一下RocketMQ ACL的請求流程:

    對於上述具體的實現,將在後續文章中重點講解,本文的目的只是希望給讀者一個大概的了解。

    3、如何配置ACL

    3.1 acl配置文件

    acl默認的配置文件名:plain_acl.yml,需要放在${ROCKETMQ_HOME}/store/config目錄下。下面對其配置項一一介紹。

    3.1.1 globalWhiteRemoteAddresses

    全局白名單,其類型為數組,即支持多個配置。其支持的配置格式如下:


    • 表示不設置白名單,該條規則默認返回false。
    • “*”
      表示全部匹配,該條規則直接返回true,將會阻斷其他規則的判斷,請慎重使用。
    • 192.168.0.{100,101}
      多地址配置模式,ip地址的最後一組,使用{},大括號中多個ip地址,用英文逗號(,)隔開。
    • 192.168.1.100,192.168.2.100
      直接使用,分隔,配置多個ip地址。
    • 192.168..或192.168.100-200.10-20
      每個IP段使用 “*” 或”-“表示範圍。

    3.1.2 accounts

    配置用戶信息,該類型為數組類型。擁有accessKey、secretKey、whiteRemoteAddress、admin、defaultTopicPerm、defaultGroupPerm、topicPerms、groupPerms子元素。

    3.1.2.1 accessKey

    登錄用戶名,長度必須大於6個字符。

    3.1.2.2 secretKey

    登錄密碼。長度必須大於6個字符。

    3.1.2.3 whiteRemoteAddress

    用戶級別的IP地址白名單。其類型為一個字符串,其配置規則與globalWhiteRemoteAddresses,但只能配置一條規則。

    3.1.2.4 admin

    boolean類型,設置是否是admin。如下權限只有admin=true時才有權限執行。

    • UPDATE_AND_CREATE_TOPIC
      更新或創建主題。
    • UPDATE_BROKER_CONFIG
      更新Broker配置。
    • DELETE_TOPIC_IN_BROKER
      刪除主題。
    • UPDATE_AND_CREATE_SUBSCRIPTIONGROUP
      更新或創建訂閱組信息。
    • DELETE_SUBSCRIPTIONGROUP
      刪除訂閱組信息。
    3.1.2.5 defaultTopicPerm

    默認topic權限。該值默認為DENY(拒絕)。

    3.1.2.6 defaultGroupPerm

    默認消費組權限,該值默認為DENY(拒絕),建議值為SUB。

    3.1.2.7 topicPerms

    設置topic的權限。其類型為數組,其可選擇值在下節介紹。

    3.1.2.8 groupPerms

    設置消費組的權限。其類型為數組,其可選擇值在下節介紹。可以為每一消費組配置不一樣的權限。

    3.2 RocketMQ ACL權限可選值

    • DENY
      拒絕。
    • PUB
      擁有發送權限。
    • SUB
      擁有訂閱權限。

    3.3、權限驗證流程

    上面定義了全局白名單、用戶級別的白名單,用戶級別的權限,為了更好的配置ACL權限規則,下面給出權限匹配邏輯。

    4、使用示例

    4.1 Broker端安裝

    首先,需要在broker.conf文件中,增加參數aclEnable=true。並拷貝distribution/conf/plain_acl.yml文件到${ROCKETMQ_HOME}/conf目錄。

    broker.conf的配置文件如下:

    brokerClusterName = DefaultCluster
    brokerName = broker-b
    brokerId = 0
    deleteWhen = 04
    fileReservedTime = 48
    brokerRole = ASYNC_MASTER
    flushDiskType = ASYNC_FLUSH
    listenPort=10915
    storePathRootDir=E:/SH2019/tmp/rocketmq_home/rocketmq4.5MB/store
    storePathCommitLog=E:/SH2019/tmp/rocketmq_home/rocketmq4.5MB/store/commitlog
    namesrvAddr=127.0.0.1:9876
    autoCreateTopicEnable=false
    aclEnable=true

    plain_acl.yml文件內容如下:

    globalWhiteRemoteAddresses:
    
    accounts:
    - accessKey: RocketMQ
      secretKey: 12345678
      whiteRemoteAddress:
      admin: false
      defaultTopicPerm: DENY
      defaultGroupPerm: SUB
      topicPerms:
      - TopicTest=PUB
      groupPerms:
      # the group should convert to retry topic
      - oms_consumer_group=DENY
    
    - accessKey: admin
      secretKey: 12345678
      whiteRemoteAddress:
      # if it is admin, it could access all resources
      admin: true

    從上面的配置可知,用戶RocketMQ只能發送TopicTest的消息,其他topic無權限發送;拒絕oms_consumer_group消費組的消息消費,其他消費組默認可消費。

    4.2 消息發送端示例

    public class AclProducer {
        public static void main(String[] args) throws MQClientException, InterruptedException {
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name", getAclRPCHook());
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.start();
            for (int i = 0; i < 1; i++) {
                try {
                    Message msg = new Message("TopicTest3" ,"TagA" , ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
            producer.shutdown();
        }
    
        static RPCHook getAclRPCHook() {
            return new AclClientRPCHook(new SessionCredentials("rocketmq","12345678"));
        }
    }

    運行效果如圖所示:

    4.3 消息消費端示例

    public class AclConsumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4", getAclRPCHook(),new AllocateMessageQueueAveragely());
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe("TopicTest", "*");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    
        static RPCHook getAclRPCHook() {
            return new AclClientRPCHook(new SessionCredentials("rocketmq","12345678"));
        }
    }

    發現並不沒有消費消息,符合預期。

    關於RocketMQ ACL的使用就介紹到這裏了,下一篇將介紹RocketMQ ACL實現原理。

    推薦閱讀:
    1、

    2、

    3、

    4、

    作者介紹:
    丁威,《RocketMQ技術內幕》作者,RocketMQ 社區佈道師,公眾號: 維護者,目前已陸續發表源碼分析Java集合、Java 併發包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源碼專欄。

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    ※為什麼 USB CONNECTOR 是電子產業重要的元件?

    網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司,幫您輕鬆架站!

    ※想要讓你的商品成為最夯、最多人討論的話題?網頁設計公司讓你強力曝光

    ※想知道最厲害的台北網頁設計公司推薦台中網頁設計公司推薦專業設計師”嚨底家”!!

  • 詳解Kafka Producer

    詳解Kafka Producer

    上一篇文章我們主要介紹了什麼是 Kafka,Kafka 的基本概念是什麼,Kafka 單機和集群版的搭建,以及對基本的配置文件進行了大致的介紹,還對 Kafka 的幾個主要角色進行了描述,我們知道,不管是把 Kafka 用作消息隊列、消息總線還是數據存儲平台來使用,最終是繞不過消息這個詞的,這也是 Kafka 最最核心的內容,Kafka 的消息從哪裡來?到哪裡去?都干什麼了?別著急,一步一步來,先說說 Kafka 的消息從哪來。

    生產者概述

    在 Kafka 中,我們把產生消息的那一方稱為生產者,比如我們經常回去淘寶購物,你打開淘寶的那一刻,你的登陸信息,登陸次數都會作為消息傳輸到 Kafka 後台,當你瀏覽購物的時候,你的瀏覽信息,你的搜索指數,你的購物愛好都會作為一個個消息傳遞給 Kafka 後台,然後淘寶會根據你的愛好做智能推薦,致使你的錢包從來都禁不住誘惑,那麼這些生產者產生的消息是怎麼傳到 Kafka 應用程序的呢?發送過程是怎麼樣的呢?

    儘管消息的產生非常簡單,但是消息的發送過程還是比較複雜的,如圖

    我們從創建一個ProducerRecord 對象開始,ProducerRecord 是 Kafka 中的一個核心類,它代表了一組 Kafka 需要發送的 key/value 鍵值對,它由記錄要發送到的主題名稱(Topic Name),可選的分區號(Partition Number)以及可選的鍵值對構成。

    在發送 ProducerRecord 時,我們需要將鍵值對對象由序列化器轉換為字節數組,這樣它們才能夠在網絡上傳輸。然後消息到達了分區器。

    如果發送過程中指定了有效的分區號,那麼在發送記錄時將使用該分區。如果發送過程中未指定分區,則將使用key 的 hash 函數映射指定一個分區。如果發送的過程中既沒有分區號也沒有,則將以循環的方式分配一個分區。選好分區后,生產者就知道向哪個主題和分區發送數據了。

    ProducerRecord 還有關聯的時間戳,如果用戶沒有提供時間戳,那麼生產者將會在記錄中使用當前的時間作為時間戳。Kafka 最終使用的時間戳取決於 topic 主題配置的時間戳類型。

    • 如果將主題配置為使用 CreateTime,則生產者記錄中的時間戳將由 broker 使用。
    • 如果將主題配置為使用LogAppendTime,則生產者記錄中的時間戳在將消息添加到其日誌中時,將由 broker 重寫。

    然後,這條消息被存放在一個記錄批次里,這個批次里的所有消息會被發送到相同的主題和分區上。由一個獨立的線程負責把它們發到 Kafka Broker 上。

    Kafka Broker 在收到消息時會返回一個響應,如果寫入成功,會返回一個 RecordMetaData 對象,它包含了主題和分區信息,以及記錄在分區里的偏移量,上面兩種的時間戳類型也會返回給用戶。如果寫入失敗,會返回一個錯誤。生產者在收到錯誤之後會嘗試重新發送消息,幾次之後如果還是失敗的話,就返回錯誤消息。

    創建 Kafka 生產者

    要往 Kafka 寫入消息,首先需要創建一個生產者對象,並設置一些屬性。Kafka 生產者有3個必選的屬性

    • bootstrap.servers

    該屬性指定 broker 的地址清單,地址的格式為 host:port。清單里不需要包含所有的 broker 地址,生產者會從給定的 broker 里查找到其他的 broker 信息。不過建議至少要提供兩個 broker 信息,一旦其中一個宕機,生產者仍然能夠連接到集群上。

    • key.serializer

    broker 需要接收到序列化之後的 key/value值,所以生產者發送的消息需要經過序列化之後才傳遞給 Kafka Broker。生產者需要知道採用何種方式把 Java 對象轉換為字節數組。key.serializer 必須被設置為一個實現了org.apache.kafka.common.serialization.Serializer 接口的類,生產者會使用這個類把鍵對象序列化為字節數組。這裏拓展一下 Serializer 類

    Serializer 是一個接口,它表示類將會採用何種方式序列化,它的作用是把對象轉換為字節,實現了 Serializer 接口的類主要有 ByteArraySerializerStringSerializerIntegerSerializer ,其中 ByteArraySerialize 是 Kafka 默認使用的序列化器,其他的序列化器還有很多,你可以通過 查看其他序列化器。要注意的一點:key.serializer 是必須要設置的,即使你打算只發送值的內容

    • value.serializer

    與 key.serializer 一樣,value.serializer 指定的類會將值序列化。

    下面代碼演示了如何創建一個 Kafka 生產者,這裏只指定了必要的屬性,其他使用默認的配置

    private Properties properties = new Properties();
    properties.put("bootstrap.servers","broker1:9092,broker2:9092");
    properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    properties = new KafkaProducer<String,String>(properties);

    來解釋一下這段代碼

    • 首先創建了一個 Properties 對象
    • 使用 StringSerializer 序列化器序列化 key / value 鍵值對
    • 在這裏我們創建了一個新的生產者對象,併為鍵值設置了恰當的類型,然後把 Properties 對象傳遞給他。

    實例化生產者對象后,接下來就可以開始發送消息了,發送消息主要由下面幾種方式

    直接發送,不考慮結果

    使用這種發送方式,不會關心消息是否到達,會丟失一些消息,因為 Kafka 是高可用的,生產者會自動嘗試重發,這種發送方式和 UDP 運輸層協議很相似。

    同步發送

    同步發送仍然使用 send() 方法發送消息,它會返回一個 Future 對象,調用 get() 方法進行等待,就可以知道消息時候否發送成功。

    異步發送

    異步發送指的是我們調用 send() 方法,並制定一個回調函數,服務器在返迴響應時調用該函數。

    下一節我們會重新討論這三種實現。

    向 Kafka 發送消息

    簡單消息發送

    Kafka 最簡單的消息發送如下:

    ProducerRecord<String,String> record =
                    new ProducerRecord<String, String>("CustomerCountry","West","France");
    
    producer.send(record);

    代碼中生產者(producer)的 send() 方法需要把 ProducerRecord 的對象作為參數進行發送,ProducerRecord 有很多構造函數,這個我們下面討論,這裏調用的是

    public ProducerRecord(String topic, K key, V value) {}

    這個構造函數,需要傳遞的是 topic主題,key 和 value。

    把對應的參數傳遞完成后,生產者調用 send() 方法發送消息(ProducerRecord對象)。我們可以從生產者的架構圖中看出,消息是先被寫入分區中的緩衝區中,然後分批次發送給 Kafka Broker。

    發送成功后,send() 方法會返回一個 Future(java.util.concurrent) 對象,Future 對象的類型是 RecordMetadata 類型,我們上面這段代碼沒有考慮返回值,所以沒有生成對應的 Future 對象,所以沒有辦法知道消息是否發送成功。如果不是很重要的信息或者對結果不會產生影響的信息,可以使用這種方式進行發送。

    我們可以忽略發送消息時可能發生的錯誤或者在服務器端可能發生的錯誤,但在消息發送之前,生產者還可能發生其他的異常。這些異常有可能是 SerializationException(序列化失敗)BufferedExhaustedException 或 TimeoutException(說明緩衝區已滿),又或是 InterruptedException(說明發送線程被中斷)

    同步發送消息

    第二種消息發送機制如下所示

    ProducerRecord<String,String> record =
                    new ProducerRecord<String, String>("CustomerCountry","West","France");
    
    try{
      RecordMetadata recordMetadata = producer.send(record).get();
    }catch(Exception e){
      e.printStackTrace();
    }
    

    這種發送消息的方式較上面的發送方式有了改進,首先調用 send() 方法,然後再調用 get() 方法等待 Kafka 響應。如果服務器返回錯誤,get() 方法會拋出異常,如果沒有發生錯誤,我們會得到 RecordMetadata 對象,可以用它來查看消息記錄。

    生產者(KafkaProducer)在發送的過程中會出現兩類錯誤:其中一類是重試錯誤,這類錯誤可以通過重發消息來解決。比如連接的錯誤,可以通過再次建立連接來解決;無錯誤則可以通過重新為分區選舉首領來解決。KafkaProducer 被配置為自動重試,如果多次重試后仍無法解決問題,則會拋出重試異常。另一類錯誤是無法通過重試來解決的,比如消息過大對於這類錯誤,KafkaProducer 不會進行重試,直接拋出異常。

    異步發送消息

    同步發送消息都有個問題,那就是同一時間只能有一個消息在發送,這會造成許多消息無法直接發送,造成消息滯后,無法發揮效益最大化。

    比如消息在應用程序和 Kafka 集群之間一個來回需要 10ms。如果發送完每個消息后都等待響應的話,那麼發送100個消息需要 1 秒,但是如果是異步方式的話,發送 100 條消息所需要的時間就會少很多很多。大多數時候,雖然Kafka 會返回 RecordMetadata 消息,但是我們並不需要等待響應。

    為了在異步發送消息的同時能夠對異常情況進行處理,生產者提供了回掉支持。下面是回調的一個例子

    ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("CustomerCountry", "Huston", "America");
            producer.send(producerRecord,new DemoProducerCallBack());
    
    
    class DemoProducerCallBack implements Callback {
    
      public void onCompletion(RecordMetadata metadata, Exception exception) {
        if(exception != null){
          exception.printStackTrace();;
        }
      }
    }

    首先實現回調需要定義一個實現了org.apache.kafka.clients.producer.Callback的類,這個接口只有一個 onCompletion方法。如果 kafka 返回一個錯誤,onCompletion 方法會拋出一個非空(non null)異常,這裏我們只是簡單的把它打印出來,如果是生產環境需要更詳細的處理,然後在 send() 方法發送的時候傳遞一個 Callback 回調的對象。

    生產者分區機制

    Kafka 對於數據的讀寫是以分區為粒度的,分區可以分佈在多個主機(Broker)中,這樣每個節點能夠實現獨立的數據寫入和讀取,並且能夠通過增加新的節點來增加 Kafka 集群的吞吐量,通過分區部署在多個 Broker 來實現負載均衡的效果。

    上面我們介紹了生產者的發送方式有三種:不管結果如何直接發送發送並返回結果發送並回調。由於消息是存在主題(topic)的分區(partition)中的,所以當 Producer 生產者發送產生一條消息發給 topic 的時候,你如何判斷這條消息會存在哪個分區中呢?

    這其實就設計到 Kafka 的分區機制了。

    分區策略

    Kafka 的分區策略指的就是將生產者發送到哪個分區的算法。Kafka 為我們提供了默認的分區策略,同時它也支持你自定義分區策略。

    如果要自定義分區策略的話,你需要显示配置生產者端的參數 Partitioner.class,我們可以看一下這個類它位於 org.apache.kafka.clients.producer 包下

    public interface Partitioner extends Configurable, Closeable {
      
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
    
      public void close();
      
      default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {}
    }

    Partitioner 類有三個方法,分別來解釋一下

    • partition(): 這個類有幾個參數: topic,表示需要傳遞的主題;key 表示消息中的鍵值;keyBytes表示分區中序列化過後的key,byte數組的形式傳遞;value 表示消息的 value 值;valueBytes 表示分區中序列化后的值數組;cluster表示當前集群的原數據。Kafka 給你這麼多信息,就是希望讓你能夠充分地利用這些信息對消息進行分區,計算出它要被發送到哪個分區中。
    • close() : 繼承了 Closeable 接口能夠實現 close() 方法,在分區關閉時調用。
    • onNewBatch(): 表示通知分區程序用來創建新的批次

    其中與分區策略息息相關的就是 partition() 方法了,分區策略有下面這幾種

    順序輪訓

    順序分配,消息是均勻的分配給每個 partition,即每個分區存儲一次消息。就像下面這樣

    上圖表示的就是輪訓策略,輪訓策略是 Kafka Producer 提供的默認策略,如果你不使用指定的輪訓策略的話,Kafka 默認會使用順序輪訓策略的方式。

    隨機輪訓

    隨機輪訓簡而言之就是隨機的向 partition 中保存消息,如下圖所示

    實現隨機分配的代碼只需要兩行,如下

    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    return ThreadLocalRandom.current().nextInt(partitions.size());

    先計算出該主題總的分區數,然後隨機地返回一個小於它的正整數。

    本質上看隨機策略也是力求將數據均勻地打散到各個分區,但從實際表現來看,它要遜於輪詢策略,所以如果追求數據的均勻分佈,還是使用輪詢策略比較好。事實上,隨機策略是老版本生產者使用的分區策略,在新版本中已經改為輪詢了。

    按照 key 進行消息保存

    這個策略也叫做 key-ordering 策略,Kafka 中每條消息都會有自己的key,一旦消息被定義了 Key,那麼你就可以保證同一個 Key 的所有消息都進入到相同的分區裏面,由於每個分區下的消息處理都是有順序的,故這個策略被稱為按消息鍵保序策略,如下圖所示

    實現這個策略的 partition 方法同樣簡單,只需要下面兩行代碼即可:

    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    return Math.abs(key.hashCode()) % partitions.size();

    上面這幾種分區策略都是比較基礎的策略,除此之外,你還可以自定義分區策略。

    生產者壓縮機制

    壓縮一詞簡單來講就是一種互換思想,它是一種經典的用 CPU 時間去換磁盤空間或者 I/O 傳輸量的思想,希望以較小的 CPU 開銷帶來更少的磁盤佔用或更少的網絡 I/O 傳輸。如果你還不了解的話我希望你先讀完這篇文章 ,然後你就明白壓縮是怎麼回事了。

    Kafka 壓縮是什麼

    Kafka 的消息分為兩層:消息集合 和 消息。一個消息集合中包含若干條日誌項,而日誌項才是真正封裝消息的地方。Kafka 底層的消息日誌由一系列消息集合日誌項組成。Kafka 通常不會直接操作具體的一條條消息,它總是在消息集合這個層面上進行寫入操作。

    在 Kafka 中,壓縮會發生在兩個地方:Kafka Producer 和 Kafka Consumer,為什麼啟用壓縮?說白了就是消息太大,需要變小一點 來使消息發的更快一些。

    Kafka Producer 中使用 compression.type 來開啟壓縮

    private Properties properties = new Properties();
    properties.put("bootstrap.servers","192.168.1.9:9092");
    properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    properties.put("compression.type", "gzip");
    
    Producer<String,String> producer = new KafkaProducer<String, String>(properties);
    
    ProducerRecord<String,String> record =
      new ProducerRecord<String, String>("CustomerCountry","Precision Products","France");

    上面代碼錶明該 Producer 的壓縮算法使用的是 GZIP

    有壓縮必有解壓縮,Producer 使用壓縮算法壓縮消息后併發送給服務器后,由 Consumer 消費者進行解壓縮,因為採用的何種壓縮算法是隨着 key、value 一起發送過去的,所以消費者知道採用何種壓縮算法。

    Kafka 重要參數配置

    在上一篇文章 中,我們主要介紹了一下 kafka 集群搭建的參數,本篇文章我們來介紹一下 Kafka 生產者重要的配置,生產者有很多可配置的參數,在文檔里(

    key.serializer

    用於 key 鍵的序列化,它實現了 org.apache.kafka.common.serialization.Serializer 接口

    value.serializer

    用於 value 值的序列化,實現了 org.apache.kafka.common.serialization.Serializer 接口

    acks

    acks 參數指定了要有多少個分區副本接收消息,生產者才認為消息是寫入成功的。此參數對消息丟失的影響較大

    • 如果 acks = 0,就表示生產者也不知道自己產生的消息是否被服務器接收了,它才知道它寫成功了。如果發送的途中產生了錯誤,生產者也不知道,它也比較懵逼,因為沒有返回任何消息。這就類似於 UDP 的運輸層協議,只管發,服務器接受不接受它也不關心。
    • 如果 acks = 1,只要集群的 Leader 接收到消息,就會給生產者返回一條消息,告訴它寫入成功。如果發送途中造成了網絡異常或者 Leader 還沒選舉出來等其他情況導致消息寫入失敗,生產者會受到錯誤消息,這時候生產者往往會再次重發數據。因為消息的發送也分為 同步異步,Kafka 為了保證消息的高效傳輸會決定是同步發送還是異步發送。如果讓客戶端等待服務器的響應(通過調用 Future 中的 get() 方法),顯然會增加延遲,如果客戶端使用回調,就會解決這個問題。
    • 如果 acks = all,這種情況下是只有當所有參与複製的節點都收到消息時,生產者才會接收到一個來自服務器的消息。不過,它的延遲比 acks =1 時更高,因為我們要等待不只一個服務器節點接收消息。

    buffer.memory

    此參數用來設置生產者內存緩衝區的大小,生產者用它緩衝要發送到服務器的消息。如果應用程序發送消息的速度超過發送到服務器的速度,會導致生產者空間不足。這個時候,send() 方法調用要麼被阻塞,要麼拋出異常,具體取決於 block.on.buffer.null 參數的設置。

    compression.type

    此參數來表示生產者啟用何種壓縮算法,默認情況下,消息發送時不會被壓縮。該參數可以設置為 snappy、gzip 和 lz4,它指定了消息發送給 broker 之前使用哪一種壓縮算法進行壓縮。下面是各壓縮算法的對比

    retries

    生產者從服務器收到的錯誤有可能是臨時性的錯誤(比如分區找不到首領),在這種情況下,reteis 參數的值決定了生產者可以重發的消息次數,如果達到這個次數,生產者會放棄重試並返回錯誤。默認情況下,生產者在每次重試之間等待 100ms,這個等待參數可以通過 retry.backoff.ms 進行修改。

    batch.size

    當有多個消息需要被發送到同一個分區時,生產者會把它們放在同一個批次里。該參數指定了一個批次可以使用的內存大小,按照字節數計算。當批次被填滿,批次里的所有消息會被發送出去。不過生產者井不一定都會等到批次被填滿才發送,任意條數的消息都可能被發送。

    client.id

    此參數可以是任意的字符串,服務器會用它來識別消息的來源,一般配置在日誌里

    max.in.flight.requests.per.connection

    此參數指定了生產者在收到服務器響應之前可以發送多少消息,它的值越高,就會佔用越多的內存,不過也會提高吞吐量。把它設為1 可以保證消息是按照發送的順序寫入服務器。

    timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms

    request.timeout.ms 指定了生產者在發送數據時等待服務器返回的響應時間,metadata.fetch.timeout.ms 指定了生產者在獲取元數據(比如目標分區的首領是誰)時等待服務器返迴響應的時間。如果等待時間超時,生產者要麼重試發送數據,要麼返回一個錯誤。timeout.ms 指定了 broker 等待同步副本返回消息確認的時間,與 asks 的配置相匹配—-如果在指定時間內沒有收到同步副本的確認,那麼 broker 就會返回一個錯誤。

    max.block.ms

    此參數指定了在調用 send() 方法或使用 partitionFor() 方法獲取元數據時生產者的阻塞時間當生產者的發送緩衝區已捕,或者沒有可用的元數據時,這些方法就會阻塞。在阻塞時間達到 max.block.ms 時,生產者會拋出超時異常。

    max.request.size

    該參數用於控制生產者發送的請求大小。它可以指能發送的單個消息的最大值,也可以指單個請求里所有消息的總大小。

    receive.buffer.bytes 和 send.buffer.bytes

    Kafka 是基於 TCP 實現的,為了保證可靠的消息傳輸,這兩個參數分別指定了 TCP Socket 接收和發送數據包的緩衝區的大小。如果它們被設置為 -1,就使用操作系統的默認值。如果生產者或消費者與 broker 處於不同的數據中心,那麼可以適當增大這些值。

    文章參考:

    《Kafka 權威指南》

    極客時間 -《Kafka 核心技術與實戰》

    Kafka 源碼

    關注公眾號獲取更多優質电子書,關注一下你就知道資源是有多好了

    本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

    【其他文章推薦】

    USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

    ※評比前十大台北網頁設計台北網站設計公司知名案例作品心得分享

    ※智慧手機時代的來臨,RWD網頁設計已成為網頁設計推薦首選

    ※評比南投搬家公司費用收費行情懶人包大公開