這 10 行比較字符串相等的代碼給我整懵逼了,不信你也來看看

抱歉用這種標題吸引你點進來了,不過你不妨看完,看看能否讓你有所收穫。​(有收穫,請評論區留個言,沒收穫,下周末我直播吃**,哈哈,這你也信)

補充說明:微信公眾號改版,對各個號主影響還挺大的。目前從後台數據來看,對我影響不大,因為我這反正都是小號,閱讀量本身就少的可憐,真相了,狗頭(剛從交流群學會的表情)。

先直接上代碼:

boolean safeEqual(String a, String b) {
   if (a.length() != b.length()) {
       return false;
   }
   int equal = 0;
   for (int i = 0; i < a.length(); i++) {
       equal |= a.charAt(i) ^ b.charAt(i);
   }
   return equal == 0;
}

上面的代碼是我根據原版(Scala)翻譯成 Java的,Scala 版本(最開始吸引程序猿石頭注意力的代碼)如下:

def safeEqual(a: String, b: String) = {
  if (a.length != b.length) {
    false
  } else {
    var equal = 0
    for (i <- Array.range(0, a.length)) {
      equal |= a(i) ^ b(i)
    }
    equal == 0
  }
}

剛開始看到這段源碼感覺挺奇怪的,這個函數的功能是比較兩個字符串是否相等,首先“長度不等結果肯定不等,立即返回”這個很好理解。

再看看後面的,稍微動下腦筋,轉彎下也能明白這其中的門道:通過異或操作1^1=0, 1^0=1, 0^0=0,來比較每一位,如果每一位都相等的話,兩個字符串肯定相等,最後存儲累計異或值的變量equal必定為 0,否則為 1。

再細想一下呢?

for (i <- Array.range(0, a.length)) {
  if (a(i) ^ b(i) != 0// or a(i) != b[i]
    return false
}

我們常常講性能優化,從效率角度上講,難道不是應該只要中途發現某一位的結果不同了(即為1)就可以立即返回兩個字符串不相等了嗎?(如上所示)

這其中肯定有……

再再細想一下呢?

結合方法名稱 safeEquals 可能知道些眉目,與安全有關。

本文開篇的代碼來自playframewok 里用來驗證cookie(session)中的數據是否合法(包含簽名的驗證),也是石頭寫這篇文章的由來。

以前知道通過延遲計算等手段來提高效率的手段,但這種已經算出結果卻延遲返回的,還是頭一回!

我們來看看,JDK 中也有類似的方法,如下代碼摘自 java.security.MessageDigest

public static boolean isEqual(byte[] digesta, byte[] digestb) {
   if (digesta == digestb) return true;
   if (digesta == null || digestb == null) {
       return false;
   }
   if (digesta.length != digestb.length) {
       return false;
   }

   int result = 0;
   // time-constant comparison
   for (int i = 0; i < digesta.length; i++) {
       result |= digesta[i] ^ digestb[i];
   }
   return result == 0;
}

看註釋知道了,目的是為了用常量時間複雜度進行比較。

但這個計算過程耗費的時間不是常量有啥風險? (腦海里響起了背景音樂:“小朋友,你是否有很多問號?”)

真相大白

再深入探索和了解了一下,原來這麼做是為了防止計時攻擊(Timing Attack)。(也有人翻譯成時序攻擊​)​

計時攻擊(Timing Attack)

計時攻擊是邊信道攻擊(或稱”側信道攻擊”, Side Channel Attack, 簡稱SCA) 的一種,邊信道攻擊是一種針對軟件或硬件設計缺陷,走“歪門邪道”的一種攻擊方式。

這種攻擊方式是通過功耗、時序、電磁泄漏等方式達到破解目的。在很多物理隔絕的環境中,往往也能出奇制勝,這類新型攻擊的有效性遠高於傳統的密碼分析的數學方法(某百科上說的)。

這種手段可以讓調用 safeEquals("abcdefghijklmn", "xbcdefghijklmn") (只有首位不相同)和調用 safeEquals("abcdefghijklmn", "abcdefghijklmn") (兩個完全相同的字符串)的所耗費的時間一樣。防止通過大量的改變輸入並通過統計運行時間來暴力破解出要比較的字符串。

舉個,如果用之前說的“高效”的方式來實現的話。假設某個用戶設置了密碼為 password,通過從a到z(實際範圍可能更廣)不斷枚舉第一位,最終統計發現 p0000000 的運行時間比其他從任意a~z的都長(因為要到第二位才能發現不同,其他非 p 開頭的字符串第一位不同就直接返回了),這樣就能猜測出用戶密碼的第一位很可能是p了,然後再不斷一位一位迭代下去最終破解出用戶的密碼。

當然,以上是從理論角度分析,確實容易理解。但實際上好像通過統計運行時間總感覺不太靠譜,這個運行時間對環境太敏感了,比如網絡,內存,CPU負載等等都會影響。

但安全問題感覺更像是 “寧可信其有,不可信其無”。為了防止(特別是與簽名/密碼驗證等相關的操作)被 timing attack,目前各大語言都提供了相應的安全比較函數。各種軟件系統(例如 OpenSSL)、框架(例如 Play)的實現也都採用了這種方式。

例如 “世界上最好的編程語言”(粉絲較少,評論區應該打不起架來)—— php中的:

// Compares two strings using the same time whether they're equal or not.
// This function should be used to mitigate timing attacks; 
// for instance, when testing crypt() password hashes.
bool hash_equals ( string $known_string , string $user_string )

//This function is safe against timing attacks.
boolean password_verify ( string $password , string $hash )

其實各種語言版本的實現方式都與上面的版本差不多,將兩個字符串每一位取出來異或(^)並用或(|)保存,最後通過判斷結果是否為 0 來確定兩個字符串是否相等。

如果剛開始沒有用 safeEquals 去實現,後續的版本還會通過打補丁的方式去修復這樣的安全隱患。

例如 JDK 1.6.0_17 中的Release Notes[1]中就提到了MessageDigest.isEqual 中的bug的修復,如下圖所示:

MessageDigest timing attack vulnerabilities

大家可以看看這次變更的的詳細信息openjdk中的 bug fix diff[2]為:

MessageDigest.isEqual計時攻擊

Timing Attack 真的可行嗎?

我覺得各大語言的 API 都用這種實現,肯定還是有道理的,理論上應該可以被利用的。 這不,學術界的這篇論文就宣稱用這種計時攻擊的方法破解了 OpenSSL 0.9.7 的RSA加密算法了。關於 RSA 算法的介紹可以看看之前本人寫的這篇文章。

這篇Remote Timing Attacks are Practical[3] 論文中指出(我大致翻譯下摘要,感興趣的同學可以通過文末鏈接去看原文):

計時攻擊往往用於攻擊一些性能較弱的計算設備,例如一些智能卡。我們通過實驗發現,也能用於攻擊普通的軟件系統。本文通過實驗證明,通過這種計時攻擊方式能夠攻破一個基於 OpenSSL 的 web 服務器的私鑰。結果證明計時攻擊用於進行網絡攻擊在實踐中可行的,因此各大安全系統需要抵禦這種風險。

最後,本人畢竟不是專研完全方向,以上描述是基於本人的理解,如果有不對的地方,還請大家留言指出來。感謝。

補充說明2:感謝正在閱讀文章的你,讓我還有動力繼續堅持更新原創。

本人發文不多,但希望寫的文章能達到的目的是:佔用你的閱讀時間,就盡量能夠讓你有所收穫。

如果你覺得我的文章有所幫助,還請你幫忙轉發分享,另外請別忘了點擊公眾號右上角加個星標,好讓你別錯過後續的精彩文章(微信改版了,或許我發的文章都不能推送到你那了)。

​原創真心不易,希望你能幫我個小忙唄,如果本文內容你覺得有所啟發,有所收穫,請幫忙點個“在看”唄,或者轉發分享讓更多的小夥伴看到。 ​ 參考資料:

  • Timing Attacks on RSA: Revealing Your Secrets through the Fourth Dimension
  • Remote Timing Attacks are Practical

 

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

【其他文章推薦】

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

網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※推薦評價好的iphone維修中心

SpringColud Eureka的服務註冊與發現

一、Eureka簡介

本文中所有代碼都會上傳到git上,請放心瀏覽
項目git地址:https://github.com/839022478/Spring-Cloud

在傳統應用中,組件之間的調用,通過有規範的約束的接口來實現,從而實現不同模塊間良好的協作。但是被拆分成微服務后,每個微服務實例的網絡地址都可能動態變化,數量也會變化,使得原來硬編碼的地址失去了作用。需要一个中心化的組件來進行服務的登記和管理,為了解決上面的問題,於是出現了服務治理,就是管理所有的服務信息和狀態,也就是我們所說的註冊中心

1.1 註冊中心

比如我們去做火車或者汽車,需要去買票乘車,只看我們有沒有票(有沒有服務),有就去買票(獲取註冊列表),然後乘車(調用),不用關心到底有多少車在運行

流程圖:

使用註冊中心,我們不需要關心有多少提供方,只管去調用就可以了,那麼註冊中心有哪些呢?

註冊中心:Eureka,Nacos,Consul,Zookeeper

本文中講解的是比較火熱的Spring Cloud微服務下的Eureka,Eureka是Netflix開發的服務發現框架,是一個RESTful風格的服務,是一個用於服務發現和註冊的基礎組件,是搭建Spring Cloud微服務的前提之一,它屏蔽了Server和client的交互細節,使得開發者將精力放到業務上。

服務註冊與發現主要包括兩個部分:服務端(Eureka Server)和客戶端(Eureka Client)

  • 服務端(Eureka Server): 一個公共服務,為Client提供服務註冊和發現的功能,維護註冊到自身的Client的相關信息,同時提供接口給Client獲取註冊表中其他服務的信息,使得動態變化的Client能夠進行服務間的相互調用。

  • 客戶端(Eureka Client): Client將自己的服務信息通過一定的方式登記到Server上,並在正常範圍內維護自己信息一致性,方便其他服務發現自己,同時可以通過Server獲取到自己依賴的其他服務信息,完成服務調用,還內置了負載均衡器,用來進行基本的負載均衡

Eureka GIt官網:https://github.com/Netflix/Eureka

1.3 服務註冊與發現

服務註冊與發現關係圖:

1.2 client功能和server功能

1.2.1 client功能

  1. 註冊:每個微服務啟動時,將自己的網絡地址等信息註冊到註冊中心,註冊中心會存儲(內存中)這些信息。
  2. 獲取服務註冊表:服務消費者從註冊中心,查詢服務提供者的網絡地址,並使用該地址調用服務提供者,為了避免每次都查註冊表信息,所以client會定時去server拉取註冊表信息到緩存到client本地。
  3. 心跳:各個微服務與註冊中心通過某種機制(心跳)通信,若註冊中心長時間和服務間沒有通信,就會註銷該實例。
  4. 調用:實際的服務調用,通過註冊表,解析服務名和具體地址的對應關係,找到具體服務的地址,進行實際調用。

1.2.2 server註冊中心功能

  1. 服務註冊表:記錄各個微服務信息,例如服務名稱,ip,端口等。
    註冊表提供 查詢API(查詢可用的微服務實例)和管理API(用於服務的註冊和註銷)。
  2. 服務註冊與發現:註冊:將微服務信息註冊到註冊中心。發現:查詢可用微服務列表及其網絡地址。
  3. 服務檢查:定時檢測已註冊的服務,如發現某實例長時間無法訪問,就從註冊表中移除。

二、Eureka單節點搭建

2.1 pom.xml

在有的教程中,會引入spring-boot-starter-web,這個依賴其實不用,因為spring-cloud-starter-netflix-eureka-server的依賴已經包含了它,在pom依賴進去,就可以了

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
</dependency>

2.2 application.yml

server:
  port: 8500
eureka:
  client:
    #是否將自己註冊到Eureka Server,默認為true,由於當前就是server,故而設置成false,表明該服務不會向eureka註冊自己的信息
    register-with-eureka: false
    #是否從eureka server獲取註冊信息,由於單節點,不需要同步其他節點數據,用false
    fetch-registry: false
    #設置服務註冊中心的URL,用於client和server端交流
    service-url:
      defaultZone: http://localhost:8080/eureka/

2.3 服務端啟動類

啟動類上添加此註解標識該服務為配置中心
@EnableEurekaServer

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;

@EnableEurekaServer
@SpringBootApplication
public class EurekaServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(EurekaServerApplication.class, args);
    }

}

2.4 啟動

我們啟動EurekaDemoApplication ,然後在瀏覽器中輸入地址 http://localhost:8500/,就可以啟動我們的 Eureka 了,我們來看下效果,出現了這個畫面,就說明我們已經成功啟動~,只是此時我們的服務中是還沒有客戶端進行註冊

三、服務註冊

注意:在客戶端pom裏面我們需要加上spring-boot-starter-web,否則服務是無法正常啟動的

3.1 pom.xml

       <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>     

3.2 application.yml

#註冊中心
eureka:
  client:
    #設置服務註冊中心的URL
    service-url:
      defaultZone: http://localhost:8500/eureka/
  #服務名
  instance:
    appname: mxn

3.3 客戶端啟動類

在客戶端啟動類中我們需要加上 @EnableDiscoveryClient註解

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;

@EnableDiscoveryClient
@SpringBootApplication
public class EurekaClientApplication {

    public static void main(String[] args) {
        SpringApplication.run(EurekaClientApplication.class, args);
    }
}

3.4 查看效果

工程啟動后,刷新http://localhost:8500/頁面,我們可以發現服務註冊成功了

並且我們可以在idea日誌打印中看到DiscoveryClient_MXN/DESKTOP-5BQ3UK8 - registration status: 204,說明就是註冊成功了
Eureka Server與Eureka Client之間的聯繫主要通過心跳的方式實現。心跳(Heartbeat)即Eureka Client定時向Eureka Server彙報本服務實例當前的狀態,維護本服務實例在註冊表中租約的有效性。

Eureka Client將定時從Eureka Server中拉取註冊表中的信息,並將這些信息緩存到本地,用於服務發現

四、Eureka 端點

官網地址:https://github.com/Netflix/eureka/wiki/Eureka-REST-operations

Eureka服務器還提供了一個端點(eureka/apps/{applicaitonName})可以查看所註冊的服務詳細信息 。applicaitonName就是微服務的名稱,比如這裏我們訪問 http://localhost:8500/eureka/apps/mxn

五、Eureka 原理

5.1 本質

存儲了每個客戶端的註冊信息。EurekaClient從EurekaServer同步獲取服務註冊列表。通過一定的規則選擇一個服務進行調用

5.2 Eureka架構圖

  • 服務提供者: 是一個eureka client,向Eureka Server註冊和更新自己的信息,同時能從Eureka Server註冊表中獲取到其他服務的信息。
  • 服務註冊中心: 提供服務註冊和發現的功能。每個Eureka Cient向Eureka Server註冊自己的信息,也可以通過Eureka Server獲取到其他服務的信息達到發現和調用其他服務的目的。
  • 服務消費者: 是一個eureka client,通過Eureka Server獲取註冊到其上其他服務的信息,從而根據信息找到所需的服務發起遠程調用。
  • 同步複製: Eureka Server之間註冊表信息的同步複製,使Eureka Server集群中不同註冊表中服務實例信息保持一致。
  • 遠程調用: 服務客戶端之間的遠程調用。
  • 註冊: Client端向Server端註冊自身的元數據以供服務發現。
  • 續約: 通過發送心跳到Server以維持和更新註冊表中服務實例元數據的有效性。當在一定時長內,Server沒有收到Client的心跳信息,將默認服務下線,會把服務實例的信息從註冊表中刪除。
  • 下線: Client在關閉時主動向Server註銷服務實例元數據,這時Client的服務實例數據將從Server的註冊表中刪除。
  • 獲取註冊表: Client向Server請求註冊表信息,用於服務發現,從而發起服務間遠程調用。

5.3 Eureka自我保護

有時候我們會看到這樣的提示信息:EMERGENCY! EUREKA MAY BE INCORRECTLY CLAIMING INSTANCES ARE UP WHEN THEY'RE NOT. RENEWALS ARE LESSER THAN THRESHOLD AND HENCE THE INSTANCES ARE NOT BEING EXPIRED JUST TO BE SAFE.,這是因為默認情況下,Eureka Server在一定時間內,沒有接收到某個微服務心跳,會將某個微服務註銷(90S)。但是當網絡故障時,微服務與Server之間無法正常通信,上述行為就非常危險,因為微服務正常,不應該註銷,它的指導思想就是 寧可保留健康的和不健康的,也不盲目註銷任何健康的服務
我們也可以通過命令去關閉自我保護的功能:

eureka:
  server: 
    enable-self-preservation: false

那麼自我保護是如何觸發的呢?
自我保護機制的觸發條件是,當每分鐘心跳次數( renewsLastMin) 小於 numberOfRenewsPerMinThreshold時,並且開啟自動保護模式開關( eureka.server.enable-self-preservation = true) 時,觸發自我保護機制,不再自動過期租約
上面我們所有的小於 numberOfRenewsPerMinThreshold,到底是怎麼計算的呢,我們在eureka源碼中可以得知

numberOfRenewsPerMinThreshold = expectedNumberOfRenewsPerMin * 續租百分比(默認為0.85)
expectedNumberOfRenewsPerMin = 當前註冊的應用實例數 x 2
當前註冊的應用實例數 x 2 是因為,在默認情況下,註冊的應用實例每半分鐘續租一次,那麼一分鐘心跳兩次,因此 x 2

例如:我們有10個服務,期望每分鐘續約數:10 * 2=20,期望閾值:20*0.85=17,當少於17時,就會觸發自我保護機制

5.4 健康檢查

由於server和client通過心跳保持 服務狀態,而只有狀態為UP的服務才能被訪問。看eureka界面中的status。

比如心跳一直正常,服務一直UP,但是此服務DB(數據庫)連不上了,無法正常提供服務。

此時,我們需要將 微服務的健康狀態也同步到server。只需要啟動eureka的健康檢查就行。這樣微服務就會將自己的健康狀態同步到eureka。配置如下即可。

在client端配置:將自己的健康狀態傳播到server。

eureka:
  client:
    healthcheck:
      enabled: true

5.5 Eureka監聽事件

import com.netflix.appinfo.InstanceInfo;
import org.springframework.cloud.netflix.eureka.server.event.*;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

@Component
public class CustomEvent {

    @EventListener
    public void listen(EurekaInstanceCanceledEvent event ) {
        System.out.println(LocalDateTime.now()+"服務下線事件:"+event.getAppName()+"---"+event.getServerId());
//發釘釘
    }

    @EventListener
    public void listen(EurekaInstanceRegisteredEvent event) {
        InstanceInfo instanceInfo = event.getInstanceInfo();
        System.out.println(LocalDateTime.now()+"服務上線事件:"+instanceInfo.getAppName()+"---"+instanceInfo.getInstanceId());
    }

    @EventListener
    public void listen(EurekaInstanceRenewedEvent event) {
        System.out.println(LocalDateTime.now()+"服務續約/心跳上報事件:"+event.getAppName()+"---"+event.getServerId());

    }

    @EventListener
    public void listen(EurekaRegistryAvailableEvent event) {
        System.out.println(LocalDateTime.now()+"註冊中心可用事件");
    }

    @EventListener
    public void listen(EurekaServerStartedEvent event) {
        System.out.println(LocalDateTime.now()+"註冊中心啟動事件");

    }
}

5.6 Renew: 服務續約

Eureka Client 會每隔 30 秒發送一次心跳來續約。 通過續約來告知 Eureka Server 該 Eureka Client 運行正常,沒有出現問題。 默認情況下,如果 Eureka Server 在 90 秒內沒有收到 Eureka Client 的續約,Server 端會將實例從其註冊表中刪除,此時間可配置,一般情況不建議更改。

5.6 服務剔除

如果Eureka Client在註冊后,既沒有續約,也沒有下線(服務崩潰或者網絡異常等原因),那麼服務的狀態就處於不可知的狀態,不能保證能夠從該服務實例中獲取到回饋,所以需要服務剔除此方法定時清理這些不穩定的服務,該方法會批量將註冊表中所有過期租約剔除,剔除是定時任務,默認60秒執行一次。延時60秒,間隔60秒

剔除的限制:
1.自我保護期間不清除。
2.分批次清除。

六、Eureka缺陷

由於集群間的同步複製是通過HTTP的方式進行,基於網絡的不可靠性,集群中的Eureka Server間的註冊表信息難免存在不同步的時間節點,不滿足CAP中的C(數據一致性)

七、總結

中間我們講解了eureka的節點搭建,以及原理,對於現在很火熱的微服務,我們對Eureka是非常有必要進行了解的,如果覺得文章對你有幫助,來個點贊支持吧,如果對文章有疑問或建議,歡迎討論留言,謝謝大家~

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

【其他文章推薦】

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

台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨,RWD網頁設計為架站首選

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

※回頭車貨運收費標準

線上服務的FGC問題排查,看這篇就夠了!

線上服務的GC問題,是Java程序非常典型的一類問題,非常考驗工程師排查問題的能力。同時,幾乎是面試必考題,但是能真正答好此題的人並不多,要麼原理沒吃透,要麼缺乏實戰經驗。

過去半年時間里,我們的廣告系統出現了多次和GC相關的線上問題,有Full GC過於頻繁的,有Young GC耗時過長的,這些問題帶來的影響是:GC過程中的程序卡頓,進一步導致服務超時從而影響到廣告收入。

這篇文章,我將以一個FGC頻繁的線上案例作為引子,詳細介紹下GC的排查過程,另外會結合GC的運行原理給出一份實踐指南,希望對你有所幫助。內容分成以下3個部分:

1、從一次FGC頻繁的線上案例說起

2、GC的運行原理介紹

3、排查FGC問題的實踐指南

01 從一次FGC頻繁的線上案例說起

去年10月份,我們的廣告召回系統在程序上線后收到了FGC頻繁的系統告警,通過下面的監控圖可以看到:平均每35分鐘就進行了一次FGC。而程序上線前,我們的FGC頻次大概是2天一次。下面,詳細介紹下該問題的排查過程。

1. 檢查JVM配置

通過以下命令查看JVM的啟動參數:
ps aux | grep “applicationName=adsearch”

-Xms4g -Xmx4g -Xmn2g -Xss1024K
-XX:ParallelGCThreads=5
-XX:+UseConcMarkSweepGC
-XX:+UseParNewGC
-XX:+UseCMSCompactAtFullCollection
-XX:CMSInitiatingOccupancyFraction=80

可以看到堆內存為4G,新生代為2G,老年代也為2G,新生代採用ParNew收集器,老年代採用併發標記清除的CMS收集器,當老年代的內存佔用率達到80%時會進行FGC。

進一步通過 jmap -heap 7276 | head -n20 可以得知新生代的Eden區為1.6G,S0和S1區均為0.2G。

2. 觀察老年代的內存變化

通過觀察老年代的使用情況,可以看到:每次FGC后,內存都能回到500M左右,因此我們排除了內存泄漏的情況。

3. 通過jmap命令查看堆內存中的對象

通過命令 jmap -histo 7276 | head -n20

上圖中,按照對象所佔內存大小排序,显示了存活對象的實例數、所佔內存、類名。可以看到排名第一的是:int[],而且所佔內存大小遠遠超過其他存活對象。至此,我們將懷疑目標鎖定在了 int[] .

4. 進一步dump堆內存文件進行分析

鎖定 int[] 后,我們打算dump堆內存文件,通過可視化工具進一步跟蹤對象的來源。考慮堆轉儲過程中會暫停程序,因此我們先從服務管理平台摘掉了此節點,然後通過以下命令dump堆內存:

jmap -dump:format=b,file=heap 7276

通過JVisualVM工具導入dump出來的堆內存文件,同樣可以看到各個對象所佔空間,其中int[]佔到了50%以上的內存,進一步往下便可以找到 int[] 所屬的業務對象,發現它來自於架構團隊提供的codis基礎組件。

5. 通過代碼分析可疑對象

通過代碼分析,codis基礎組件每分鐘會生成約40M大小的int數組,用於統計TP99 和 TP90,數組的生命周期是一分鐘。而根據第2步觀察老年代的內存變化時,發現老年代的內存基本上也是每分鐘增加40多M,因此推斷:這40M的int數組應該是從新生代晉陞到老年代。

我們進一步查看了YGC的頻次監控,通過下圖可以看到大概1分鐘有8次左右的YGC,這樣基本驗證了我們的推斷:因為CMS收集器默認的分代年齡是6次,即YGC 6次后還存活的對象就會晉陞到老年代,而codis組件中的大數組生命周期是1分鐘,剛好滿足這個要求。

至此,整個排查過程基本結束了,那為什麼程序上線前沒出現此問題呢?通過上圖可以看到:程序上線前YGC的頻次在5次左右,此次上線后YGC頻次變成了8次左右,從而引發了此問題。

6. 解決方案

為了快速解決問題,我們將CMS收集器的分代年齡改成了15次,改完后FGC頻次恢復到了2天一次,後續如果YGC的頻次超過每分鐘15次還會再次觸發此問題。當然,我們最根本的解決方案是:優化程序以降低YGC的頻率,同時縮短codis組件中int數組的生命周期,這裏就不做展開了。

02 GC的運行原理介紹

上面整個案例的分析過程中,其實涉及到很多GC的原理知識,如果不懂得這些原理就着手處理,其實整個排查過程是很抓瞎的。

這裏,我選擇幾個最核心的知識點,展開介紹下GC的運行原理,最後再給出一份實踐指南。

1. 堆內存結構

大家都知道: GC分為YGC和FGC,它們均發生在JVM的堆內存上。先來看下JDK8的堆內存結構:

可以看到,堆內存採用了分代結構,包括新生代和老年代。新生代又分為:Eden區,From Survivor區(簡稱S0),To Survivor區(簡稱S1區),三者的默認比例為8:1:1。另外,新生代和老年代的默認比例為1:2。

堆內存之所以採用分代結構,是考慮到絕大部分對象都是短生命周期的,這樣不同生命周期的對象可放在不同的區域中,然後針對新生代和老年代採用不同的垃圾回收算法,從而使得GC效率最高。

2. YGC是什麼時候觸發的?

大多數情況下,對象直接在年輕代中的Eden區進行分配,如果Eden區域沒有足夠的空間,那麼就會觸發YGC(Minor GC),YGC處理的區域只有新生代。因為大部分對象在短時間內都是可收回掉的,因此YGC后只有極少數的對象能存活下來,而被移動到S0區(採用的是複製算法)。

當觸發下一次YGC時,會將Eden區和S0區的存活對象移動到S1區,同時清空Eden區和S0區。當再次觸發YGC時,這時候處理的區域就變成了Eden區和S1區(即S0和S1進行角色交換)。每經過一次YGC,存活對象的年齡就會加1。

3. FGC又是什麼時候觸發的?

下面4種情況,對象會進入到老年代中:

1、YGC時,To Survivor區不足以存放存活的對象,對象會直接進入到老年代。

2、經過多次YGC后,如果存活對象的年齡達到了設定閾值,則會晉陞到老年代中。

3、動態年齡判定規則,To Survivor區中相同年齡的對象,如果其大小之和佔到了 To Survivor區一半以上的空間,那麼大於此年齡的對象會直接進入老年代,而不需要達到默認的分代年齡。

4、大對象:由-XX:PretenureSizeThreshold啟動參數控制,若對象大小大於此值,就會繞過新生代, 直接在老年代中分配。

當晉陞到老年代的對象大於了老年代的剩餘空間時,就會觸發FGC(Major GC),FGC處理的區域同時包括新生代和老年代。除此之外,還有以下4種情況也會觸發FGC:

1、老年代的內存使用率達到了一定閾值(可通過參數調整),直接觸發FGC。

2、空間分配擔保:在YGC之前,會先檢查老年代最大可用的連續空間是否大於新生代所有對象的總空間。如果小於,說明YGC是不安全的,則會查看參數 HandlePromotionFailure 是否被設置成了允許擔保失敗,如果不允許則直接觸發Full GC;如果允許,那麼會進一步檢查老年代最大可用的連續空間是否大於歷次晉陞到老年代對象的平均大小,如果小於也會觸發 Full GC。

3、Metaspace(元空間)在空間不足時會進行擴容,當擴容到了-XX:MetaspaceSize 參數的指定值時,也會觸發FGC。

4、System.gc() 或者Runtime.gc() 被顯式調用時,觸發FGC。

4. 在什麼情況下,GC會對程序產生影響?

不管YGC還是FGC,都會造成一定程度的程序卡頓(即Stop The World問題:GC線程開始工作,其他工作線程被掛起),即使採用ParNew、CMS或者G1這些更先進的垃圾回收算法,也只是在減少卡頓時間,而並不能完全消除卡頓。

那到底什麼情況下,GC會對程序產生影響呢?根據嚴重程度從高到底,我認為包括以下4種情況:

1、FGC過於頻繁:FGC通常是比較慢的,少則幾百毫秒,多則幾秒,正常情況FGC每隔幾個小時甚至幾天才執行一次,對系統的影響還能接受。但是,一旦出現FGC頻繁(比如幾十分鐘就會執行一次),這種肯定是存在問題的,它會導致工作線程頻繁被停止,讓系統看起來一直有卡頓現象,也會使得程序的整體性能變差。

2、YGC耗時過長:一般來說,YGC的總耗時在幾十或者上百毫秒是比較正常的,雖然會引起系統卡頓幾毫秒或者幾十毫秒,這種情況幾乎對用戶無感知,對程序的影響可以忽略不計。但是如果YGC耗時達到了1秒甚至幾秒(都快趕上FGC的耗時了),那卡頓時間就會增大,加上YGC本身比較頻繁,就會導致比較多的服務超時問題。

3、FGC耗時過長:FGC耗時增加,卡頓時間也會隨之增加,尤其對於高併發服務,可能導致FGC期間比較多的超時問題,可用性降低,這種也需要關注。

4、YGC過於頻繁:即使YGC不會引起服務超時,但是YGC過於頻繁也會降低服務的整體性能,對於高併發服務也是需要關注的。

其中,「FGC過於頻繁」和「YGC耗時過長」,這兩種情況屬於比較典型的GC問題,大概率會對程序的服務質量產生影響。剩餘兩種情況的嚴重程度低一些,但是對於高併發或者高可用的程序也需要關注。

03 排查FGC問題的實踐指南

通過上面的案例分析以及理論介紹,再總結下FGC問題的排查思路,作為一份實踐指南供大家參考。

1. 清楚從程序角度,有哪些原因導致FGC?

1、大對象:系統一次性加載了過多數據到內存中(比如SQL查詢未做分頁),導致大對象進入了老年代。

2、內存泄漏:頻繁創建了大量對象,但是無法被回收(比如IO對象使用完后未調用close方法釋放資源),先引發FGC,最後導致OOM.

3、程序頻繁生成一些長生命周期的對象,當這些對象的存活年齡超過分代年齡時便會進入老年代,最後引發FGC. (即本文中的案例)

4、程序BUG導致動態生成了很多新類,使得 Metaspace 不斷被佔用,先引發FGC,最後導致OOM.

5、代碼中顯式調用了gc方法,包括自己的代碼甚至框架中的代碼。

6、JVM參數設置問題:包括總內存大小、新生代和老年代的大小、Eden區和S區的大小、元空間大小、垃圾回收算法等等。

2. 清楚排查問題時能使用哪些工具

1、公司的監控系統:大部分公司都會有,可全方位監控JVM的各項指標。

2、JDK的自帶工具,包括jmap、jstat等常用命令:

查看堆內存各區域的使用率以及GC情況
jstat -gcutil -h20 pid 1000

查看堆內存中的存活對象,並按空間排序
jmap -histo pid | head -n20

dump堆內存文件
jmap -dump:format=b,file=heap pid

3、可視化的堆內存分析工具:JVisualVM、MAT等

3. 排查指南

1、查看監控,以了解出現問題的時間點以及當前FGC的頻率(可對比正常情況看頻率是否正常)

2、了解該時間點之前有沒有程序上線、基礎組件升級等情況。

3、了解JVM的參數設置,包括:堆空間各個區域的大小設置,新生代和老年代分別採用了哪些垃圾收集器,然後分析JVM參數設置是否合理。

4、再對步驟1中列出的可能原因做排除法,其中元空間被打滿、內存泄漏、代碼顯式調用gc方法比較容易排查。

5、針對大對象或者長生命周期對象導致的FGC,可通過 jmap -histo 命令並結合dump堆內存文件作進一步分析,需要先定位到可疑對象。

6、通過可疑對象定位到具體代碼再次分析,這時候要結合GC原理和JVM參數設置,弄清楚可疑對象是否滿足了進入到老年代的條件才能下結論。

04 最後的話

這篇文章通過線上案例並結合GC原理詳細介紹了FGC的排查過程,同時給出了一份實踐指南。

後續會以類似的方式,再分享一個YGC耗時過長的案例,希望能幫助大家吃透GC問題排查,如果覺得本文對你有幫助,請大家關注我的個人公眾號!

– End –

作者簡介:程序員,985碩士,前亞馬遜Java工程師,現58轉轉技術總監。持續分享技術和管理方向的文章。如果感興趣,可微信掃描下面的二維碼關注我的公眾號:『IT人的職場進階』

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

【其他文章推薦】

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

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

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

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

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

※教你寫出一流的銷售文案?

3dTiles 數據規範詳解[1] 介紹

版權:轉載請帶原地址。https://www.cnblogs.com/onsummer/p/12799366.html @秋意正寒

Web中的三維

html5和webgl技術使得瀏覽器三維變成了可能。

巧婦難為無米之炊,三維數據(三維模型)是三維可視化重要的一環,事實上就是:三維數據眾多,行業跨界廣。

參考資料:http://www.bgteach.com/article/132

three.js的各種加載器實現了大部分通用三維格式的加載,屏蔽了格式不同的數據結構差異。

然而,這樣還是不能滿足日益增長的效果需求,比如場景一大,模型文件體積變大,解析所耗費的時間越來越長。

webgl,包括所有gpu有關的圖形渲染編程,幾乎只認這樣的三維數據:頂點、頂點顏色、頂點法線、着色語言…

所以,三維圖形界的通用格式:glTF應運而生,它面向終點,它按照圖形編程所需的格式來存儲數據,藉以二進制編碼提高傳輸速度。

它不再使用面向對象的思維存儲三維模型、貼圖紋理,而是按顯卡的思維存儲,存的是頂點、法線、頂點顏色等最基礎的信息,只不過組織結構上進行了精心的設計。

它面向終點,就意味着可編輯性差,因為渲染性能的提高犧牲了可編輯性,它不再像3ds、dae甚至是max、skp一樣容易編輯和轉換。

事實上,大多數三維軟件提供了glTF格式的轉換,或多一步,或一步到位。

地理真三維

早年,地理的三維還處於地形三維上,即数字高程模型(DEM)提供地表的高度拉伸。柵格高程數據、等高線、不規則三角網等均是数字高程模型的具體案例。
下圖是不規則三角網,也即所謂的三角面片(圖形渲染中很常見):

隨着學科的融合、計算機技術和硬件的更新換代,使得有模型、有細節的真三維融入到GIS中成為了可能,或者說,計算機技術和硬件的升級,給GIS以更廣闊的視角觀察世界。

cesium.js 號稱是 webgl 封裝的三維地理庫,是支持 gltf 模型的加載的。

面對大規模精細三維數據的加載,還要照顧到GIS的各種坐標系統、分析計算,gltf這種單個模型的方案顯得力不從心。

2016年,Cesium 團隊借鑒傳統2DGIS的地圖規範——WMTS,借鑒圖形學中的層次細節模型,打造出大規模的三維數據標準—— 3d-Tiles,中文譯名:三維瓦片。

它在模型上利用了 gltf 渲染快的特點,對大規模的三維數據進行組織,包括層次細節模型、模型的屬性數據、模型的層級數據等。

3dTiles的設計思想

3dTiles 繼承了 gltf 的優點:貼合圖形渲染 API 的邏輯,討 GPU 喜愛,webgl 對其內部組織起來的三維模型數據,不需要轉換,可以直接渲染(glTF 的功勞)。

關於 glTF 是如何嵌入到 3dTiles 中的,開篇不談,後續精講。

我們區分一組概念:規範和實現。

3dTiles 是一種規範,在規範的指導下,各種資源文件可以是獨立存在於硬盤中的目錄、文件,也可以以二進制形式寫入數據庫中。目前,3dTiles 的官方實現只有 “散列文件”,也就是文件、文件夾的形式存儲在硬盤中,有關如何存儲到數據庫中的討論,官方仍在進行中(截至發博客)。

glTF 也是一種規範,它的數據文件不一定就是後綴名為 .gltf 的文件,也不一定只有一個文件(glTF 的文件還可以是二進制文件、紋理貼圖文件等,扯遠了哈)。
在本文,會嚴格指明是數據還是數據標準,如果我說的是 “XXX文件(例如 Bird.glb 文件)” ,那就是在指特定的文件。

3dTiles還有一個特點:那就是不記錄模型數據,只記錄各級“Tile”的邏輯關係,以及“Tile”自己的屬性信息。所謂的模型數據,是指三維模型的頂點、貼圖材質、法線、顏色等信息。邏輯關係是指,各級Tile是如何在空間中保持連續的,LOD是如何組織的。屬性信息就很簡單啦,門有門的生產商,窗戶有窗戶的使用年限等,往大了說,建築還有它自己的壽命、法人、施工單位等屬性信息。

3dTiles的特點總結如下:

  • 三維模型使用了 glTF 規範,繼承它的渲染高性能
  • 除了嵌入的 glTF,3dTiles 自己 只記錄各級Tile的空間邏輯關係(如何構成整個3dtiles)和屬性信息,以及模型與屬性如何掛接在一起的信息

我覺得你還是雲里霧裡的,下一節將展示3dTiles具體數據,說說3dTiles的組織結構,說說3dTiles中的”Tile”,也就是“三維瓦片數據”中的“瓦片”是什麼。

3dTiles系列博客最終目錄:

01 引入與博客目錄:3dTiles 數據規範詳解

02 Tileset與Tile

03 內嵌在瓦片文件中的兩大數據表

04.1 B3dm 類型

04.2 I3dm 類型

04.3 Pnts 類型

04.4 Cmpt 類型

04.5 未發布的瓦片規範

05 3dTiles強大的擴展能力

06 優缺點

07 與I3S比較

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

【其他文章推薦】

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

網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※推薦評價好的iphone維修中心

併發系列(一)——線程池源碼(ThreadPoolExecutor類)簡析

前言

  本文主要是結合源碼去線程池執行任務的過程,基於JDK 11,整個過程基本與JDK 8相同。

  個人水平有限,文中若有表達有誤的,歡迎大夥留言指出,謝謝了!

一、線程池簡介

  1.1 使用線程池的優點

    1)通過復用已創建的線程,降低資源的消耗(線程的創建/銷毀是要消耗資源的)、提高響應速度;

    2)管理線程的個數,線程的個數在初始化線程池的時候指定;

    3)統一管理線程,比如停止,stop()方法;

  1.2 線程池執行任務過程

    線程池執行任務的過程如下圖所示,主要分為以下4步,其中參數的含義會在後面詳細講解:

    1)判斷工作的線程是否小於核心線程數據(workerCountOf(c) < corePoolSize),若小於則會新建一個線程去執行任務,這一步僅僅的是根據線程個數決定;

    2)若核心線程池滿了,就會判斷線程池的狀態,若是running狀態,則嘗試加入任務隊列,若加入成功后還會做一些事情,後面詳細說;

    3)若任務隊列滿了,則加入失敗,此時會判斷整個線程池線程是否滿,若沒有則創建非核心線程執行任務;

    4)若線程池滿了,則根據拒絕測試處理無法執行的任務;

    整體過程如下圖:

二、ThreadPoolExecutor類解析

  2.1 ThreadPoolExecutor的構造函數

    ThreadPoolExecutor類一共提供了4個構造函數,涉及5~7個參數,下面就5個必備參數的構造函數進行說明:

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

    1)corePoolSize :初始化核心線程池中線程個數的大小;

    2)maxmumPoolSize:線程池中線程大小;

    3)keepAliveTime:非核心線程的超時時長;

      非核心線程空閑時常大於該值就會被終止。

    4)unit :keepAliveTime的單位,類型可以參見TimeUnit類;

    5)BlockingQueue workQueue:阻塞隊列,維護等待執行的任務;

  2.2  私有類Worker

    在ThreadPoolExecutor類中有兩個集合類型比較重要,一個是用於放置等待任務的workQueue,其類型是阻塞對列;一個是用於用於存放工作線程的works,其是Set類型,其中存放的類型是Worker。

    進一步簡化線程池執行過程,可以理解為works中的工作線程不停的去阻塞對列中取任務,執行結束,線程重新加入大works中。

    為此,有必要簡單了解一下Work類型的組成。

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /** Thread this worker is running in.  Null if factory fails. */
        //工作線程,由線程的工廠類初始化
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;
        //不可重入的鎖
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        .......
    }

    Worker類繼承於隊列同步器(AbstractQueueSynchronizer),隊列同步器是採取鎖或其他同步組件的基礎框架,其主要結構是自旋獲取鎖的同步隊列和等待喚醒的等待隊列,其方法因此可以分為兩類:對state改變的方法 和 入、出隊列的方法,即獲取獲取鎖的資格的變化(可能描述的不準確)。關於隊列同步器後續博客會詳細分析,此處不展開討論。

    Work類中通過CAS設置狀態失敗后直接返回false,而不是判斷當前線程是否已獲取鎖來實現不可重入的鎖,源碼註釋中解釋這樣做的原因是因為避免work tash重新獲取到控制線程池全局的方法,如setCorePoolSize。

  2.3  拒絕策略類

    ThreadPoolExecutor的拒絕策略類是以私有類的方式實現的,有四種策略:

    1)AbortPolicy:丟棄任務並拋出RejectedExecutionException異常(默認拒絕處理策略)。

      2)DiscardPolicy:拋棄新來的任務,但是不拋出異常。

      3)DiscardOldestPolicy:拋棄等待隊列頭部(最舊的)的任務,然後重新嘗試執行程序(失敗則會重複此過程)。

      4)CallerRunsPolicy:由調用線程處理該任務。

    其代碼相對簡單,可以參考源碼。

三、任務執行過程分析

  3.1 execute(Runnable)方法

    execute(Runnable)方法的整體過程如上文1.2所述,其實現方式如下:

public void execute(Runnable command) {
        //執行的任務為空,直接拋出異常
        if (command == null)
            throw new NullPointerException();
        //ctl是ThreadPoolExecutor中很關鍵的一個AtomicInteger,主線程池的控制狀態
        int c = ctl.get();
        //1、判斷是否小於核心線程池的大小,若是則直接嘗試新建一個work線程
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //2、大於核心線程池的大小或新建work失敗(如創建thread失敗),會先判斷線程池是否是running狀態,若是則加入阻塞對列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //重新驗證線程池是否為running,若否,則嘗試從對列中刪除,成功后執行拒絕策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //若線程池的狀態為shutdown則,嘗試去執行完阻塞對列中的任務
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //3、新建非核心線程去執行任務,若失敗,則採取拒絕策略
        else if (!addWorker(command, false))
            reject(command);
    }

  3.2 addWorker(Runnable,boole)方法

    execute(Runnable)方法中,新建(非)核心線程執行任務主要是通過addWorker方法實現的,其執行過程如下:

private boolean addWorker(Runnable firstTask, boolean core) {
        //此處反覆檢查線程池的狀態以及工作線程是否超過給定的值
        retry:
        for (int c = ctl.get();;) {
            // Check if queue empty only if necessary.
            if (runStateAtLeast(c, SHUTDOWN)
                && (runStateAtLeast(c, STOP)
                    || firstTask != null
                    || workQueue.isEmpty()))
                return false;

            for (;;) {
            //核心和非核心線程的區別
                if (workerCountOf(c)
                    >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateAtLeast(c, SHUTDOWN))
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            //通過工廠方法初始化,可能失敗,即可能為null
            final Thread t = w.thread;
            if (t != null) {
            //獲取全局鎖
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int c = ctl.get();
                    //線程池處於running狀態
                    //或shutdown狀態但無需要執行的task,個人理解為用於去阻塞隊列中取任務執行
                    if (isRunning(c) ||
                        (runStateLessThan(c, STOP) && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    //執行任務,這裡會執行thread的firstTask獲取阻塞對列中取任務
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
            //開始失敗,則會從workers中刪除新建的work,work數量減1,嘗試關閉線程池,這些過程會獲取全局鎖
                addWorkerFailed(w);
        }
        return workerStarted;
    }

  3.3  runWorker(this) 方法

     在3.2 中當新建的worker線程加入在workers中成功后,就會啟動對應任務,其調用的是Worker類中的run()方法,即調用runWorker(this)方法,其過程如下:

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
        //while()循環中,前者是新建線程執行firstTask,對應線程個數小於核心線程和阻塞隊列滿的情況,
        //getTask()則是從阻塞對列中取任務執行
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                //僅線程池狀態為stop時,線程響應中斷,這裏也就解釋了調用shutdown時,正在工作的線程會繼續工作
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    try {
                    //執行任務
                        task.run();
                        afterExecute(task, null);
                    } catch (Throwable ex) {
                        afterExecute(task, ex);
                        throw ex;
                    }
                } finally {
                    task = null;
                    //完成的個數+1
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            //處理後續工作
            processWorkerExit(w, completedAbruptly);
        }
    }

   3.4 processWorkerExit(Worker,boole)方法

    當任務執行結果后,在滿足一定條件下會新增一個worker線程,代碼如下:

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            //對工作線程的增減需要加全局鎖
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }
        //嘗試終止線程池
        tryTerminate();

        int c = ctl.get();
        if (runStateLessThan(c, STOP)) {
        //線程不是中斷,會維持最小的個數
            if (!completedAbruptly) {
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            //執行完任務后,線程重新加入workers中
            addWorker(null, false);
        }
    }

  至此,線程池執行任務的過程分析結束,其他方法的實現過程可以參考源碼。

 

Ref:

[1]http://concurrent.redspider.group/article/03/12.html

[2]《Java併發編程的藝術》

 

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

【其他文章推薦】

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

台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨,RWD網頁設計為架站首選

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

※回頭車貨運收費標準

雀巢號召新創尖兵 加速開發乳製品替代品

摘錄自2020年9月29日中央社報導

瑞士食品業巨擘雀巢集團(Nestle)今(29日)發表聲明稿說:「公司擬將旗下位於瑞士科諾爾芬根(Konolfingen)的研發中心,開放給新創公司、學生和科學家。」,加速開發以植物為主的乳製品替代品。

雀巢表示,將會有內部、外部以及混合編組團隊在研發中心工作,為期六個月。

除了對永續乳製品進行測試外,集團也計畫鼓勵開發以植物為基礎的乳製品替代品。雀巢發表以此程序研發出來的一種使用蔬菜為基礎乳品。

氣候變遷
國際新聞
瑞士
乳製品
素食

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※教你寫出一流的銷售文案?

為疫苗研發犧牲? 保育團體憂:25萬尾鯊魚恐間接受害

摘錄自2020年9月29日自由時報報導

為盡速結束武漢肺炎(COVID-19)疫情所帶來的災難,各國無不卯足全力研發疫苗,然而近期有研發廠商為了讓疫苗的效果穩定,決定在疫苗內添加高效的醫學物質「鯊烯」(Squalene),然而鯊烯這種物質相當稀少,美國有鯊魚保育團體推估,光是製造提供給美國使用的疫苗量,恐就要殺死超過2.1萬條鯊魚,若是全世界範圍,「可能要殺害25萬條鯊魚」。

綜合外媒報導,「鯊烯」主要從鯊魚的肝油中提煉而出,在醫學與美容方面都是極為重要的高效素質,不僅可以滋潤皮膚,也可以提高、增強免疫力,加在疫苗中可使疫苗的效力提升,然而鯊魚肝油中提煉出的鯊烯十分稀少,平均一噸的鯊烯大約需要3000條鯊魚。

英國製藥大廠葛蘭素史克(GlaxoSmithKline)專門生產用於流感疫苗中的以鯊烯製成的佐劑,而葛蘭素史克曾表示,將在2021年生產10億個以鯊烯製成的「武漢肺炎疫苗用佐劑」,引起保育團體的憂慮;位在美國加州的鯊魚保育團體鯊魚同盟(Shark Allies)表示,如果全世界的人都會接種這種加入佐劑的疫苗,「那可能要殺害25萬條鯊魚」,如果再算上二次接種,數量還會翻倍到50萬條。

海洋
國際新聞
鯊魚

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※教你寫出一流的銷售文案?

加州野火每5秒燒1英畝 吞噬酒莊數萬人撤離

摘錄自2020年9月28日中央社報導

美國加州野火在強風助長下,每5秒鐘延燒約1英畝的土地,蔓延到世界知名的葡萄酒之鄉,納帕(Napa)與索諾馬(Sonoma)山谷今天有數以萬計的民眾被迫逃離家園。

根據美國國家海洋暨大氣總署(NOAA)衛星影像,昨天清晨約4時從納帕山谷爆發的「玻璃之火」(Glass Fire),昨晚延燒了2500英畝的土地,到了今早擴大到1萬1000英畝,相當於每5秒鐘燒掉約1英畝(約0.4公頃)。

法新社報導,加州森林防火廳(Cal Fire)說,加州野火把天空染成橘紅色,在悶熱的熱浪侵襲之下,火勢以「危險的速度」蔓延,且沒有一處獲得控制,沿途燒毀數座葡萄園與建築物。

官員說,當局已下令近3萬4000名居民疏散,並要求約1萬4000人準備立即撤離,因為「迅速蔓延的火勢」延燒到乾燥的植被以及難以進入的山區。

氣候變遷
國際新聞
美國
加州
野火
森林野火

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※推薦評價好的iphone維修中心

自己動手實現深度學習框架-8 RNN文本分類和文本生成模型

代碼倉庫: https://github.com/brandonlyg/cute-dl

目標

        上階段cute-dl已經可以構建基礎的RNN模型。但對文本相模型的支持不夠友好, 這個階段的目標是, 讓框架能夠友好地支持文本分類和本文生成任務。具體包括:

  1. 添加嵌入層, 為文本尋找高效的向量表示。
  2. 添加類別抽樣函數, 根據模型輸出的類別分佈抽樣得到生成的文本。
  3. 使用imdb-review數據集驗證文本分類模型。
  4. 使用一個古詩數據集驗證文本生成模型。

        這階段涉及到的代碼比較簡單因此接下來會重點描述RNN語言相關模型中涉及到的數學原理和工程方法。

數學原理

文本分類模型

        可以把文本看成是一個詞的序列\(W=[w_1, w_2, …, w_T]\), 在訓練數據集中每個文本屬於一個類別\(a_i\), \(a_i∈A\), 集合 \(A = \{ a_1, a_2, …, a_k \}\) 是一個類別別集合. 分類模型要做的是給定一個文本W, 計算所有類別的后驗概率:

\[P(a_i|W) = P(a_i|w_1,w_2,…,w_T), \quad i=1,2,…k \]

        那麼文本序列W的類別為:

\[a = arg \max_{a_i} P(a_i|w_1,w_2,…,w_T) \]

        即在給定文本的條件下, 具有最大后驗概率的類別就是文本序列W所屬的類別.

文本預測模型

        設任意一個文本序列為\(W=[w_1,w_2,…,W_T]\), 任意一個詞\(w_i ∈ V\), V是所有詞彙的集合,也叫詞彙表, 這裏需要強調的是\(w_i\)在V中是無序的, 但在W中是有序的, 文本預測的任務是, 計算任意一個詞\(w_i ∈ V\)在給定一個序列中的任意一個位置出現的概率:

\[P(w_1,…,W_T) = ∏_{t=1}^T P(w_t|w_1,…,w_{t-1}) \]

        文本預測輸出一個\(w_i ∈ V\)的分佈列, 根據這個分佈列從V中抽取一個詞即為預測結果。不同於分類任務,這裏不是取概率最大的詞, 這裏的預測結果是某個詞出現的在一個序列特定位置的個概率,只要概率不是0都有可能出現,所以要用抽樣的方法確定某次預測的結果。

詞的数字化表示

        任意一條數據在送入模型之前都要表示為一個数字化的向量, 文本數據也不例外。一個文本可以看成詞的序列,因此只要把詞数字化了,文本自然也就数字化了。對於詞來說,最簡單的方式是用詞在詞彙表中的唯一ID來表示, ID需要遵守兩個最基本的規則:

  1. 每個詞的ID在詞彙表中必須是唯一的.
  2. 每個詞的ID一旦確定不能變化.

        這種表示很難表達詞之間的關係, 例如: 在詞彙表中把”好”的ID指定為100, 如果希望ID能夠反映詞意的關係, 需要把”好”的近意詞: “善”, “美”, “良”, “可以”編碼為98, 99, 101, 102. 目前為止這看起還行. 如果還希望ID能夠反映詞之間的語法關係, “好”前後經常出現的詞: “友”, “人”, “的”, 這幾個詞的ID就很難選擇, 不論怎樣, 都會發現兩個詞它們在語義和語法上的關係都很遠,但ID卻很接近。這也說明了標量的表達能力很有限,無法表達多個維度的關係。為了能夠表達詞之間多個維度的的關係,多維向量是一個很好的選擇. 向量之間的夾大小衡量它們之間的關係:

\[cos(θ) = \frac{<A, B>}{|A||B|} \]

        對於兩個向量A, B使用它們的點積, 模的乘積就能得到夾角θ餘弦值。當cos(θ)->1表示兩個向量的相似度高, cos(θ)->0 表示兩個向量是不相關的, cos(θ)->-1 表示兩個向量是相反的。

        把詞的ID轉換成向量,最簡單的辦法是使用one-hot編碼, 這樣得到的向量有兩個問題:

  1. 任意兩個向量A,B, <A,B>=0, 夾角的餘弦值cos(θ)=0, 不能表達詞之間的關係.
  2. 向量的維度等於詞彙表的大小, 而且是稀疏向量,這和導致模型有大量的參數,模型訓練過程的運算量也很大.

        詞嵌入技術就是為解決詞表示的問題而提出的。詞嵌入把詞ID映射到一個合適維度的向量空間中, 在這個向量空間中為每個ID分配一個唯一的向量, 把這些向量當成參數看待, 在特定任務的模型中學習這些參數。當模型訓練完成后, 這些向量就是詞在這個特定任務中的一個合適的表示。詞嵌入向量的訓練步驟有:

  1. 收集訓練數據集中的詞彙, 構建詞彙表。
  2. 為詞彙表中的每個詞分配一個唯一的ID。假設詞彙表中的詞彙量是N, 詞ID的取值為:0,1,2,…,N-1, 對人任意一個0<ID<N-1, 必然存在ID-1, ID+1.
  3. 隨機初始化N個D維嵌入向量, 向量的索引為0,1,2,…,N-1. 這樣詞ID就成了向量的索引.
  4. 定義一個模型, 把嵌入向量作為模型的輸入層參与訓練.
  5. 訓練模型.

嵌入層實現

        代碼: cutedl/rnn_layers.py, Embedding類.

        初始化嵌入向量, 嵌入向量使用(-1, 1)區間均勻分佈的隨機變量初始化:

'''
dims 嵌入向量維數
vocabulary_size 詞彙表大小
need_train 是否需要訓練嵌入向量
'''
def __init__(self, dims, vocabulary_size, need_train=True):
    #初始化嵌入向量
    initializer = self.weight_initializers['uniform']
    self.__vecs = initializer((vocabulary_size, dims))

    super().__init__()

    self.__params = None
    if need_train:
        self.__params = []
        self.__cur_params = None
        self.__in_batch = None

        初始化層參數時把所有的嵌入向量變成參与訓練的參數:

def init_params(self):
    if self.__params is None:
        return

    voc_size, _ = self.__vecs.shape
    for i in range(voc_size):
        pname = 'weight_%d'%i
        p = LayerParam(self.name, pname, self.__vecs[i])
        self.__params.append(p)

        向前傳播時, 把形狀為(m, t)的數據轉換成(m, t, n)形狀的數據, 其中t是序列長度, n是嵌入向量的維數.

'''
in_batch shape=(m, T)
return shape (m, T, dims)
'''
def forward(self, in_batch, training):
    m,T = in_batch.shape
    outshape = (m, T, self.outshape[-1])
    out = np.zeros(outshape)

    #得到每個序列的嵌入向量表示
    for i in range(m):
        out[i] = self.__vecs[in_batch[i]]

    if training and self.__params is not None:
        self.__in_batch = in_batch

    return out

        反向傳播時只關注當前批次使用到的向量, 注意同一個向量可能被多次使用, 需要累加同一個嵌入向量的梯度.

def backward(self, gradient):
    if self.__params is None:
        return

    #pdb.set_trace()
    in_batch = self.__in_batch
    params = {}
    m, T, _ = gradient.shape
    for i in range(m):
        for t in range(T):
            grad = gradient[i, t]
            idx = self.__in_batch[i, t]

            #更新當前訓練批次的梯度
            if idx not in params:
                #當前批次第一次發現該嵌入向量
                params[idx] = self.__params[idx]
                params[idx].gradient = grad
            else:
                #累加當前批次梯度
                params[idx].gradient += grad

    self.__cur_params = list(params.values())

驗證

imdb-review數據集上的分類模型

        代碼: examples/rnn/text_classify.py.

        數據集下載地址: https://pan.baidu.com/s/13spS_Eac_j0uRvCVi7jaMw 密碼: ou26

數據集處理

        數據集處理時有幾個需要注意的地方:

  1. imdb-review數據集由長度不同的文本構成, 送入模型的數據形狀為(m, t, n), 至少要求一個批次中的數據具有相同的序列長度, 因此在對數據進行分批時, 對數據按批次填充.
  2. 一般使用0為填充編碼. 在構建詞彙表時, 假設有v個詞彙, 詞彙的編碼為1,2,…,v.
  3. 由於對文本進行分詞, 編碼比較耗時。可以把編碼后的數據保存起來,作為數據集的預處理數據, 下次直接加載使用。

模型

def fit_gru():
    print("fit gru")
    model = Model([
                rnn.Embedding(64, vocab_size+1),
                wrapper.Bidirectional(rnn.GRU(64), rnn.GRU(64)),
                nn.Filter(),
                nn.Dense(64),
                nn.Dropout(0.5),
                nn.Dense(1, activation='linear')
            ])
    model.assemble()
    fit('gru', model)

        訓練報告:

這個模型和tensorflow給出的模型略有差別, 少了一個RNN層wrapper.Bidirectional(rnn.GRU(32), rnn.GRU(32)), 這個模型經過16輪的訓練達到了tensorflow模型的水平.

文本生成模型

        我自己收集了一個古由詩詞構成的小型數據集, 用來驗證文本生成模型. 代碼: examples/rnn/text_gen.py.

        數據集下載地址: https://pan.baidu.com/s/14oY_wol0d9hE_9QK45IkzQ 密碼: 5f3c

        模型定義:

def fit_gru():
    vocab_size = vocab.size()
    print("vocab size: ", vocab_size)
    model = Model([
                rnn.Embedding(256, vocab_size),
                rnn.GRU(1024, stateful=True),
                nn.Dense(1024),
                nn.Dropout(0.5),
                nn.Dense(vocab_size, activation='linear')
            ])

    model.assemble()
    fit("gru", model)

        訓練報告:

        生成七言詩:

def gen_text():
    mpath = model_path+"gru"

    model = Model.load(mpath)
    print("loadding model finished")
    outshape = (4, 7)

    print("vocab size: ", vocab.size())

    def do_gen(txt):
        #編碼
        #pdb.set_trace()
        res = vocab.encode(sentence=txt)

        m, n = outshape

        for i in range(m*n - 1):
            in_batch = np.array(res).reshape((1, -1))
            preds = model.predict(in_batch)
            #取最後一維的預測結果
            preds = preds[:, -1]
            outs = dlmath.categories_sample(preds, 1)
            res.append(outs[0,0])

        #pdb.set_trace()
        txt = ""
        for i in range(m):
            txt = txt + ''.join(vocab.decode(res[i*n:(i+1)*n])) + "\n"

        return txt


    starts = ['雲', '故', '畫', '花']
    for txt in starts:
        model.reset()
        res = do_gen(txt)
        print(res)

        生成的文本:

雲填纜首月悠覺
纜濯醉二隱隱白
湖杖雨遮雙雨鄉
焉秣都滄楓寓功

故民民時都人把
陳雨積存手菜破
好纜簾二龍藕卻
趣晚城矣中村桐

畫和春覺上蓋騎
滿楚事勝便京兵
肯霆唇恨朔上楊
志月隨肯八焜著

花夜維他客陳月
客到夜狗和悲布
關欲摻似瓦闊靈
山商過牆灘幽惘

        是不是很像李商隱的風格?

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

【其他文章推薦】

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

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

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

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

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

※教你寫出一流的銷售文案?

RocketMQ系列(六)批量發送與過濾

今天我們再來看看RocketMQ的另外兩個小功能,消息的批量發送和過濾。這兩個小功能提升了我們使用RocketMQ的效率。

批量發送

以前我們發送消息的時候,都是一個一個的發送,這樣效率比較低下。能不能一次發送多個消息呢?當然是可以的,RocketMQ為我們提供了這樣的功能。但是它也有一些使用的條件:

  • 同一批發送的消息的Topic必須相同;
  • 同一批消息的waitStoreMsgOK 必須相同;
  • 批量發送的消息不支持延遲,就是上一節說的延遲消息;
  • 同一批次的消息,大小不能超過1MiB;

好了,只要我們滿足上面的這些限制,就可以使用批量發送了,我們來看看發送端的代碼吧,

@Test
public void producerBatch() throws Exception {

    List<Message> messages = new ArrayList<>();
    for (int i = 0;i<3;i++) {
        MessageExt message = new MessageExt();
        message.setTopic("cluster-topic");
        message.setKeys("key-"+i);
        message.setBody(("this is batchMQ,my NO is "+i+"---"+new Date()).getBytes());
        messages.add(message);
    }
    SendResult sendResult = defaultMQProducer.send(messages);
    System.out.println("sendResult:" + sendResult.getSendStatus().toString());
}
  • 其實批量發送很簡單,我們只是把消息放到一個List當中,然後統一的調用send方法發送就可以了。

再來看看消費端的代碼,

@Bean(initMethod = "start",destroyMethod = "shutdown")
public DefaultMQPushConsumer pushConsumer()  {
    try {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultMQPushConsumer");
        consumer.setNamesrvAddr("192.168.73.130:9876;192.168.73.131:9876;192.168.73.132:9876;");
        consumer.subscribe("cluster-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("msgs.size():"+msgs.size());
                if (msgs != null && msgs.size() > 0) {
                    for (MessageExt msg : msgs) {
                        System.out.println(new String(msg.getBody()));
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        return consumer;
    }catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}
  • 消費端的代碼沒有任何的變化,正常的接收消息就可以了,我們只是打印出了msgs.size(),看看一次接收一個消息,還是一次可以批量的接收多個消息。

我們啟動項目,批量發送一下,看看效果吧,

發送端的日誌如下:

sendResult:SEND_OK

發送成功,看來我們批量發送的3個消息都進入到了隊列中,再看看消費端,是一次消費一個,還是一次消費3個,如下:

msgs.size():1
this is batchMQ,my NO is 0---Mon Jun 15 09:31:04 CST 2020
msgs.size():1
this is batchMQ,my NO is 1---Mon Jun 15 09:31:04 CST 2020
msgs.size():1
this is batchMQ,my NO is 2---Mon Jun 15 09:31:04 CST 2020

看樣子是一次只消費了一個消息,那麼能不能一次消費3個消息呢?當然是可以的,不過要進行特殊的設置,

consumer.setConsumeMessageBatchMaxSize(5);

在消費端,我們設置批量消費消息的數量是5,這個值默認是1。我們再看看消費端的日誌,

msgs.size():3
this is batchMQ,my NO is 0---Mon Jun 15 09:35:47 CST 2020
this is batchMQ,my NO is 1---Mon Jun 15 09:35:47 CST 2020
this is batchMQ,my NO is 2---Mon Jun 15 09:35:47 CST 2020

這次一次消費了3個消息,如果消息比較多的話,最大一次能消費5個。這就是RocketMQ的批量發送和批量消費。

消息過濾

其實我們在大多數情況下,使用tag標籤就能夠很好的實現消息過濾。雖然tag標籤咱們並沒有過多的介紹,其實也很好理解,就是一個子Topic的概念,咱們在構建消息message的時候,message.setTags("xxx")。然後在消費的時候,訂閱Topic的時候,也可以指定訂閱的tag,

consumer.subscribe("cluster-topic", "*");

看到那個”*”了嗎?它就是訂閱的tag,”*”代表全部的tag,如果您想訂閱其中的一個或幾個,可以使用這種方式”tagA || tagB || tagC”,這是訂閱了cluster-topic下的3個tag,其他的tag是不會被消費的。

這裏我們所說的消息過濾比tag要高級很多,是可以支持sql的,怎麼樣?高級吧。比如:我們訂閱”a > 5 and b = ‘abc'”的消息,如下圖:

但是,RocketMQ畢竟不是數據庫,它只能支持一些基礎的SQL語句,並不是所有的SQL都支持,

  • 数字型的支持,>, >=, <, <=, BETWEEN, =

  • 字符串支持,=, <>, IN

  • IS NULL或者IS NOT NULL

  • 邏輯判斷,ANDORNOT

字段的類型也只是簡單的幾種,

  • 数字型,支持123,543.123,整型、浮點都可以;
  • 字符串,必須使用單引號”括起來;
  • 空值,NULL;
  • 布爾型,TRUE或者FALSE;

並且對消費者的類型也有一定的限制,只能使用push consumer才可以進行消息過濾。好了,說了這麼多了,我們看看怎麼使用吧,消費端和生產端都要進行相應的改造,先看看生產端吧,

@Test
public void producerBatch() throws Exception {

    List<Message> messages = new ArrayList<>();
    for (int i = 0;i<3;i++) {
        MessageExt message = new MessageExt();
        message.setTopic("cluster-topic");
        message.setKeys("key-"+i);
        message.setBody(("this is batchMQ,my NO is "+i+"---"+new Date()).getBytes());

        int a = i+4;
        message.putUserProperty("a",String.valueOf(a));

        messages.add(message);
    }
    SendResult sendResult = defaultMQProducer.send(messages);
    System.out.println("sendResult:" + sendResult.getSendStatus().toString());
}

我們在之前批量發送的基礎上進行了修改,定義了a的值,等於i+4,這樣循環3次,a的值就是4,5,6。然後調用message.putUserProperty("a",String.valueOf(a))注意,在使用消息過濾的時候,這些附加的條件屬性都是通過putUserProperty方法進行設置。這裏,我們設置了a的值。再看看消費端,

consumer.subscribe("cluster-topic", MessageSelector.bySql("a > 5"));

消費端,整體上沒有變化,只是在訂閱的方法中,使用MessageSelector.bySql("a > 5"),進行了條件的過濾。有的小夥伴可能會有疑問,我既想用sql過濾又想用tag過濾怎麼辦?當然也是可以,我們可以使用MessageSelector.bySql("a > 5").byTag("xx),byTag和bySql不分前後,怎麼樣,很強大吧。我們運行一下程序,看看效果吧。

我們啟動一下服務,報錯了,怎麼回事?錯誤信息如下:

The broker does not support consumer to filter message by SQL92

隊列不支持過濾消息,我們查詢了RocketMQ源碼中的BrokerConfig類,這個類就是對broker的一些設置,其中發現了這兩個屬性,

// whether do filter when retry.
private boolean filterSupportRetry = false;
private boolean enablePropertyFilter = false;
  • filterSupportRetry是在重試的時候,是否支持filter;
  • enablePropertyFilter,這個就是是否支持過濾消息的屬性;

我們把這兩個屬性在broker的配置文件改為true吧,如下:

filterSupportRetry=true
enablePropertyFilter=true

然後,再重新部署一下我們兩主兩從的集群環境。環境部署完以後,我們再重啟應用,沒有報錯。在生產端發送一下消息看看吧,

sendResult:SEND_OK

生產端發送消息沒有問題,說明3個消息都發送成功了。再看看消費端的日誌,

msgs.size():1
this is batchMQ,my NO is 2---Mon Jun 15 10:59:37 CST 2020

只消費了一個消息,並且這個消息中i的值是2,那麼a的值就是2+4=6,它是>5的,滿足SQL的條件,所以被消費掉了。這完全符合我們的預期。

總結

今天的兩個小功能還是比較有意思的,但裡邊也有需要注意的地方,

  • 消息的批量發送,只要我們滿足它的條件,然後使用List發送就可以了;批量消費,默認的消費個數是1,我們可以調整它的值,這樣就可以一次消費多個消息了;
  • 過濾消息中,最大的坑就是隊列的配置里,需要設置enablePropertyFilter=true,否則消費端在啟動的時候報不支持SQL的錯誤;

我們在使用的時候,多加留意就可以了,有問題,評論區留言吧~

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

【其他文章推薦】

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

網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※推薦評價好的iphone維修中心