一、簡介

　　pytest+allure+jenkins進行接口測試、生成測試報告、結合jenkins進行集成。

　　pytest是python的一種單元測試框架，與python自帶的unittest測試框架類似，但是比unittest框架使用起來更簡潔，效率更高

　　allure-pytest是python的一個第三方庫。用於連接pytest和allure，使它們可以配合在一起使用。

　　allure-pytest基於pytest的原始執行結果生成適用於allure的json格式結果。該json格式結果可以用於後續適用allure生成html結果。

二、安裝　　

　　1、安裝pytest，命令行或終端中輸入

1 pip install pytest

　　2、安裝allure-pytest，安裝成功

1 pip install allure-pytest

　　allure-pytest安裝成功后截圖如下。

　　3、下載安裝JDK

　　官方下載：https://www.oracle.com/java/technologies/javase-jdk11-downloads.html

　　安裝與配置不作闡述請諒解　　

　　4、下載安裝Jenkins

　　官方下載：https://www.jenkins.io/

　　安裝與配置不作闡述請諒解

三、下載Allure並配置

　　下載allure並配置

　　1、allure官網下載：https://github.com/allure-framework/allure2/releases

　　如下圖所示

　　2、allure2下載下來是一個zip的壓縮包，我們要解壓至自己的文件目錄下（可解壓放至項目的測試用例下或python安裝目錄下），自己可找到文件即可。

　　3、打開allure2目錄，找到bin目錄，複製bin文件目錄， 然後進行環境變量的配置，設置環境變量的目的就是讓系統無論在哪個目錄下都可以運行allure2。

　　4、環境變量設置：（桌面——我的電腦——右鍵屬性——高級系統配置——環境變量——系統變量——Path——編輯環境變量——把我們上面複製的目錄路徑新增至環境變量中即可）

　　設置環境變量，如下圖所示。

 

　　5、配置好后，打開cmd終端，輸入allure，出現以下幫助文檔，就說明配置成功了。

 四、Allure裝飾器描述

　　Allure裝飾器

 五、Pytest+Allure的應用

　　上述我們講了一些理論的知識，下面我們就來實戰練習一下吧。進一步理解Pytest+allure如何結合應用的。

　　1、新建testcase文件夾，用來存放測試用例，新建test_Demo.py文件，作為pytest的具體測試用例文件。在test_Demo.py文件中輸入以下代碼。

 1 # test_Demo.py
 2 # Creator：wuwei
 3 # Date：2020-06-09
 4 
 5 import pytest
 6 import requests
 7 import allure
 8 import sys
 9 sys.dont_write_bytecode = True
10 
11 @allure.epic('測試描述'.center(30, '*'))
12 @allure.feature('測試模塊')
13 @allure.suite('測試套件')
14 class TestPytestOne():
15     @allure.story('用戶故事描述：用例一')
16     @allure.title('測試標題：用例一')
17     @allure.description('測試用例描述：用例一')
18     @allure.testcase('測試用例地址:https://www.baidu.com/')
19     @allure.tag('測試用例標籤：用例一')
20     def test_one(self):
21         print('執行第一個用例')
22         assert 1 == 1
23 
24     @allure.story('用戶故事描述：用例二')
25     @allure.title('測試標題：用例二')
26     @allure.description('測試用例描述：用例二')
27     @allure.testcase('測試用例地址:https://www.sogou.com/')
28     @allure.tag('測試用例標籤：用例二')
29     def test_two(self,action):
30         print('執行第二個用例')
31         assert True == True
32 
33 # pytest運行
34 if __name__ == "__main__":
35     pytest.main(['-s', '-v', 'test_Demo.py', '-q', '--alluredir', '../reports'])

　　2、我們再來創建一個conftest.py，conftest用來共享數據及不同層次之間共享使用的文件，測試用例的前置和後置中一般都可以用到的。

 1 # conftest.py
 2 # Creator：wuwei
 3 # Date：2020-06-09
 4 
 5 import pytest
 6 import sys
 7 sys.dont_write_bytecode = True
 8 
 9 @pytest.fixture()
10 def action():
11     print("測試用例開始".center(30, '*'))
12     yield
13     print("測試用例結束".center(30, '*'))

　　3、運行test_Demo.py文件，test_Demo文件中已經pytest+allure的結合，可查看allure的運行結果，可看出在根目錄中生成了一個reports文件夾，其中生成了測試報告的json文件，這裏面的json文件可通過allure生成html的測試報告。
　　運行test_Demo.py，終端显示如下圖所示。

 　　生成的Json格式的測試報告，如下圖所示。

 　　4、使用allure將json文件生成html的測試報告，定位至項目文件根目錄下，運行以下命令，會在項目根目錄下生成一個名為allure_reports的文件夾，用來存放html測試報告。命令下如所示。

1 allure generate reports -o allure_reports/

　　成功運行allure，結果如下圖所示。

　　項目根目錄下的allure_reports文件，存放的是allure生成的測試報告。可看出文件下有一個HTML文件，可通過Python的編輯器Pycharm來打開該HTML文件（測試報告），或可通過allure命令來打開該HTML，展示HTML測試報告。如下所示。

　　測試報告文件，HTML測試報告如下。

　　allure命令打開HTML測試報告。命令如下所示。

1 allure open allure_reports/

　　如下圖所示。

 　　打開生成的HTML測試報告如下圖所示。

 六、Pytest+Allure+Jenkins的應用

　　1、Jenkins插件網站上下載allure插件最新版本：

　　　　http://mirrors.jenkins-ci.org/plugins/allure-jenkins-plugin/

　　2、Jenkins的安裝我已經在Postman+Newman+Git+Jenkins的篇章中講過了，沒看小夥伴可以看一下那篇文章。確認Jenkins服務是否開啟。確認開啟后，在瀏覽器中輸入：http://localhost:8080/，進入Jenkins配置頁面。

　　3、http://localhost:8080/，登錄Jenkins的頁面，在管理Jenkins——插件管理——高級中找到上傳插件。將（1）步驟中下載的.hpi的文件上傳至jenkins上。

　　上傳安裝好的allure-jenkins-plugin的插件，安裝完成並成功，是藍色圓點显示，因我已經安裝過一次，會提示已經安裝，重啟Jenkins即可生效。（注意：不是關閉瀏覽器重新打開，而是重啟Jenkins服務）

　　4、全局變量中配置allure路徑與JDK的路徑，

　　配置JDK安裝的路徑，如下圖所示。

　　配置allure安裝的路徑，如下圖所示。

　　5、新建Item，配置構建后的allure測試報告生成。這裏配置Pytest執行完成之後，生成的allure文件所在的目錄位置。

　　項目中生成allure的json測試報告的位置。需與下面構建后操作中的Results的Path文件一致。

　　構建后操作的allure生成測試報告的配置，如下圖所示

　　6、配置構建命令。就是上述在cmd中運行項目時的命令。如下圖所示。

注意：運行后發現有報錯。“Build step ‘Execute Windows batch command’ marked build as failure”，解決方案，在運行項目的命令后添加exit 0。如下圖所示。

　　7、修改運行命令后我們再來運行一下。我們可發現運行后，allure裏面沒任務數據。因為我們還沒設置運行的項目路徑。設置工作空間，打開工作空間目錄，將我們的項目複製到jenkins的工作目錄中。

　　我們可將代碼傳至GitHub上，在Jenkins中設置相關Github項目的配置，也可進行Jenkins部署。我在Postman+Newman+Git+Jenkins這篇博客里就應用到了。有興趣的可參考看看這篇Jenkins如何Git項目。在這裏我們使用本地項目來部署。

　　測試報告無數據因為工作空間裏面沒有項目配置。

　　複製項目至Jenkins工作空間的目錄中。

　　8、添加項目后，我們再運行一下，藍點則為運行成功，可看到後面已經生成了allure的測試報告了。可直接點擊後面的alluree圖標跳轉至HTML的測試報告。如下圖所示。

　　allure生成的HTML測試報告

八、總結

　　上述我們聊了下pytest+allure+jenkins如何結合集成一起使用的，本地啟動jenkins，運行項目，調用allure生成測試報告。也簡單的做了一個小Demo。後期我將結合Requests接口測試和seleniumWeb測試應用至具體項目中。

 

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

※回頭車貨運收費標準

以上車型都是適合年輕人的，即使是使用貸款的方式購買也不會造成太大的壓力，關鍵的一點是這些車型基本可以陪伴你度過初初成家、孩子長大的過程。

以上車型都是適合年輕人的，即使是使用貸款的方式購買也不會造成太大的壓力，關鍵的一點是這些車型基本可以陪伴你度過初初成家、孩子長大的過程。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

※回頭車貨運收費標準

最典型如座椅加熱/方向盤加熱和座椅通風。座椅加熱對於冬至還在穿短袖的廣州人民是一個雞肋的配置，通風功能則要實用得多。相反加熱功能對北方的小夥伴來說簡直是一個“雪中送炭”的配置。大家都知道如今霧霾已經成為威脅人類健康的頭等“殺手”，因此不少主機廠都會在優化車內空氣質量上花心思，比如裝配pM2。

要不是主編大人提醒，叫獸我都忘了昨天居然是冬至。冬至，意味着一年中最寒冷的節氣來臨。此時此刻，你在北方的霧霾里醉生夢死，我在南方的艷陽里穿着短袖···

每逢佳節倍思親，作為一隻“南漂單身狗”，每每到了過節的時候就會感到格外凄涼。而辦公室的另一頭，同事們還在為冬至吃湯圓還是水餃這個問題爭論的時候，叫獸我已經默默提筆，於是有了今天這篇文章。

到底湯圓還是水餃，說白了就是南北的習俗差異。只能說咱們祖國地大物博，不光是過節的習俗有很大不同，就連買車和用車都有很大差異。今天我就化思緒為動力，來和大家探討這個問題。

一．購車習慣差異

都說北方人粗獷大氣，不拘小節，選車習慣也非常符合他們的脾性。相比而言，他們往往對車內的做工和配置等方面不會過於在意，但發動機動力性能要強，車輛高速行駛時的穩定性要好。這裏以德系和美系車最具代表性。

南方人的消費觀念則比較精打細算，平日生活里追求“小而精”。如廣州茶餐廳里精緻的點心一樣，嘗過一回廣式早茶便能體會到其中的精髓。因此他們在選車時會對車輛的做工和油耗等方面比較在意；除此外，性價比也是他們比較看重的部分。這裏則以日系車為代表。

其實從國內幾大主機廠的分佈情況也能看出一二，北方多德系（一汽大眾、奧迪），南方則幾乎成了日系的“大本營”（廣汽旗下的本田、豐田、謳歌）。

二.配置需求不盡相同

這一點很好理解，由於地理環境的影響，南北方消費者對車輛配置方面的要求也有挺大差別。最典型如座椅加熱/方向盤加熱和座椅通風。座椅加熱對於冬至還在穿短袖的廣州人民是一個雞肋的配置，通風功能則要實用得多；相反加熱功能對北方的小夥伴來說簡直是一個“雪中送炭”的配置。

大家都知道如今霧霾已經成為威脅人類健康的頭等“殺手”，因此不少主機廠都會在優化車內空氣質量上花心思，比如裝配pM2.5凈化系統等配置。在這樣的氣候環境下，以往這個看起來毫不起眼的配置變得令不少消費者都重視了起來，北方朋友們則更情有獨鍾。

三．消費觀念不同

*提醒一點，這部分是叫獸以個人經驗得出看法，沒有實際數據作支撐，僅供參考。

同為國內相對富裕地區，廣東與江浙土豪們在汽車消費觀方面也有明顯差異。去過杭州、寧波等地朋友們一定會發現，那裡的超豪車尤其多，繁華地區幾乎隨處可見法拉利、蘭博基尼；相比下廣深兩地的超豪華車的數量則要“理性”得多。

叫獸斗膽以為，廣東人是先富起來的那批人，這裏擁有大量的實業經濟（東莞曾被成為世界工廠），除了政策方面的激勵，他們更多的是靠勤勞和汗水打出來的天下，財富是一點一滴積累起來的；而江浙滬地帶則多金融行業，財富來得有些“大起大落”，因此土豪們花起錢來也要大手大腳一些，畢竟坐一台勞斯萊斯、賓利去談生意可比坐奔馳要容易得多。

四．用車養車的區別

冬季北方氣溫低、雪量大，因此不少注重車輛保養和行車安全的車主都會在冬季來臨前給愛車做一次“大保健”，換粘度更低的機油和凝結點更低的純防凍液；還有最重要的一點，一套雪地胎是必須的。關乎安全，一點也不能放鬆。

夏季南方則是暴雨和暴晒交替進行，所以南方車主對車輛的漆面保養尤為注意，漆面鍍膜則是個不錯的辦法；另外要注意的是，雨季時務必將汽車停在安全地帶防止泡水。據保險公司的朋友介紹，由於天氣的多變性，已經有越來越多的車主會主動購買車輛自燃險和涉水險。

以上是叫獸總結出幾點南方和北方朋友們在選車和用車時的不同之處，如果大家還有更多的看法，歡迎大家留言與叫獸互動。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

※別再煩惱如何寫文案,掌握八大原則!

※網頁設計最專業,超強功能平台可客製化

而且1。3T發動機配合CVT變速箱，動力也平順。起步的動力不算強，不過挺平順的，像我這些開車不急的人，它的動力夠用了。我的遠景SUV行駛了有3200公里，平均油耗只有8。3L，這個表現讓人滿意。車主：彩色購買車型：1。3T CVT豪華型裸車價格：9。

吉利汽車-遠景SUV

指導價：7.49-10.19萬

車主：南通達人

購買車型：1.8L 手動尊貴型

裸車價格：8.49萬

最滿意的是它的做工質感很不錯，落鎖就自動關閉天窗這個功能很實用！空間夠用，座椅也舒適。動力方面，一擋、二擋起步很有推背感，手動變速箱的手感也不差。

就是發動機怠速時聲音有點大，還有音響的音質一般般，後排座椅不能完全放平，不過空間真的夠大。

目前我的車行駛了2410公里，平均的百公里油耗是8.17L，因為是SUV，所以不算高。

車主：北方的小強

購買車型：1.3T CVT旗艦型

裸車價格：10.19萬

遠景SUV空間大、底盤高。日常溫柔駕駛很舒適！內飾用了很多軟質材料，摸着舒服。而且1.3T發動機配合CVT變速箱，動力也平順。

起步的動力不算強，不過挺平順的，像我這些開車不急的人，它的動力夠用了！

我的遠景SUV行駛了有3200公里，平均油耗只有8.3L，這個表現讓人滿意。

車主：彩色

購買車型：1.3T CVT豪華型

裸車價格：9.49萬

空間大、價格不高、油耗不高！是我選擇它的原因。車子在150km/h內都比較穩，不發飄。轉向也比較精準，回饋力度也合適。

尾部設計有些不夠大氣，不過尾燈卻挺好看的。高速過彎的時候側傾會有一點明顯，不過這麼高的一輛SUV,當然以舒適性位為主。

現在我的車開了2900公里了，平時愛激烈駕駛，所以油耗一般要9.6L，比其他車主都要高！

編者點評：

遠景SUV的空間、用料、配置都表現不錯，加上一個不高的售價讓它性價比表現出色！而1.3T發動機在平順駕駛的情況下還是比較省油的。喜歡這款車型的話，快去4S店試駕一番吧。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

本文源碼：GitHub·點這裏 || GitEE·點這裏

一、Kafka集群環境

1、環境版本

版本：kafka2.11，zookeeper3.4

注意：這裏zookeeper3.4也是基於集群模式部署。

2、解壓重命名

tar -zxvf kafka_2.11-0.11.0.0.tgz
mv kafka_2.11-0.11.0.0 kafka2.11

創建日誌目錄

[root@en-master kafka2.11]# mkdir logs

注意：以上操作需要同步到集群下其他服務上。

3、添加環境變量

vim /etc/profile
export KAFKA_HOME=/opt/kafka2.11
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile

4、修改核心配置

[root@en-master /opt/kafka2.11/config]# vim server.properties
-- 核心修改如下
# 唯一編號
broker.id=0
# 開啟topic刪除
delete.topic.enable=true
# 日誌地址
log.dirs=/opt/kafka2.11/logs
# zk集群
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181

注意：broker.id安裝集群服務個數編排即可，集群下不能重複。

5、啟動kafka集群

# 啟動命令
[root@node02 kafka2.11]# bin/kafka-server-start.sh -daemon config/server.properties
# 停止命令
[root@node02 kafka2.11]# bin/kafka-server-stop.sh
# 進程查看
[root@node02 kafka2.11]# jps

注意：這裏默認啟動了zookeeper集群服務，並且集群下的kafka分別啟動。

6、基礎管理命令

創建topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--create --replication-factor 3 --partitions 1 --topic one-topic

參數說明：

• replication-factor 定義副本個數
• partitions 定義分區個數
• topic：定義topic名稱

查看topic列表

bin/kafka-topics.sh --zookeeper zk01:2181 --list

修改topic分區

bin/kafka-topics.sh --zookeeper zk01:2181 --alter --topic one-topic --partitions 5

查看topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--describe --topic one-topic

發送消息

bin/kafka-console-producer.sh \
--broker-list 192.168.72.133:9092 --topic one-topic

消費消息

bin/kafka-console-consumer.sh \
--bootstrap-server 192.168.72.133:9092 --from-beginning --topic one-topic

刪除topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--delete --topic first

7、Zk集群用處

Kafka集群中有一個broker會被選舉為Controller，Controller依賴Zookeeper環境，管理集群broker的上下線，所有topic的分區副本分配和leader選舉等工作。

二、消息攔截案例

1、攔截器簡介

Kafka中間件的Producer攔截器主要用於實現消息發送的自定義控制邏輯。用戶可以在消息發送前以及回調邏輯執行前有機會對消息做一些自定義，比如消息修改等，發送狀態監控等，用戶可以指定多個攔截器按順序執行攔截。

核心方法

• configure：獲取配置信息和初始化數據時調用；
• onSend：消息被序列化以及和計算分區前調用該方法，可以對消息做操作；
• onAcknowledgement：消息發送到Broker之後，或發送過程失敗時調用；
• close：關閉攔截器調用，執行一些資源清理工作；

注意：這裏說的攔截器是針對消息發送流程。

2、自定義攔截

定義方式：實現ProducerInterceptor接口即可。

攔截器一：在onSend方法中，對攔截的消息進行修改。

@Component
public class SendStartInterceptor implements ProducerInterceptor<String, String> {

    private final Logger LOGGER = LoggerFactory.getLogger("SendStartInterceptor");
    @Override
    public void configure(Map<String, ?> configs) {
        LOGGER.info("configs...");
    }
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 修改消息內容
        return new ProducerRecord<>(record.topic(), record.partition(),
                                    record.timestamp(), record.key(),
                              "onSend：{" + record.value()+"}");
    }
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        LOGGER.info("onAcknowledgement...");
    }
    @Override
    public void close() {
        LOGGER.info("SendStart close...");
    }
}

攔截器二：在onAcknowledgement方法中，判斷消息是否發送成功。

@Component
public class SendOverInterceptor implements ProducerInterceptor<String, String> {

    private final Logger LOGGER = LoggerFactory.getLogger("SendOverInterceptor");
    @Override
    public void configure(Map<String, ?> configs) {
        LOGGER.info("configs...");
    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        LOGGER.info("record...{}", record.value());
        return record ;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception != null){
            LOGGER.info("Send Fail...exe-msg",exception.getMessage());
        }
        LOGGER.info("Send success...");
    }

    @Override
    public void close() {
        LOGGER.info("SendOver close...");
    }
}

加載攔截器：基於一個KafkaProducer配置Bean，加入攔截器。

@Configuration
public class KafkaConfig {

    @Bean
    public Producer producer (){
        Properties props = new Properties();
        // 省略其他配置...
        // 添加攔截器
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.kafka.cluster.interceptor.SendStartInterceptor");
        interceptors.add("com.kafka.cluster.interceptor.SendOverInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        return new KafkaProducer<>(props) ;
    }
}

3、代碼案例

@RestController
public class SendMsgWeb {
    @Resource
    private KafkaProducer<String,String> producer ;
    @GetMapping("/sendMsg")
    public String sendMsg (){
        producer.send(new ProducerRecord<>("one-topic", "msgKey", "msgValue"));
        return "success" ;
    }
}

基於上述自定義Bean類型，進行消息發送，關注攔截器中打印日誌信息。

三、Kafka存儲分析

說明：該過程基於上述案例producer.send方法追蹤的源碼執行流程，源碼中的過程相對清楚，涉及的核心流程如下。

1、消息生成過程

Producer發送消息採用的是異步發送的方式,消息發送過程如下：

• Producer發送消息之後，經過攔截器，序列化，事務判斷；
• 流程執行后，消息內容放入容器中；
• 容器在指定時間內如果裝滿(size),會喚醒Sender線程;
• 容器如果在指定時間內沒有裝滿，也會執行一次Sender線程喚醒；
• 喚醒Sender線程之後，把容器數據拉取到topic中；

絮叨一句：讀這些中間件的源碼，不僅能開闊思維，也會讓自己意識到平時寫的代碼可能真的叫搬磚。

2、存儲機制

Kafka中消息是以topic進行標識分類，生產者面向topic生產消息，topic分區(partition)是物理上的存儲，基於消息日誌文件的方式。

• 每個partition對應於一個log文件,發送的消息不斷追加到該log文件末端；
• log文件中存儲的就是producer生產的消息數據，採用分片和索引機制；
• partition分為多個segment。每個segment對應兩個(.index)和(.log)文件；
• index文件類型存儲的索引信息；
• log文件存儲消息的數據；
• 索引文件中的元數據指向對應數據文件中message的物理偏移地址；
• 消費者組中的每個消費者，都會實時記錄消費的消息offset位置；
• 當然消息消費出錯時，恢復是從上次的記錄位置繼續消費；

3、事務控制機制

Kafka支持消息的事務控制

Producer事務

跨分區跨會話的事務原理，引入全局唯一的TransactionID，並將Producer獲得的PID和TransactionID綁定。Producer重啟后可以通過正在進行的TransactionID獲得原來的PID。
Kafka基於TransactionCoordinator組件管理Transaction，Producer通過和TransactionCoordinator交互獲得TransactionID對應的任務狀態。TransactionCoordinator將事務狀態寫入Kafka的內部Topic，即使整個服務重啟，進行中的事務狀態可以得到恢復。

Consumer事務

Consumer消息消費，事務的保證強度很低，無法保證消息被精確消費，因為同一事務的消息可能會出現重啟后已經被刪除的情況。

四、源代碼地址

GitHub·地址
https://github.com/cicadasmile/data-manage-parent
GitEE·地址
https://gitee.com/cicadasmile/data-manage-parent

推薦關聯閱讀：數據源管理系列

序號	標題
01	數據源管理：主從庫動態路由，AOP模式讀寫分離
02	數據源管理：基於JDBC模式，適配和管理動態數據源
03	數據源管理：動態權限校驗，表結構和數據遷移流程
04	數據源管理：關係型分庫分表，列式庫分佈式計算
05	數據源管理：PostGreSQL環境整合，JSON類型應用
06	數據源管理：基於DataX組件，同步數據和源碼分析
07	數據源管理：OLAP查詢引擎，ClickHouse集群化管理

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案

曹工說JDK源碼（1）–ConcurrentHashMap，擴容前大家同在一個哈希桶，為啥擴容后，你去新數組的高位，我只能去低位？

曹工說JDK源碼（2）–ConcurrentHashMap的多線程擴容，說白了，就是分段取任務

曹工說JDK源碼（3）–ConcurrentHashMap，Hash算法優化、位運算揭秘

什麼是緩存雪崩

基本概念梳理

這個基本也是redis 面試的經典題目了，然而，網上不少博客對這個詞的定義都含糊不清，各執一詞。

主要有兩類說法：

• 大量緩存key，由於設置了相同的過期時間，在某個時刻同時失效，導致此刻的查詢請求，全部湧向db，本來db的tps大概是幾千左右，結果湧入了幾十萬的請求，那db肯定直接就扛不住了

  這種說法下面，解決方案一般是，把過期時間增加一個隨機值，這樣，也就不會大批量的key同時失效了

• 另外一種說法是，本來redis扛下了大部分的請求，但是，由於緩存所在的機器，發生了宕機。此時，緩存這台機器之間就連不上了，redis服務也掛了，此時，你的服務里，發現redis取不到，然後全都跑去查數據庫，那，就發生和前面一樣的情況了，請求全部湧向db，db無響應。

兩類說法，也不用覺得，這個對，那個不對，不過是一個技術名詞，當初發明這個詞的人，估計也沒想那麼多，結果傳播開來之後，就變成了現在這個樣子。

我們這裏主要採用下面那一種說法，因為下面這種說法，其實是已經包含了上面的情景。但是，下面這種場景，要複雜的多，因為redis此時就是一個完全不可信的東西了，你得想好，怎麼不讓它掛掉，那是不是應該部署sentinel、cluster集群？同時，持久化必須要開啟。

這樣呢，掛掉后，短暫的不可用之後，大概幾十s吧，緩存集群就恢復了，就又可用了。

同時，我們還得考慮，假設，現在redis掛了，我們代碼的降級策略是什麼？

大家發現redis掛了，首先，估計是會拋異常了，連接超時；拋了異常后，要直接拋到前端嗎？作為一個穩健的後端程序，那肯定是不行的，你redis掛了，數據庫又沒掛；好吧，那我們就大家一起去查數據庫。

結果，大量的查詢請求，就烏泱泱地跑去查庫了，然後，db卒。這個肯定不行。

所以，我們必須要控制的一點是，當發現某個key失效了，不是大家都去查庫，而是要進行 併發控制。

什麼是併發控制？就是不能全部放過去查庫，只能放少部分，免得把脆弱的db打死。

併發控制，基本就是要爭奪去查庫的權利了，這一步，基本就是一個選舉的過程，可以通過搶鎖的方式，比如Reentrentlock，synchronized，cas也可以。

1. 搶到鎖的線程，有資格去查庫，其他線程要麼被阻塞，要麼自旋

2. 搶到鎖的線程，去查庫，查到數據后，將數據存放在某個地方，通知其他線程去取（如果其他線程被阻塞的話）；或者，如果其他線程沒被阻塞，比如sleep 50ms，再去指定的地方拿數據那種，這種就不需要通知

   總之，如果其他線程要我們通知，我們就通知；不要我們通知，我們就不通知。

搶到鎖的線程，在構建緩存時，其他線程應該干什麼？

1. 在while(true)里，sleep 50ms，然後再去取數據

   這種類似於忙等待，但是每次sleep一會，所以還不錯

2. 將自己阻塞，等待搶到鎖的線程，構建完緩存后，來喚醒

3. 在while(true)里，一直忙循環，期間一直檢查數據是否已經ok了，這種方案呢，要看裏面：檢查數據的操作，是否耗時；如果只是檢查jvm內存里的數據，那還好；否則的話，假設要去檢查redis的話，這種io比較耗時的操作的話，就不合適了，cpu會一直空轉。

本文採用的方案

主線程構建緩存時，其他線程，在while（true）里，sleep 一定時間，然後再檢查數據是否ready。

說了這麼多，好像和題目里的concurrenthashmap沒啥關係，不，是有關係的，因為，這個思路，其實就是來自於concurrentHashMap。

ConcurrentHashMap中，是怎麼去初始化底層數組的

在我們用無參構造函數，去new一個ConcurrentHashMap時，此時還不會去創建底層數組，這個是一個小優化。什麼時候創建數組呢，是在我們第一次去put的時候。

put的時候，會調用putVal。

其中，putVal代碼如下：

    transient volatile Node<K,V>[] table;

	final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        int binCount = 0;
      	// 1
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
          	// 2
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();

• 1處，把field table，賦值給局部變量tab

• 2處，如果tab為null，則進行initTable初始化

  這個2處，在多線程put的時候，是可能多個線程同時進來的。有併發問題。

我們接下來，看看initTable是怎麼解決這個問題的，畢竟，我們new數組，只new一次即可，new那麼多次，沒用，對性能有損耗。所以，這裏面肯定會多線程爭奪初始化權利的代碼。

	private transient volatile int sizeCtl;
	transient volatile Node<K,V>[] table;

	/**
     * Initializes table, using the size recorded in sizeCtl.
     */
    private final Node<K,V>[] initTable() {
        Node<K,V>[] tab;
      	int sc;
      	
      	// 0
        while ((tab = table) == null || tab.length == 0) {
          	// 1
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
          	// 2
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                  	// 3
                    if ((tab = table) == null || tab.length == 0) {
                      	// 4
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                  	// 5
                    sizeCtl = sc;
                }
                break;
              
            }// end if
          
        }// end while
        return tab;
    }

• 1處，這裏把sizeCtl，賦值給局部變量sc。這裏的sizeCtl是一個很重要的field，當我們new完之後，默認這個字段，要麼為0，要麼為準備創建的底層數組的長度。

  這裏去判斷是否小於0，那肯定不滿足，小於0，會是什麼意思？當某個線程，搶到了這個initTable中的底層數組的創建權利時，就會把sizeCtl改為 -1。

  所以，這裏的意思是，看看是否已經有其他線程在初始化了，如果已經有了，則直接調用：

  Thread.yield();

  這個方法的意思是，暗示操作系統，自己準備放棄cpu；但操作系統，自有它自己的線程調度規則，所以，這個方法可能沒什麼效果；我們業務代碼，這裏一般可以修改為Thread.sleep。

  這個方法調用完成后，後續也沒有其他代碼，所以會直接跳轉到循環開始處（0處代碼），判斷table是否初始化ok了，如果沒有ok，則會繼續進來。

• 2處，使用cas，如果此時，sizeCtl的值等於sc的值，就修改sizeCtl為 -1；如果成功，則返回true，進入3處

  否則，會跳轉到0處，繼續循環。

• 3處，雖然搶到了控制權，但是這裏還是要再判斷一下，不然可能出現重複初始化，即，不加這一行，4處的代碼，會被重複執行

• 4處開始，這裏去執行真正的初始化邏輯。

  // 
  int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  @SuppressWarnings("unchecked")
  // 1
  Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  // 2
  table = tab = nt;
  sc = n - (n >>> 2);
  

  這裏的1處，new數組；2處，賦值給field：table；此時，因為table 這個field是volatile修飾的，所以其他線程會馬上感知到。0處代碼就不會為true了，就不會繼續循環了。

• 5處，修改sizeCtl為正數。

這裏說下，為啥要加3處的那個判斷。

現在，假設線程A，在初始化完成后，走到了5處，修改了sizeCtl為正數；而線程B，剛好執行1處代碼：

// 1
if ((sc = sizeCtl) < 0)

那肯定，1處就不滿足了；然後就會進到2處，cas修改成功，進行初始化。沒有3處判斷的話，就會重複初始化。

基於concurrentHashmap，實現我們的緩存雪崩方案

我這裏的方案，還是比較簡單那種，就是，n個線程同時爭奪構建緩存的權利；winner線程，構建緩存后，會把緩存設置到redis；其他線程則是一直在while（true）里sleep一段時間，然後檢查redis里的數據是否不為空。

這個方案中，redis掛了這種情況，是沒在考慮中的，但是一個方案，沒辦法立馬各方面全部到位，後續我再完善一下。

不考慮緩存雪崩的代碼

@Override
public Users getUser(long userId) {
    ValueOperations<String, Users> ops = redisTemplate.opsForValue();
  	// 1
    Users s = ops.get(String.valueOf(userId));
    if (s == null) {
        /**
         * 2 這裏要去查庫獲取值
         */
        Users users = getUsersFromDB(userId);
		// 3
        ops.set(String.valueOf(users.getUserId()),users);

        return users;
    }

    return s;
}

private Users getUsersFromDB(long userId) {
    Users users = new Users();

    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("spent 1s to get user from db");
    users.setUserId(userId);
    users.setUserName("zhangsan");

    return users;
}

直接看上面的1,2,3處。就是檢查、構建緩存，設置到緩存的過程。

考慮緩存雪崩的代碼

	// 1
	private volatile int initControl;

	@Override
    public Users getUser(long userId) {
        ValueOperations<String, Users> ops = redisTemplate.opsForValue();

        Users users;
        while (true) {
          	// 2
            users = ops.get(String.valueOf(userId));
            if (users != null) {
              	// 3 
                break;
            }
			
          	// 4
            int initControlLocal = initControl;
            /**
             * 5 如果已經有線程在進行獲取了，則直接放棄cpu
             */
            if (initControlLocal < 0) {
//                log.info("initControlLocal < 0,just yield and wait");
//                Thread.yield();
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    log.warn("e:{}", e);
                }
                continue;
            }


            /**
             * 6 爭奪控制權
             */
            boolean bGotChanceToInit = U.compareAndSwapInt(this,
                    INIT_CONTROL, initControlLocal, -1);
          	// 7
            if (bGotChanceToInit) {
                try {
                  	// 8
                    users = ops.get(String.valueOf(userId));
                    if (users == null) {
                        log.info("got change to init");
                        /**
                         * 9 這裏要去查庫獲取值
                         */
                        users = getUsersFromDB(userId);
                        ops.set(String.valueOf(users.getUserId()), users);
                        log.info("init over");
                    }
                } finally {
                  	// 10
                    initControl = 0;
                }

                break;
            }// end if (bGotChanceToInit)
        }// end while


        return users;
    }

• 1處，定義了一個field，initControl；默認為0.線程們會去使用cas，修改為-1，成功的線程，即獲得初始化緩存的權利。

  注意，要定義為volatile，保證線程間的可見性

• 2處，去redis獲取緩存，如果不為null，直接返回

• 4處，如果沒取到緩存，則進入此處；此處，將field：initControl賦值給局部變量

• 5處，判斷局部變量initControlLocal，是否小於0；小於0，說明已經有線程在進行初始化了，直接contine，繼續下一次循環

• 6處，如果當前還沒有線程在初始化，則開始競爭初始化的權利，誰成功地用cas，修改field：initControl為-1，誰就獲得這個權利

• 7處，如果當前線程獲得了權利，則進入8處，否則，會繼續下一次循環

• 8處，再次去redis，獲取緩存，如果不為空，則進入9處

• 9處，查庫，設置緩存

• 10處，修改field：initControl為0，表示退出初始化

這裏的代碼，整體和hashmap中的initTable是一模一樣的。

如何測試

上面的方案，怎麼測試沒問題呢？我寫了一段測試代碼。

    ThreadPoolExecutor executor = new ThreadPoolExecutor(100, 100,
            60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new RejectedExecutionHandler() 	{
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            log.info("discard:{}",r);
        }
    });
	
	@RequestMapping("/test.do")
    public void test() {
      	// 0
        iUsersService.deleteUser(111L);

        CyclicBarrier barrier = new CyclicBarrier(100);

        for (int i = 0; i < 100; i++) {

            executor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        barrier.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    long start = System.currentTimeMillis();
                  	// 1
                    Users users = iUsersService.getUser(111L);
                    log.info("result:{},spent {} ms", users, System.currentTimeMillis() - start);
                }
            });
        }

    }

上面模擬100併發下，獲取緩存。

0處，把緩存刪了，模擬緩存失效

1處，調用方法，獲取緩存。

效果如下：

可以看到，只有一個線程拿到了初始化權利。

源碼位置

https://gitee.com/ckl111/all-simple-demo-in-work-1/tree/master/redis-cache-avalanche

總結

jdk的併發包，寫得真是有水平，大家仔細研究的話，必有收穫。

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

【其他文章推薦】

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

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

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

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

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

※超省錢租車方案