李嘉誠投資電動車 F-立凱受惠

香港首富李嘉誠所投資的香港五龍電動車日前與台灣F-立凱簽署交易契約,共同打造垂直整合的電動車廠。五龍電動車由李嘉誠所投資,未來將斥資新台幣20億元增資F-立凱,取得21.8%股權。

根據立凱、五龍電動車與五龍動力的協議,三方將以策略聯盟、資本合作的方式取得股權交換方案,五龍電動車與五龍動力將對立凱電共同投資新台幣20億元左右,取得F-立凱的21.8%股權。

F-立凱先前已獲潤泰集團總裁尹衍樑旗下相關企業投資持股逾12%,以生產電動車關鍵零組件磷酸鋰鐵正極材料為主要業務。

李嘉誠看好電動車關鍵零組件,從2010年起陸續購入五龍電動車股權,目前持股比例約為6.6%。本次五龍電動車將與F-立凱合作,等於李嘉誠加入台灣電動車業者投資行列。

同時,F-立凱將以每股港幣0.5元的價格認購五龍電動車集團所發行的430萬普通股與2.75億元港幣的無擔保可轉換公司債,總投資額預計將達港幣4.9億元,持股比率為4.2%。

整體交易完成後,李嘉誠在五龍電動車的時股比重將從6.6%降至6.3%。同時,五龍與F-立凱將著手打造全球化的一條龍供應鏈,並結合南韓SK集團技術與美國Smith物流車市場,於兩岸三地、國際市場形成明確分工。

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

go語言教程之淺談數組和切片的異同

Hello ,各位小夥伴大家好,我是小棧君,上次分享我們講到了Go語言關於項目工程結構的管理,本期的分享我們來講解一下關於go語言的數組和切片的概念、用法和區別。

在go語言的程序開發過程中,我們避免不了數組和切片。關於他們的用法和區別卻使得有的小夥伴感覺困惑。所以小棧君這裏也歸納和總結了關於數組和切片的乾貨幫助小夥伴進行理解。

數組的定義

數組是具有相同唯一類型的一組已編號且長度固定的數據項序列,這種類型可以是任意的原始類型例如整形、字符串或者自定義類型。

相對於去聲明 number0, number1, …, number99 的變量,使用數組形式 numbers[0], numbers[1] …, numbers[99] 更加方便且易於擴展。

數組元素可以通過索引(位置)來讀取(或者修改),索引從 0 開始,第一個元素索引為 0,第二個索引為 1,以此類推。

總體來講的話數組就是同一種類型的固定長度的序列。

在go語言中數組的定義是很簡單的。

如圖所示,我們定義了一個長度為2的數組,在數組定義的過程中,系統已經對這個數組進行了初始化並分配了空間。所以我們如果想要進行賦值可以通過數組名加上下標的方式進行賦值。但是值得注意的一點是我們並不能進行數組的超長處理。這一點有別於java的數組定義,java的定長數組添加值后如果對於超出的值會有自動擴容功能。

但是在go語言中並沒有方法來進行增刪改查值,只有通過下標的方式,所以我們如果進行了越界處理編譯都會進行報錯。所以才入門的小夥伴們需要注意一下哦。數組的下標在數組的合法範圍之外就會出發訪問越界,會有panic出現。所以小棧君也是通過了一個實例給大家說明一下,因為編譯可能會不通過,所以我們巧妙的避開編譯器的編譯進行數組的越界操作說明。

當然需要值得注意的一點是,數組的長度也是數組類型的一部分,因此var a [2]int 和 var b [3] int 是兩個不同的類型。

知識點來了,在go語言中的數組是屬於值類型傳遞,當我們傳遞一個數組到一個方法中,改變副本的值並不會修改到原本數組的值。所以得到的數組還是原來的樣子。

因此如果我們要對數組進行值的修改的話,就只有進行指針操作啦~。

切片的概念

在go語言中數組中長度不可以更改,所以在實際的應用環境中並不是非常實用,所以Go語言衍生出了一種靈活性強和功能更強大的內置類型,即為切片。

與上面所講的數組相比,切片的長度是不固定的,並且切片是可以進行擴容。切片對象非常小,是因為它是只有3個字段的數據結構:一個是指向底層數組的指針,一個是切片的長度,一個是切片的容量。這3個字段,就是Go語言操作底層數組的元數據,有了它們,我們就可以任意的操作切片了。

當然,切片作為數組的引用,所以切片屬於是引用類型,各位小夥伴可千萬要記住了哦。在切片的使用過程當中,我們可以通過遍曆數組的方式進行對於切片的遍歷,我們也可以內置方法len對數組或切片進行長度的計算。

當然我們也可以對切片的容量進行計算,之前有講過Go語言有豐富的內置庫提供給我們使用,所以我們也可以cap內置函數進行容量的計算。多個切片如果表示同一個數組的片段,它們可以共享數據;因此一個切片和相關數組的其他切片是共享存儲的,相反,不同的數組總是代表不同的存儲。數組實際上是切片的構建塊。

上面的例子小棧君分別用數組和切片進行了測試,我們可以看到數組的容量一旦確定后就無法進行更改,當我們的切片進行初始化,初始的容量是2,此時切片的容量和長度都是2,但是我通過內置的append方法進行了切片的增加。此時的切片的容量和長度都是4。此時我們並不能確定切片內置擴容的機制,但是隱約猜測是倍增。

言歸正傳,為了測試一下切片的擴容機制,所以小棧君又進行了切片的增加,此時,細心的小夥伴應該發現,這次小棧君一次性增加了兩個元素在一個append裏面,因為這是append方法是一個可變長度的傳值。這也是一個小知識點哦。

如果切片的底層數組,沒有足夠的容量時,就會新建一個底層數組,把原來數組的值複製到新底層數組裡,再追加新值,這時候就不會影響原來的底層數組了。

append目前的算法是:容量小於1000個時,總是成倍的增長,一旦容量超過1000個,增長因子設為1.25,也就是說每次會增加25%的容量。

之後我們發現切片的容量和長度發生了變化,如果說上次容量的擴張是4是我們猜測的倍數擴容方式,那麼這次我們就實錘了他的擴容機制就是倍增。而且在Go語言的容量和長度不一樣,所以我們也可以得出結論,就是在 0 <= len(arry) <= cap(slice)。

在我們聲明好切片后我們可以使用new或是make方法對切片進行初始化,當然小棧君也試着嘗試證明切片如果沒有進行初始化是會panic的。結果並沒有出現。因為如果slice沒有初始化,它僅僅相當於一個nil,長度和容量都為0,並不會panic。

小棧君也考慮到可能是因為沒有內置增加方法或是沒有報錯僅僅只是因為我後面利用對Carry數組的切割進行賦值的緣故。所以不甘心又做了一次嘗試,定義好相應的切片后直接使用append方法,結果如下:

我們同樣可以通過上述的例子了解到切片的下標是左閉右開區間,因為我們carry數組的內容如上圖所示, 我們最終得到的結果是IT乾貨棧,下標來講的話是屬於1。所以我們得到的結論和事實是一樣的。對於切片我們也有很多用法,如下圖所示:

下圖是Python中對於切片的操作,並且Python中的數組更為靈活,在後面我為大家分享Python教程的時候會詳細分享哦。

切片的初始化

對於切片的初始化我們可以make方法和new方法

new(T) 為每個新的類型T分配一片內存,初始化為 0 並且返回類型為*T的內存地址:這種方法 返回一個指向類型為 T,值為 0 的地址的指針,它適用於值類型如數組和結構體;它相當於 &T{}。

make(T) 返回一個類型為 T 的初始值,它只適用於3種內建的引用類型:切片、map 和 channel。

拷貝

因為在go語言的數組是屬於值傳遞,之前的方法也證實了這一點,在方法傳遞值的時候系統會進行拷貝一份副本進行傳遞,如果需要改變的值的話就需要使用指針。但是在使用切片處理的時候,因為切片屬於引用傳遞,所以go語言有內置的函數copy方法進行值的拷貝。

上述的一個例子可以綜合說明幾點問題了,最開始我們定義了一個切片並且容量是2,內容是1和2,我們同樣定義了切片b但是並沒有做初始化處理。直接使用copy操作。使用copy操作的時候,小棧君也複製了源碼出來,第一個是我們的數據源,第二個參數傳遞我們的目標源。直接使用的話我們可以從結果看出並沒有成功。

所以接下來小棧君又進行了初始化操作。這裏舉例的目的是提醒各位,在操作切片的時候沒有初始化就相當於nil,最好是進行切片的初始化操作。在早期go語言的版本中,沒有初始化切片會直接報錯。接下來我又進行了再一次的複製操作並且打印出他們的地址和容量、長度。可以看出進行切片的拷貝是不會進行切片的擴容處理。而且他們分別指向了不同的地址說明拷貝成功。

好了,今天的分享就到這啦,如果你喜歡我的分享,麻煩你點擊一個好看或贊,我是小棧君,不定期分享IT乾貨,包括但不限於區塊鏈、大數據、Python、go、等系列專題。希望與你共同成長。我們下期再見啦,拜了個拜~

本文由博客一文多發平台 發布!

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

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

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

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

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

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

域名解析 | A記錄 ,CNAME,MX,NS 你懂了嗎

域名解析 | A記錄 ,CNAME,MX,NS 你懂了嗎

域名解析

什麼是域名解析?域名解析就是國際域名或者國內域名以及中文域名等域名申請后做的到IP地址的轉換過程。IP地址是網路上標識您站點的数字地址,為了簡單好記,採用域名來代替ip地址標識站點地址。域名的解析工作由DNS服務器完成。

如下圖,百度 www.baidu.com 就是域名,解析出來的IP 14.216.177.30 這個過程就是域名解析。

A 記錄

A (Address) 記錄是用來指定主機名(或域名)對應的IP地址記錄。用戶可以將該域名下的網站服務器指向到自己的web server上。同時也可以設置您域名的二級域名。又稱IP指向,用戶可以在此設置子域名並指向到自己的目標主機地址上,從而實現通過域名找到服務器。
  說明:
  ·指向的目標主機地址類型只能使用IP地址;
  1) 添加A記錄
  在“主機名”中填入子域的名字,“類型”為A,“IP地址/主機名”中填入web服務器的IP地址,點擊“新增”按鈕即可。
  注:如果“主機名”一欄為空則表示對域名mydomain.com本身做指向。
  2) 修改A記錄
  只要在“IP地址/主機名”一欄中將原來的IP地址直接修改為新IP地址,然後點擊“修改”按鈕即可。
  3) 刪除A記錄
  點擊要刪除的A記錄後面對應的“刪除”按鈕即可。
  4) 泛域名解析
  即將該域名所有未指定的子域名都指向一個空間。
  在“主機名”中填入*,“類型”為A,“IP地址/主機名”中填入web服務器的IP地址,點擊“新增”按鈕即可。
  5) 負載均衡的實現:
  負載均衡(Server Load Balancing,SLB)是指在一系列資源上面動態地分佈網絡負載。負載均衡可以減少網絡擁塞,提高整體網絡性能,提高自愈性,並確保企業關鍵性應用的可用性。
  當相同子域名有多個目標地址時,表示輪循,可以達到負載均衡的目的,但需要虛擬主機服務商支持。

CNAME記錄

別名記錄。這種記錄允許您將多個名字映射到另外一個域名。通常用於同時提供WWW和MAIL服務的計算機。例如,有一台計算機名為“host.mydomain.com”(A記錄)。它同時提供WWW和MAIL服務,為了便於用戶訪問服務。可以為該計算機設置兩個別名(CNAME):WWW和MAIL。這兩個別名的全稱就 http://www.mydomain.com/和“mail.mydomain.com”。實際上他們都指向 “host.mydomain.com”。通常稱別名指向。您可以為一個主機設置別名。比如設置test.mydomain.com,用來指向一個主機www.rddns.com那麼以後就可以用test.mydomain.com來代替訪問www.rddns.com了。
  說明:
  ·CNAME的目標主機地址只能使用主機名,不能使用IP地址;
  ·主機名前不能有任何其他前綴,如:http://等是不被允許的;
  ·A記錄優先於CNAME記錄。即如果一個主機地址同時存在A記錄和CNAME記錄,則CNAME記錄不生效。
  1) 添加CNAME記錄
  在“主機名”中填入子域的名字,“類型”為CNAME,“IP地址/主機名”中填入目標地址,點擊“新增”按鈕即可。
  2) 修改CNAME記錄
  只要在“IP地址/主機名”一欄中將原來的目標地址直接修改為新地址,然後點擊“修改”按鈕即可。
  3) 刪除CNAME記錄
  點擊要刪除的CNAME記錄後面對應的“刪除”按鈕即可。

MX 記錄

郵件交換記錄。用於將以該域名為結尾的电子郵件指向對應的郵件服務器以進行處理。如:用戶所用的郵件是以域名mydomain.com為結尾的,則需要在管理界面中添加該域名的MX記錄來處理所有以@mydomain.com結尾的郵件。
  說明:
  ·MX記錄可以使用主機名或IP地址;
  ·MX記錄可以通過設置優先級實現主輔服務器設置,“優先級”中的数字越小表示級別越高。也可以使用相同優先級達到負載均衡的目的;
  ·如果在“主機名”中填入子域名則此MX記錄只對該子域名生效。
  1)添加MX記錄
  “主機名”一欄為空,“類型”為MX,“IP地址/主機名”中填入目標地址或IP,“優先級”里填入数字(必須為整數),點擊“新增”按鈕即可。
  2) 修改MX記錄
  只要在“IP地址/主機名”一欄中將原來的目標地址或IP直接修改為新地址或IP,然後點擊“修改”按鈕即可。
  3) 刪除MX記錄
  點擊要刪除的MX記錄後面對應的“刪除”按鈕即可。
  4) 負載均衡
  服務器負載均衡(Server Load Balancing,SLB)是指在一系列資源上面智能地分佈網絡負載。負載均衡可以減少網絡擁塞,提高整體網絡性能,提高自愈性,並確保企業關鍵性應用的可用性。
  當域名的MX記錄有多個目標地址且優先級相同時,表示輪循,可以達到負載均衡的目的,但需要郵箱服務商支持。

NS 記錄

解析服務器記錄。用來表明由哪台服務器對該域名進行解析。這裏的NS記錄只對子域名生效。例如用戶希望由12.34.56.78這台服務器解析news.mydomain.com,則需要設置news.mydomain.com的NS記錄。
  說明:
  ·“優先級”中的数字越小表示級別越高;
   ·“IP地址/主機名”中既可以填寫IP地址,也可以填寫像ns.mydomain.com這樣的主機地址,但必須保證該主機地址有效。如,將 news.mydomain.com的NS記錄指向到ns.mydomain.com,在設置NS記錄的同時還需要設置ns.mydomain.com的 指向,否則NS記錄將無法正常解析;
  ·NS記錄優先於A記錄。即,如果一個主機地址同時存在NS記錄和A記錄,則A記錄不生效。這裏的NS記錄只對子域名生效。
  1) 添加NS記錄
  在“主機名”中填入子域的名字,“類型”為NS,“IP地址/主機名”中填入解析服務器的IP地址或名稱(如ns.mydomain.com),點擊“新增”按鈕即可。
  2) 修改NS記錄
  只要在“IP地址/主機名”一欄中將原來的服務器目標地址或IP直接修改為新地址或IP,然後點擊“修改”按鈕即可。
  3) 刪除NS記錄
  點擊要刪除的NS記錄後面對應的“刪除”按鈕即可。

使用A記錄和CNAME進行域名解析的區別

A記錄就是把一個域名解析到一個IP地址(Address,特製数字IP地址),而CNAME記錄就是把域名解析到另外一個域名。其功能是差不多,CNAME將幾個主機名指向一個別名,其實跟指向IP地址是一樣的,因為這個別名也要做一個A記錄的。但是使用CNAME記錄可以很方便地變更IP地址。如果一台服務器有100個網站,他們都做了別名,該台服務器變更IP時,只需要變更別名的A記錄就可以了。

使用A記錄和CNAME哪個好?

域名解析CNAME記錄A記錄哪一種比較好?如果論對網站的影響,就沒有多大區別。但是:CNAME有一個好處就是穩定,就好像一個IP與一個域名的區別。服務商從方便維護的角度,一般也建議用戶使用CNAME記錄綁定域名的。如果主機使用了雙線IP,顯然使用CNAME也要方便一些。

A記錄也有一些好處,例如可以在輸入域名時不用輸入WWW.來訪問網站哦!從SEO優化角度來看,一些搜索引擎如alex或一些搜索查詢工具網站等等則默認是自動去掉WWW.來辨別網站,CNAME記錄是必須有如:WWW(別名)前綴的域名,有時候會遇到這樣的麻煩,前綴去掉了默認網站無法訪問。

有人認為,在SEO優化網站的時候,由於搜索引擎找不到去掉WWW.的域名時,對網站權重也會有些影響。因為有些網民客戶也是不喜歡多寫三個W來訪問網站的,網站無法訪問有少量網民客戶會放棄繼續嘗試加WWW.訪問域名了,因此網站訪問瀏覽量也會減少一些。

也有人認為同一個域名加WWW.和不加WWW.訪問網站也會使網站權重分散,這也是個問題。但是可以使用301跳轉把不加WWW.跳轉到加WWW.的域名,問題就解決了

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

【其他文章推薦】

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

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

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

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

Appium+python自動化(三十九)-Appium自動化測試框架綜合實踐 – 代碼實現(超詳解)

簡介

  前邊一直在分享testng的相關文章,看了點贊量和閱讀數不是很高,宏哥猜測估計是大家確實是用不到或者不喜歡吧!不過宏哥經過一段時間的準備,appium的自動化測試框架完善的差不多了,那麼接下來宏哥繼續給小夥伴和童鞋們分享有關Appium自動化測試框架綜合實踐。想必小夥伴們有點等不及了吧!

driver配置封裝

kyb_caps.yaml 配置表

主要是一些配置信息的封裝。

 參考代碼

platformName: Android
#模擬器
platformVersion: 5.1.1
deviceName: 127.0.0.1:62025

#mx4真機
#platformVersion: 5.1
#udid: 750BBKL22GDN
#deviceName: MX4

appname: kaoyan3.1.0.apk
noReset: False
unicodeKeyboard: True
resetKeyboard: True

appPackage: com.tal.kaoyan
appActivity: com.tal.kaoyan.ui.activity.SplashActivity
ip: 127.0.0.1
port: 4723

desired_caps.py

主要是用來讀取配置文件的信息的封裝。

 參考代碼

# coding=utf-8
# 1.先設置編碼,utf-8可支持中英文,如上,一般放在第一行

# 2.註釋:包括記錄創建時間,創建人,項目名稱。
'''
Created on 2019-11-14
@author: 北京-宏哥   QQ交流群:707699217
Project:Appium自動化測試框架綜合實踐 - 代碼實現
'''
# 3.導入模塊
from appium import webdriver
import yaml
import logging
import logging.config
import os

CON_LOG='../config/log.conf'
logging.config.fileConfig(CON_LOG)
logging=logging.getLogger()

def appium_desired():
    with open('../config/kyb_caps.yaml','r',encoding='utf-8') as file:
        data=yaml.load(file)

    desired_caps={}
    desired_caps['platformName']=data['platformName']
    desired_caps['platformVersion']=data['platformVersion']
    desired_caps['deviceName']=data['deviceName']

    base_dir = os.path.dirname(os.path.dirname(__file__))
    app_path = os.path.join(base_dir, 'app', data['appname'])
    desired_caps['app']=app_path

    desired_caps['appPackage']=data['appPackage']
    desired_caps['appActivity']=data['appActivity']
    desired_caps['noReset']=data['noReset']


    desired_caps['unicodeKeyboard']=data['unicodeKeyboard']
    desired_caps['resetKeyboard']=data['resetKeyboard']

    logging.info('start app...')
    driver=webdriver.Remote('http://'+str(data['ip'])+':'+str(data['port'])+'/wd/hub',desired_caps)
    driver.implicitly_wait(8)
    return driver

if __name__ == '__main__':
    appium_desired()

    # with open('../config/kyb_caps.yaml', 'r', encoding='utf-8') as file:
    #     data = yaml.load(file)
    #
    # base_dir=os.path.dirname(os.path.dirname(__file__))
    # print(os.path.dirname(__file__))
    # print(base_dir)
    #
    # app_path=os.path.join(base_dir,'app',data['appname'])
    # print(app_path)

相對路徑符號含義

  1. “.”表示當前目錄
  2. “..” 表示當前目錄的上一級目錄。
  3. “./”表示當前目錄下的某個文件或文件夾,視後面跟着的名字而定
  4. “../”表示當前目錄上一級目錄的文件或文件夾,視後面跟着的名字而定。

基類封裝

baseView.py

主要是一些元素定位方法的封裝。

 參考代碼

# coding=utf-8
# 1.先設置編碼,utf-8可支持中英文,如上,一般放在第一行

# 2.註釋:包括記錄創建時間,創建人,項目名稱。
'''
Created on 2019-11-14
@author: 北京-宏哥   QQ交流群:707699217
Project:Appium自動化測試框架綜合實踐 - 代碼實現
'''
# 3.導入模塊
class BaseView(object):
    def __init__(self,driver):
        self.driver=driver

    def find_element(self,*loc):
        return self.driver.find_element(*loc)

    def find_elements(self,*loc):
        return self.driver.find_elements(*loc)

    def get_window_size(self):
        return self.driver.get_window_size()

    def swipe(self,start_x, start_y, end_x, end_y, duration):
        return self.driver.swipe(start_x, start_y, end_x, end_y, duration)

common公共模塊封裝

公共方法封裝 : common_fun.py

主要是一些公共方法的封裝。

參考代碼

# coding=utf-8
# 1.先設置編碼,utf-8可支持中英文,如上,一般放在第一行

# 2.註釋:包括記錄創建時間,創建人,項目名稱。
'''
Created on 2019-11-13
@author: 北京-宏哥   QQ交流群:707699217
Project:Appium自動化測試框架綜合實踐 - 代碼實現
'''
# 3.導入模塊
from kyb_testProject.baseView.baseView import BaseView
from kyb_testProject.common.desired_caps import appium_desired
from selenium.common.exceptions import NoSuchElementException
import logging
from selenium.webdriver.common.by import By
import time,os
import csv

class Common(BaseView):
    cancelBtn=(By.ID,'android:id/button2')
    skipBtn=(By.ID,'com.tal.kaoyan:id/tv_skip')
    wemedia_cacel=(By.ID,'com.tal.kaoyan:id/view_wemedia_cacel')


    def check_cancelBtn(self):
        logging.info('==========check_cancelBtn=========')
        try:
            cancelBtn = self.driver.find_element(*self.cancelBtn)
        except NoSuchElementException:
            logging.info('no cancelBtn')
        else:
            cancelBtn.click()

    def check_skipBtn(self):
        logging.info('=========check skipBtn=============')

        try:
            skipBtn = self.driver.find_element(*self.skipBtn)
        except NoSuchElementException:
            logging.info('no skipBtn')
        else:
            skipBtn.click()

    def get_size(self):
        x = self.driver.get_window_size()['width']
        y = self.driver.get_window_size()['height']
        return x, y

    def swipeLeft(self):
        logging.info('swipeLeft')
        l = self.get_size()
        x1 = int(l[0] * 0.9)
        y1 = int(l[1] * 0.5)
        x2 = int(l[0] * 0.1)
        self.swipe(x1, y1, x2, y1, 1000)

    def getTime(self):
        self.now=time.strftime("%Y-%m-%d %H_%M_%S")
        return self.now

    def getScreenShot(self,module):
        time=self.getTime()
        image_file=os.path.dirname(os.path.dirname(__file__))+'/screenshots/%s_%s.png' %(module,time)

        logging.info('get %s screenshot' %module)
        self.driver.get_screenshot_as_file(image_file)

    def check_market_ad(self):
        logging.info('====check_market_ad====')
        try:
            element=self.driver.find_element(*self.wemedia_cacel)
        except NoSuchElementException:
            pass
        else:
            logging.info('close market ad')
            element.click()

    def get_csv_data(self,csv_file,line):
        logging.info('=====get_csv_data======')
        with open(csv_file,'r',encoding='utf-8-sig') as file:
            reader=csv.reader(file)
            for index,row in enumerate(reader,1):
                if index==line:
                    return row

if __name__ == '__main__':
    # driver=appium_desired()
    # com=Common(driver)
    # com.check_cancelBtn()
    # # com.check_skipBtn()
    # com.swipeLeft()
    # com.getScreenShot('startApp')

    list = ["這", "是", "一個", "測試", "數據"]
    # for i in range(len(list)):
        # print(i, list[i])

    list1 = ["這", "是", "一個", "測試", "數據"]
    # for index, item in enumerate(list1):
    #     print(index, item)

小結

     好了,今天的分享就到這裏。感謝您耐心的閱讀!

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

【其他文章推薦】

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

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

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

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

2015年特斯拉共銷售50580輛電動汽車 完成年度銷售計畫

據報導,特斯拉日前宣佈2015年銷售了50580輛電動汽車,完成年度銷售計畫。

特斯拉稱,2015年第四季度Model S電動汽車銷量為17192輛,比上年同期增長75%,環比增長48%。該公司第四季度生產507輛Model X電動汽車,銷售208輛。特斯拉目前計畫每週生產238輛Model X。

特斯拉完成年度銷售計畫相當重要。特斯拉一直在嘗試大幅提高Model S產量,並首次生產Model X。

這是一個頗為困難的任務。特斯拉最初計畫2015年銷售55000輛電動汽車。由於同時生產兩款車型存在不少困難,特斯拉去年8月把銷售目標下調至50000-55000輛,去年11月份再次把銷售目標下調至50000-52000輛。

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

工信部:暫停三元鋰電池客車入新能源汽車推薦目錄

1月24日,在2016中國電動汽車百人會“動力電池的發展與突破”主題峰會上,工信部裝備工業司司長張相木明確表示,出於對動力電池安全問題的考慮,目前暫停三元鋰電池客車列入新能源汽車推廣應用推薦車型目錄。

張相木表示,目前國家正在考慮進一步提高新能源汽車產品安全技術門檻,組織開展對三元鋰電池客車等車型在現行安全標準體系下的風險評估。在評估完成前,暫停三元鋰電池客車列入新能源汽車推廣應用推薦車型目錄。

張相木稱,動力蓄電池是新能源汽車的心臟,是新能源汽車產業發展的關鍵。2009年—2013年,我國新能源汽車動力蓄電池配套由0.3億瓦時逐年增加到7.9億瓦時,2014年提升到36.4億瓦時,2015年動力蓄電池單體的產能接近400億瓦時,據2020年生產200萬輛新能源汽車來測算,我國未來幾年動力蓄電池產品能夠滿足新能源汽車生產配套的要求。

目前,我國汽車動力蓄電池單體在安全性、迴圈壽命和能量密度等方面也取得了較大的進展。鋰離子電池單體能量密度達到了每公斤180瓦時以上,迴圈壽命超過2千次,單體成本降到了每瓦時2元以下,提前達到了規劃的目標。部分企業的材料、電解液、電池產品已進入國際電池及整車配套體系。
 

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

中國部分省市電動汽車充電服務費收費標準匯總

根據國家發改委下發的《關於電動汽車用電價格政策有關問題的通知》,充換電設施經營企業可向用戶收取電費和一定的充電服務費。政府希望通過這種方法吸引社會資本進入充電樁、充電站等設施的建設,甚至產生競爭,減少政府的投入。

以下是部分省市的充電服務收費標準:

北京

充電服務費按充電電量收取,每千瓦時收費上限標準為當日本市92號汽油每升最高零售價的15%。各經營單位可按照不超過上限標準,制定具體收費標準。

各充電設施經營單位應嚴格執行明碼標價規定,在經營場所的醒目位置標明相關的服務專案、服務內容、收費標準等事項。對擅自提高收費標準等價格違法行為,各級價格執法機構將依法予以查處。

上海

2020年前,電動汽車充電服務費執行政府指導價,暫定為每千瓦時不超過1.6元,試行一年,試行期滿後,結合《上海市鼓勵電動汽車充電設施發展暫行辦法》修訂並予以明確。今後將結合市場發展情況,逐步放開充電服務費,由市場競爭形成價格。

南京

由南京市人民政府辦公廳於2015年6月29日印發《南京市政府關於進一步支持新能源汽車推廣應用的若干意見》。

其中第11條,完善用電價格政策。落實新能源汽車充換電設施用電和充電服務價格政策。對向電網經營企業直接報裝接電的經營性集中式充電設施用電,執行大工業用電價格,暫免收基本電費;其他充電設施按其所在場所執行分類目錄電價。在國家、省相關政策出臺後,適時實施峰穀分時電價政策。

2020年前充換電服務費實行政府指導價管理,標準上限按不高於用油成本費用原則,具體標準由價格主管部門會同有關部門制定。目前南京的電動乘用車的充電收費標準為1.44元,公車收費為1.23元。

江西

2014年12月5日,由江西發展改革委員會下發的《江西省發展改革委關於電動汽車充換電服務費有關問題的通知》中明確了電動汽車充電的費用。

為便於充換電服務費計收和管理,提供充電服務模式的,按充電電度收取費用,包括電費和服務費用;提供換電服務模式的,按車輛行駛里程收取費用,包括電費、電池租賃和服務費用。

上限標準暫定為每千瓦時2.36元(含電費),其中電動公車(客車)上限標準暫定為每千瓦時1.36元(含電費)。

合肥

電動汽車充電價格由“電費服務費”組成,按電動汽車使用成本顯著低於燃油、低於燃氣汽車使用成本的原則核定。

其中對向電網經營企業直接報裝接電的經營性集中式充換電設施用電,執行大工業用電價格,2020年前,免收基本電費。其他充電設施按其所在場所執行分類目錄電價。電動汽車充換電設施用電執行峰穀分時電價政策。

直流快速充電樁充電服務費中准價格為0.90元/kwh,其中:集中向國家電網報裝並執行大工業電價可在上浮30%內協商確定,即最高1.17元/kwh,下降31%;

交流充電樁充電服務費在直流快速充電樁中准價格基礎上下浮30%,即0.63元/kwh,其中:利用電動汽車使用者單位場地等建設的充電樁,其充電服務費在不高於規定的交流充電樁充電服務費基礎上由雙方協商確定。

市物價局核定電動公車充電站充電服務費0.63元/kwh;充電電費執行大工業用電價格,免收基本電費;電動公車換電站換電服務價格按車輛行駛里程收取,含充電電費和充電服務費,換電服務價格為1.50元/公里。

青島

2015年5月15日,青島市物價局公佈了《山東省物價局轉發國家發展改革委關於電動汽車用電價格政策有關問題的通知》。

電動汽車充電服務費按“有傾斜、有優惠”的原則核定,確保電動汽車使用成本顯著低於燃油(或低於燃氣)汽車使用成本。電動汽車充電服務費按充電電度收取。其中,電動公車充電服務費最高為0.60元/千瓦時,電動乘用車充電服務費最高為0.65元/千瓦時,鼓勵充電設施建設運營單位在最高限價內給予用戶優惠。

河北

根據河北省物價局規定,河北省電動汽車充換電服務費標準實行政府指導價,並實行分類定價。類別暫劃分為純電動公車、七座及以下純電動乘用車、純電動環衛車。充換電服務費標準上限暫由省物價局制定,具體標準授權各設區市、擴權縣(市)價格主管部門按照“有傾斜、有優惠”,確保電動汽車使用成本顯著低於燃油(或燃氣)汽車使用成本原則確定,非經營性充換電設施,不得收取充換電服務費。

電動汽車充電服務費按充電電度收取,上限標準每千瓦時暫定為:純電動公車0.6元、七座及以下純電動乘用車和純電動環衛車1.6元。

電動汽車換電服務費按行駛里程收取,上限標準每公里暫定為:純電動公車1.8元、七座及以下純電動乘用車和純電動環衛車0.5元。

廣東

公車充電服務費0.8元/千瓦時,非公車充電服務費1.2元/千瓦時,充電時不再另收停車費用。

深圳

2015年12月28日,深圳市發改委發函稱“我市電動汽車的充電服務費標準繼續實行政府指導價管理,自2016年1月1日起,最高限價上調為每千瓦時1.00元(不區分車型)試行,允許下浮”。

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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使用指南

目錄

@(本節目錄)

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,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 接口的類主要有 ByteArraySerializer、StringSerializer、IntegerSerializer ,其中 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網頁設計已成為網頁設計推薦首選

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