單例模式，大家恐怕再熟悉不過了，其作用與實現方式有多種，這裏就不啰嗦了。但是，咱們在使用這些方式實現單例模式時，程序中就真的會只有一個實例嗎？

      聰明的你看到這樣的問話，一定猜到了答案是NO。這裏筆者就不賣關子了，開門見山吧！實際上，在有些場景下，如果程序處理不當，會無情地破壞掉單例模式，導致程序中出現多個實例對象。

      下面筆者介紹筆者已知的三種破壞單例模式的方式以及避免方法。

1、反射對單例模式的破壞

      我們先通過一個例子，來直觀感受一下

    （1）案例

  DCL實現的單例模式：

 1 public class Singleton{
 2     private static volatile Singleton mInstance;
 3     private Singleton(){}
 4     public static Singleton getInstance(){
 5         if(mInstance == null){
 6             synchronized (Singleton.class) {
 7                 if(mInstance == null){
 8                     mInstance = new Singleton();
 9                 }
10             }
11         }
12         return mInstance;
13     }
14 }

  測試代碼：

 1 public class SingletonDemo {
 2 
 3     public static void main(String[] args){
 4         Singleton singleton = Singleton.getInstance();
 5         try {
 6             Constructor<Singleton> constructor = Singleton.class.getDeclaredConstructor();
 7             constructor.setAccessible(true);
 8             Singleton reflectSingleton = constructor.newInstance();
 9             System.out.println(reflectSingleton == singleton);
10         } catch (Exception e) {
11             // TODO Auto-generated catch block
12             e.printStackTrace();
13         }
14     }
15 }

  執行結果：

false

      運行結果說明，採用反射的方式另闢蹊徑實例了該類，導致程序中會存在不止一個實例。

    （2）解決方案

      其思想就是採用一個全局變量，來標記是否已經實例化過了，如果已經實例化過了，第二次實例化的時候，拋出異常。實現代碼如下：

 1 public class Singleton{
 2     private static volatile Singleton mInstance;
 3     private static volatile boolean mIsInstantiated = false;
 4     private Singleton(){
 5         if (mIsInstantiated){
 6             throw new RuntimeException("Has been instantiated, can not do it again!");
 7         }
 8         mIsInstantiated = true;
 9     }
10     public static Singleton getInstance(){
11         if(mInstance == null){
12             synchronized (Singleton.class) {
13                 if(mInstance == null){
14                     mInstance = new Singleton();
15                 }
16             }
17         }
18         return mInstance;
19     }
20 }

執行結果：

 

     這種方式看起來比較暴力，運行時直接拋出異常。

 

2、clone()對單例模式的破壞 

       當需要實現單例的類允許clone()時，如果處理不當，也會導致程序中出現不止一個實例。

    （1）案例

  一個實現了Cloneable接口單例類：

 1 public class Singleton implements Cloneable{
 2     private static volatile Singleton mInstance;
 3     private Singleton(){
 4     }
 5     public static Singleton getInstance(){
 6         if(mInstance == null){
 7             synchronized (Singleton.class) {
 8                 if(mInstance == null){
 9                     mInstance = new Singleton();
10                 }
11             }
12         }
13         return mInstance;
14     }
15     @Override
16     protected Object clone() throws CloneNotSupportedException {
17         // TODO Auto-generated method stub
18         return super.clone();
19     }
20 }

  測試代碼：

 1 public class SingletonDemo {
 2 
 3     public static void main(String[] args){
 4         try {
 5             Singleton singleton = Singleton.getInstance();
 6             Singleton cloneSingleton;
 7             cloneSingleton = (Singleton) Singleton.getInstance().clone();
 8             System.out.println(cloneSingleton == singleton);
 9         } catch (CloneNotSupportedException e) {
10             e.printStackTrace();
11         }
12     }
13 }

執行結果：

false

  （2）解決方案：

     解決思想是，重寫clone()方法，調clone()時直接返回已經實例的對象

 1 public class Singleton implements Cloneable{
 2     private static volatile Singleton mInstance;
 3     private Singleton(){
 4     }
 5     public static Singleton getInstance(){
 6         if(mInstance == null){
 7             synchronized (Singleton.class) {
 8                 if(mInstance == null){
 9                     mInstance = new Singleton();
10                 }
11             }
12         }
13         return mInstance;
14     }
15     @Override
16     protected Object clone() throws CloneNotSupportedException {
17         return mInstance;
18     }
19 }

執行結果：

true

 

3、序列化對單例模式的破壞

   在使用序列化/反序列化時，也會出現產生新實例對象的情況。

  （1）案例

      一個實現了序列化接口的單例類：

 1 public class Singleton implements Serializable{
 2     private static volatile Singleton mInstance;
 3     private Singleton(){
 4     }
 5     public static Singleton getInstance(){
 6         if(mInstance == null){
 7             synchronized (Singleton.class) {
 8                 if(mInstance == null){
 9                     mInstance = new Singleton();
10                 }
11             }
12         }
13         return mInstance;
14     }
15 }

    測試代碼：

 1 public class SingletonDemo {
 2 
 3     public static void main(String[] args){
 4         try {
 5             Singleton singleton = Singleton.getInstance();
 6             FileOutputStream fos = new FileOutputStream("singleton.txt");
 7             ObjectOutputStream oos = new ObjectOutputStream(fos);
 8             oos.writeObject(singleton);
 9             oos.close();
10             fos.close();
11 
12             FileInputStream fis = new FileInputStream("singleton.txt");
13             ObjectInputStream ois = new ObjectInputStream(fis);
14             Singleton serializedSingleton = (Singleton) ois.readObject();
15             fis.close();
16             ois.close();
17             System.out.println(serializedSingleton==singleton);
18         } catch (Exception e) {
19             e.printStackTrace();
20         }
21 
22     }
23 }

     運行結果：

false

    （2）解決方案

    在反序列化時的回調方法 readResolve()中返回單例對象。

 1 public class Singleton implements Serializable{
 2     private static volatile Singleton mInstance;
 3     private Singleton(){
 4     }
 5     public static Singleton getInstance(){
 6         if(mInstance == null){
 7             synchronized (Singleton.class) {
 8                 if(mInstance == null){
 9                     mInstance = new Singleton();
10                 }
11             }
12         }
13         return mInstance;
14     }
15 
16     protected Object readResolve() throws ObjectStreamException{
17         return mInstance;
18     }
19 }

結果：

true

 

       以上就是筆者目前已知的三種可以破壞單例模式的場景以及對應的解決辦法，讀者如果知道還有其他的場景，記得一定要分享出來噢，正所謂“獨樂樂不如眾樂樂”！！！

       單例模式看起來是設計模式中最簡單的一個，但“麻雀雖小，五臟俱全”，其中有很多細節都是值得深究的。即便是本篇介紹的這幾個場景，也只是介紹了一些梗概而已，很多細節還需要讀者自己去試驗和推敲的，比如：通過枚舉方式實現單例模式，就不存在上述問題，而其它的實現方式似乎都存在上述問題！

 

       後記

       本篇參（剽）考（竊）了如下資料：

       高洪岩的《Java 多線程編程核心技術》

       博文：https://blog.csdn.net/fd2025/article/details/79711198

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

【其他文章推薦】

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

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

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

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

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

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

其實邁騰多出來的配置不止這些，還有二十多項，全部寫出來太長，故略去。低配的奧迪A4L採用了低功率的2。0T發動機，最大馬力是190匹，最大扭矩是320牛•米。頂配的邁騰使用的2。0T發動機賬面數據更好，220匹的最大馬力和350牛•米的最大扭矩均優於A4L。

隨着平台化，規模化生產的普及，很多大集團都會在自己的普通品牌和豪華品牌上通用零件，但是定價卻差了幾萬，那麼這幾萬的差價真的如大家所說，是LOGO的差異嗎？同一個集團，應該買TA的普通品牌高配車型還是豪華品牌的低配車型比較划算？

今天，我們來對比一下大眾邁騰的380TSI 旗艦版和奧迪A4L的40TFSI 進取版，看看哪個更值得買。

邁騰和奧迪A4L的外觀是同一個風格的兩份答卷。邁騰的橫貫式前臉把大燈連接起來，從視覺上來看顯得更寬更低矮。今天對比的邁騰380TSI 旗艦版配備的是18寸輪轂，所以邁騰的側面看上去更加大氣一些。

奧迪A4L則是在原有的六邊形格柵的家族設計元素上再下功夫，把大燈、格柵、車身線條設計得再銳利一些，所以奧迪A4L的外觀顯得更加年輕。不過這17寸的輪轂嘛，說實話不是很能襯托出A4L的檔次，建議後期更換。

這一局，邁騰略勝。

這一代的邁騰是歐版帕薩特直接拿來生產的，所以這一代邁騰的風格反而是繼承了上一代帕薩特的商務風。在邁騰上，木紋飾板貫穿整个中控台，空調出風口被設計得更高，所以空間更加寬闊。奧迪A4L的內飾以科技感為主要設計導向，奧迪A4L 30TSI 進取版採用中控台採用了銀色拉絲面板，中控屏採用懸浮式設計。可惜的是低配車沒有使用全液晶显示屏，空調面板也不是高配的那款金屬質感控制面板。所以質感稍顯不足。

這一局，也是邁騰勝出。

無論是邁騰還是奧迪A4L，引進國內后，都進行了不同程度的加長，邁騰的軸距為2871mm，比海外版增加了80mm，奧迪A4L的軸距是2908mm，比海外版增加了88mm。雖然奧迪A4L的軸距比邁騰要長一點，但是乘坐空間方面，邁騰比奧迪A4L要寬敞一點。這是因為邁騰是MQB（發動機橫置平台）的產物，而奧迪A4L是MLB（發動機縱置平台）的產物。發動機橫置的結構比較緊湊，占乘坐空間較少，所以邁騰的乘坐空間比奧迪A4L稍微大一點。

這一局，因為兩者的差距微乎其微，所以判定兩者打成平手。

在配置上，邁騰完勝沒什麼懸念。其實邁騰多出來的配置不止這些，還有二十多項，全部寫出來太長，故略去。

低配的奧迪A4L採用了低功率的2.0T發動機，最大馬力是190匹，最大扭矩是320牛•米。頂配的邁騰使用的2.0T發動機賬面數據更好，220匹的最大馬力和350牛•米的最大扭矩均優於A4L。從加速成績來看，邁騰比奧迪A4L更好，但也只是半秒的差距。奧迪A4L的發動機有AVS可變氣門升程技術和混合噴射技術。這兩項技術都可以提高燃油經濟性，所以從用戶反饋的數據來看，奧迪A4L比邁騰節油不少。

所以這局，也只能判五五開。

說到舒適性方面奧迪A4L的前後懸挂都使用了五連桿獨立懸挂，而邁騰則採用了前麥弗遜獨立懸挂后多連桿獨立懸挂的組合。這不僅是因為成本，還因為邁騰使用的MQB佔據了機艙空間，所以無法布局多連桿懸挂。麥弗遜式獨立懸挂有個缺點，那就是上下運動的時候對車輪外傾角變化明顯，而車輪外傾角變化會影響操縱的穩定性。所以，為了安全性考慮，麥弗遜懸挂一般都會比較硬。而使用五連桿的奧迪A4L可以把懸挂調得軟一些，所以奧迪A4L的濾震更好。在加上在隔音材料方面，豪華品牌比普通品牌的投入會更大。

所以舒適性方面，奧迪A4L完勝邁騰是沒什麼懸念的。

總結：縱觀兩車各個方面的對比，互有勝負。豪華品牌的車不是只掛着一個LOGO就賣這麼貴，如果你坐過邁騰后再坐A4L，你確確實實能夠感受到豪華車帶給你的那種舒適感，高級感。但是在靜態體驗的時候，邁騰的頂配版可以說是完虐奧迪A4L了。所以，我建議，如果你打算購買一台商務車，邁騰的頂配版更加合適。如果只是買來自用，奧迪A4L值得考慮。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

※南投搬家公司費用需注意的眉眉角角，別等搬了再說!

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

點評：放在以前，皇冠就是身份的象徵，也算是款足以媲美百萬級別車輛的車子，隨着諸多豪華品牌的入駐以及價格的降低，市場份額確實也受到了一定的衝擊。但秉承豪華穩重路線的18款皇冠在外觀和內飾上進行了年輕化，這一性格上的變化致使皇冠既穩住了老用戶也吸引了不少年輕的消費者。

在中國，汽車市場有個很奇妙的現象：有時候沒有競爭優勢的車型，卻成為銷量榜單上前幾名的常客；而有些產品競爭力很強的車型，卻好似不怎麼如意。但今天要為大家揭曉的是那些銷量看似不怎樣，在細分市場卻有不錯表現的特殊車型。

在進行分析點評之前，我們首先來看看本次分析的那些車子在今年前3個月的銷量指數。

點評：金牛座是福特品牌旗下的一款高端轎車車型，雖非屬豪華品牌但售價快30萬的它，素質可謂非常的高，大氣又低調的外觀配以變態級別的配置，以絕對性的優勢足以和豪華品牌扳手腕。但面對豪華品牌和同級車大眾的帕薩特和邁騰在20幾萬區間所帶來的巨大壓力，在今年1-3月的總銷量中金牛座還是頂住了壓力，賣出了4778輛。就性價比而言，金牛座還是很高的，就是售價有點偏高，如若售價能稍微降低一些，並配以一些購車優惠，相信在接下來的日子里，金牛的銷售表現會更令人滿意。

點評：放在以前，皇冠就是身份的象徵，也算是款足以媲美百萬級別車輛的車子，隨着諸多豪華品牌的入駐以及價格的降低，市場份額確實也受到了一定的衝擊。但秉承豪華穩重路線的18款皇冠在外觀和內飾上進行了年輕化，這一性格上的變化致使皇冠既穩住了老用戶也吸引了不少年輕的消費者。況且，皇冠本身就擁有不錯的綜合性價比和情懷感，並不會比BBA車型差，如果30萬讓消費者選擇購車，相信不少消費者都會選擇這些開起來更有格調又有品味的車子。

點評：非豪華品牌的輝昂和奧迪A7、A8等車型屬同平台打造，帶點奧迪影子，並帶有眾多賣點的輝昂在實力上絲毫也不比豪華品牌同級別車型遜色，在國內市場大眾也有不少專屬中國的車子，藉助大眾的品牌影響，輝昂這款車的知名度也有相應的提升。另外從市場行情來說，雖說中大型車的市場仍舊是豪華品牌車的天下，但是大街上千篇一律的豪華品牌車也給了輝昂一定的機遇，憑藉過硬的實力在這个中大型車的市場中還是分得一杯羹。

總結

金牛座、皇冠、輝昂三款車都很好，性價比也很不錯，在整個細分市場上表現還算是差強人意！其實吧，覺得金牛座這些怎麼也是比3系和C級車要高上一個等級，轉而有不少的消費者放棄BBA車型，選擇這些二線豪華品牌也不足為奇。畢竟與其選擇開着分分鐘與人撞車的BBA上街，不如選擇這些二線豪華品牌車，開起來更有格調和氣質。本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

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

摘要：Kafka中的位移是個極其重要的概念，因為數據一致性、準確性是一個很重要的語義，我們都不希望消息重複消費或者丟失。而位移就是控制消費進度的大佬。本文就詳細聊聊kafka消費位移的那些事，包括：

概念剖析

kafka的兩種位移

關於位移（Offset），其實在kafka的世界里有兩種位移：

• 分區位移：生產者向分區寫入消息，每條消息在分區中的位置信息由一個叫offset的數據來表徵。假設一個生產者向一個空分區寫入了 10 條消息，那麼這 10 條消息的位移依次是 0、1、…、9；

• 消費位移：消費者需要記錄消費進度，即消費到了哪個分區的哪個位置上，這是消費者位移（Consumer Offset）。

注意，這和上面所說的消息在分區上的位移完全不是一個概念。上面的“位移”表徵的是分區內的消息位置，它是不變的，即一旦消息被成功寫入到一個分區上，它的位移值就是固定的了。而消費者位移則不同，它可能是隨時變化的，畢竟它是消費者消費進度的指示器。

消費位移

消費位移，記錄的是 Consumer 要消費的下一條消息的位移，切記，是下一條消息的位移！ 而不是目前最新消費消息的位移

假設一個分區中有 10 條消息，位移分別是 0 到 9。某個 Consumer 應用已消費了 5 條消息，這就說明該 Consumer 消費了位移為 0 到 4 的 5 條消息，此時 Consumer 的位移是 5，指向了下一條消息的位移。

至於為什麼要有消費位移，很好理解，當 Consumer 發生故障重啟之後，就能夠從 Kafka 中讀取之前提交的位移值，然後從相應的位移處繼續消費，從而避免整個消費過程重來一遍。就好像書籤一樣，需要書籤你才可以快速找到你上次讀書的位置。

那麼了解了位移是什麼以及它的重要性，我們自然而然會有一個疑問，kafka是怎麼記錄、怎麼保存、怎麼管理位移的呢？

位移的提交

Consumer 需要上報自己的位移數據，這個彙報過程被稱為位移提交。因為 Consumer 能夠同時消費多個分區的數據，所以位移的提交實際上是在分區粒度上進行的，即Consumer 需要為分配給它的每個分區提交各自的位移數據。

鑒於位移提交甚至是位移管理對 Consumer 端的巨大影響，KafkaConsumer API提供了多種提交位移的方法，每一種都有各自的用途，這些都是本文將要談到的方案。

void commitSync(Duration timeout);
void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);
void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout);
void commitAsync();
void commitAsync(OffsetCommitCallback callback);
void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);

先粗略的總結一下。位移提交分為自動提交和手動提交；而手動提交又分為同步提交和異步提交。

自動提交

當消費配置enable.auto.commit=true的時候代表自動提交位移。

自動提交位移是發生在什麼時候呢？auto.commit.interval.ms默認值是50000ms。即kafka每隔5s會幫你自動提交一次位移。自動位移提交的動作是在 poll（）方法的邏輯里完成的，在每次真正向服務端發起拉取請求之前會檢查是否可以進行位移提交，如果可以，那麼就會提交上一次輪詢的位移。假如消費數據量特別大，可以設置的短一點。

越簡單的東西功能越不足，自動提交位移省事的同時肯定會帶來一些問題。自動提交帶來重複消費和消息丟失的問題：

• 重複消費： 在默認情況下，Consumer 每 5 秒自動提交一次位移。現在，我們假設提交位移之後的 3 秒發生了 Rebalance 操作。在 Rebalance 之後，所有 Consumer 從上一次提交的位移處繼續消費，但該位移已經是 3 秒前的位移數據了，故在 Rebalance 發生前 3 秒消費的所有數據都要重新再消費一次。雖然你能夠通過減少 auto.commit.interval.ms 的值來提高提交頻率，但這麼做只能縮小重複消費的時間窗口，不可能完全消除它。這是自動提交機制的一個缺陷。

• 消息丟失： 假設拉取了100條消息，正在處理第50條消息的時候，到達了自動提交窗口期，自動提交線程將拉取到的每個分區的最大消息位移進行提交，如果此時消費服務掛掉，消息並未處理結束，但卻提交了最大位移，下次重啟就從100條那消費，即發生了50-100條的消息丟失。

手動提交

當消費配置enable.auto.commit=false的時候代表手動提交位移。用戶必須在適當的時機（一般是處理完業務邏輯后），手動的調用相關api方法提交位移。比如在下面的案例中，我需要確認我的業務邏輯返回true之後再手動提交位移

 while (true) {
     try {
         ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMinutes(KafkaConfig.pollTimeoutOfMinutes));
         if (!consumerRecords.isEmpty()) {
             for (ConsumerRecord<String, String> record : consumerRecords) {
                 KafkaMessage kafkaMessage = JSON.parseObject(record.value(), KafkaMessage.class);
                 // 處理業務
                 boolean handleResult = handle(kafkaMessage);
                 if (handleResult) {
                     log.info(" handle success, kafkaMessage={}" ,kafkaMessage);
                 } else {
                     log.info(" handle failed, kafkaMessage={}" ,kafkaMessage);
                 }
             }
             // 手動提交offset
             consumer.commitSync(Duration.ofMinutes(KafkaConfig.pollTimeoutOfMinutes));
        
         } 
     } catch (Exception e) {
         log.info("kafka consume error." ,e);
     }
 }

手動提交明顯能解決消息丟失的問題，因為你是處理完業務邏輯后再提交的，假如此時消費服務掛掉，消息並未處理結束，那麼重啟的時候還會重新消費。

但是對於業務層面的失敗導致消息未消費成功，是無法處理的。因為業務層的邏輯千變萬化、比如格式不正確，你叫Kafka消費端程序怎麼去處理？應該要業務層面自己處理，記錄失敗日誌做好監控等。

但是手動提交不能解決消息重複的問題，也很好理解，假如消費0-100條消息，50條時掛了，重啟後由於沒有提交這一批消息的offset，是會從0開始重新消費。至於如何避免重複消費的問題，在這篇文章有說。

手動提交又分為異步提交和同步提交。

同步提交

上面案例代碼使用的是commitSync() ，顧名思義，是同步提交位移的方法。同步提交位移Consumer 程序會處於阻塞狀態，等待 Broker 返回提交結果。同步模式下提交失敗的時候一直嘗試提交，直到遇到無法重試的情況下才會結束。在任何系統中，因為程序而非資源限制而導致的阻塞都可能是系統的瓶頸，會影響整個應用程序的 TPS。當然，你可以選擇拉長提交間隔，但這樣做的後果是 Consumer 的提交頻率下降，在下次 Consumer 重啟回來后，會有更多的消息被重新消費。因此，為了解決這些不足，kafka還提供了異步提交方法。

異步提交

異步提交會立即返回，不會阻塞，因此不會影響 Consumer 應用的 TPS。由於它是異步的，Kafka 提供了回調函數，供你實現提交之後的邏輯，比如記錄日誌或處理異常等。下面這段代碼展示了調用 commitAsync() 的方法

 consumer.commitAsync((offsets, exception) -> {
 if (exception != null)
     handleException(exception);
 });

但是異步提交會有一個問題，那就是它沒有重試機制，不過一般情況下，針對偶爾出現的提交失敗，不進行重試不會有太大問題，因為如果提交失敗是因為臨時問題導致的，那麼後續的提交總會有成功的。所以消息也是不會丟失和重複消費的。
但如果這是發生在關閉消費者或再均衡前的最後一次提交，就要確保能夠提交成功。因此，組合使用commitAsync()和commitSync()是最佳的方式。

try {
    while (true) {
        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMinutes(KafkaConfig.pollTimeoutOfMinutes));
        if (!consumerRecords.isEmpty()) {
             for (ConsumerRecord<String, String> record : consumerRecords) {
                KafkaMessage kafkaMessage = JSON.parseObject(record.value(), KafkaMessage.class);
                boolean handleResult = handle(kafkaMessage);             
             }
             //異步提交位移               
             consumer.commitAsync((offsets, exception) -> {
             if (exception != null)
                 handleException(exception);
             });
           
        }
    }
} catch (Exception e) {
    System.out.println("kafka consumer error:" + e.toString());
} finally {
    try {
        //最後同步提交位移
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

讓位移提交更加靈活和可控

如果細心的閱讀了上面所有demo的代碼，那麼你會發現這樣幾個問題：

1、所有的提交，都是提交 poll 方法返回的所有消息的位移，poll 方法一次返回1000 條消息，則一次性地將這 1000 條消息的位移一併提交。可這樣一旦中間出現問題，位移沒有提交，下次會重新消費已經處理成功的數據。所以我想做到細粒度控制，比如每次提交100條，該怎麼辦？

答：可以通過commitSync(Map<TopicPartition, OffsetAndMetadata>) 和 commitAsync(Map<TopicPartition, OffsetAndMetadata>)對位移進行精確控制。

2、poll和commit方法對於普通的開發人員而言是一個黑盒，無法精確地掌控其消費的具體位置。我都不知道這次的提交，是針對哪個partition，提交上去的offset是多少。

答：可以通過record.topic()獲取topic信息, record.partition()獲取分區信息,record.offset() + 1獲取消費位移，記住消費位移是指示下一條消費的位移，所以要加一。

3、我想自己管理offset怎麼辦？一方面更加保險，一方面下次重啟之後可以精準的從數據庫讀取最後的offset就不存在丟失和重複消費了。
答：可以將消費位移保存在數據庫中。消費端程序使用comsumer.seek方法指定從某個位移開始消費。

綜合以上幾個可優化點，並結合全文，可以給出一個比較完美且完整的demo：聯合異步提交和同步提交，對處理過程中所有的異常都進行了處理。細粒度的控制了消費位移的提交，並且保守的將消費位移記錄到了數據庫中，重新啟動消費端程序的時候會從數據庫讀取位移。這也是我們消費端程序位移提交的最佳實踐方案。你只要繼承這個抽象類，實現你具體的業務邏輯就可以了。

public abstract class PrefectCosumer {
    private Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
    int count = 0;
    public final void consume() {
        Properties properties = PropertiesConfig.getConsumerProperties();
        properties.put("group.id", getGroupId());
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(getTopics());
        consumer.poll(0);
        // 把offset記錄到數據庫中 從指定的offset處消費 
        consumer.partitionsFor(getTopics()).stream().map(info ->
        new TopicPartition(getTopics(), info.partition()))
        .forEach(tp -> {
               consumer.seek(tp, JdbcUtils.queryOffset().get(tp.partition()));   
         });
        try {
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMinutes(KafkaConfig.pollTimeoutOfMinutes));
                if (!consumerRecords.isEmpty()) {
                    for (ConsumerRecord<String, String> record : consumerRecords) {

                        KafkaMessage kafkaMessage = JSON.parseObject(record.value(), KafkaMessage.class);
                        boolean handleResult = handle(kafkaMessage);
                        if (handleResult) {
                            //注意：提交的是下一條消息的位移。所以OffsetAndMetadata 對象時，必須使用當前消息位移加 1。
                            offsets.put(new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1));

                            // 細粒度控制提交 每10條提交一次offset
                            if (count % 10 == 0) {
                                // 異步提交offset
                                consumer.commitAsync(offsets, (offsets, exception) -> {
                                    if (exception != null) {
                                        handleException(exception);
                                    }
                                    // 將消費位移再記錄一份到數據庫中
                                    offsets.forEach((k, v) -> {
                                        String s = "insert into kafka_offset(`topic`,`group_id`,`partition_id`,`offset`) values" +
                                                " ('" + k.topic() + "','" + getGroupId() + "'," + k.partition() + "," + v.offset() + ")" +
                                                " on duplicate key update offset=values(offset);";
                                        JdbcUtils.insertTable(s);
                                    });


                                });
                            }
                            count++;
                        } else {         
                            System.out.println("消費消息失敗 kafkaMessage={}" + getTopics() + getGroupId() + kafkaMessage.toString());                         
                        }
                    }


                }
            }
        } catch (Exception e) {
            System.out.println("kafka consumer error:" + e.toString());
        } finally {
            try {
                // 最後一次提交 使用同步提交offset
                consumer.commitSync();
            } finally {
                consumer.close();
            }


        }
    }


    /**
     * 具體的業務邏輯
     *
     * @param kafkaMessage
     * @return
     */
    public abstract boolean handle(KafkaMessage kafkaMessage);

    public abstract List<String> getTopics();

    public abstract String getGroupId();

    void handleException(Exception e) {
        //異常處理
    }
}

控制位移提交的N種方式

剛剛我們說自己控制位移，使用seek方法可以指定offset消費。那到底怎麼控制位移？怎麼重設消費組位移？seek是什麼？現在就來仔細說說。

並不是所有的消息隊列都可以重設消費者組位移達到重新消費的目的。比如傳統的RabbitMq，它們處理消息是一次性的，即一旦消息被成功消費，就會被刪除。而Kafka消費消息是可以重演的，因為它是基於日誌結構（log-based）的消息引擎，消費者在消費消息時，僅僅是從磁盤文件上讀取數據而已，所以消費者不會刪除消息數據。同時，由於位移數據是由消費者控制的，因此它能夠很容易地修改位移的值，實現重複消費歷史數據的功能。

了解如何重設位移是很重要的。假設這麼一個場景，我已經消費了1000條消息后，我發現處理邏輯錯了，所以我需要重新消費一下，可是位移已經提交了，我到底該怎麼重新消費這1000條呢？？假設我想從某個時間點開始消費，我又該如何處理呢？

首先說個誤區：auto.offset.reset=earliest/latest這個參數大家都很熟悉，但是初學者很容易誤會它。大部分朋友都覺得在任何情況下把這兩個值設置為earliest或者latest ，消費者就可以從最早或者最新的offset開始消費,但實際上並不是那麼回事,他們生效都有一個前提條件,那就是對於同一個groupid的消費者,如果這個topic某個分區有已經提交的offset,那麼無論是把auto.offset.reset=earliest還是latest,都將失效,消費者會從已經提交的offset開始消費。因此這個參數並不能解決用戶想重設消費位移的需求。

kafka有七種控制消費組消費offset的策略，主要分為位移維度和時間維度，包括：

• 位移維度。這是指根據位移值來重設。也就是說，直接把消費者的位移值重設成我們給定的位移值。包括Earliest/Latest/Current/Specified-Offset/Shift-By-N策略

• 時間維度。我們可以給定一個時間，讓消費者把位移調整成大於該時間的最小位移；也可以給出一段時間間隔，比如 30 分鐘前，然後讓消費者直接將位移調回 30 分鐘之前的位移值。包括DateTime和Duration策略

說完了重設策略，我們就來看一下具體應該如何實現，可以從兩個角度，API方式和命令行方式。

重設位移的方法之API方式

API方式只要記住用seek方法就可以了，包括seek，seekToBeginning 和 seekToEnd。

void seek(TopicPartition partition, long offset);    
void seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata);    
void seekToBeginning(Collection<TopicPartition> partitions);    
void seekToEnd(Collection<TopicPartition> partitions);    

從方法簽名我們可以看出seekToBeginning和seekToEnd是可以一次性重設n個分區的位移，而seek 只允許重設指定分區的位移，即為每個分區都單獨設置位移，因為不難得出，如果要自定義每個分區的位移值則用seek，如果希望kafka幫你批量重設所有分區位移，比如從最新數據消費或者從最早數據消費，那麼用seekToEnd和seekToBeginning。

Earliest 策略：從最早的數據開始消費

從主題當前最早位移處開始消費，這個最早位移不一定就是 0 ，因為很久遠的消息會被 Kafka 自動刪除，主要取決於你的刪除配置。

代碼如下:

Properties properties = PropertiesConfig.getConsumerProperties();
properties.put("group.id", getGroupId());
Consumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(getTopics());
consumer.poll(0);
consumer.seekToBeginning(
consumer.partitionsFor(getTopics()).stream().map(partitionInfo ->
   new TopicPartition(getTopics(), partitionInfo.partition()))
   .collect(Collectors.toList()));

首先是構造consumer對象，這樣我們可以通過partitionsFor獲取到分區的信息，然後我們就可以構造出TopicPartition集合,傳給seekToBegining方法。需要注意的一個地方是：需要用consumer.poll(0)，而不能用consumer.poll(Duration.ofMillis(0))

在poll(0)中consumer會一直阻塞直到它成功獲取了所需的元數據信息，之後它才會發起fetch請求去獲取數據。而poll(Duration)會把元數據獲取也計入整個超時時間。由於本例中使用的是0，即瞬時超時，因此consumer根本無法在這麼短的時間內連接上coordinator，所以只能趕在超時前返回一個空集合。

Latest策略：從最新的數據開始消費

    consumer.seekToEnd(
        consumer.partitionsFor(getTopics().get(0)).stream().map(partitionInfo ->
            new TopicPartition(getTopics().get(0), partitionInfo.partition()))
              .collect(Collectors.toList()));

Current策略：從當前已經提交的offset處消費

consumer.partitionsFor(getTopics().get(0)).stream().map(info ->
        new TopicPartition(getTopics().get(0), info.partition()))
        .forEach(tp -> {
            long committedOffset = consumer.committed(tp).offset();
            consumer.seek(tp, committedOffset);
        });

**Special-offset策略：從指定的offset處消費 **

該策略使用的方法和current策略一樣，區別在於，current策略是直接從kafka元信息中讀取中已經提交的offset值，而special策略需要用戶自己為每一個分區指定offset值，我們一般是把offset記錄到數據庫中然後可以從數據庫去讀取這個值

    consumer.partitionsFor(getTopics().get(0)).stream().map(info ->
                new TopicPartition(getTopics().get(0), info.partition()))
                .forEach(tp -> {
                    try {
                        consumer.seek(tp, JdbcUtils.queryOffset().get(tp.partition()));
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                });

以上演示了用API方式重設位移，演示了四種常見策略的代碼，另外三種沒有演示，一方面是大同小異，另一方面在實際生產中，用API的方式不太可能去做時間維度的重設，而基本都是用命令行方式。

重設位移的方法之命令行方式

命令行方式重設位移是通過 kafka-consumer-groups 腳本。比起 API 的方式，用命令行重設位移要簡單得多。

Earliest 策略指定–to-earliest。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-earliest –execute

Latest 策略指定–to-latest。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-latest --execute

Current 策略指定–to-current。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-current --execute

Specified-Offset 策略指定–to-offset。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --all-topics --to-offset <offset> --execute

Shift-By-N 策略指定–shift-by N。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --shift-by <offset_N> --execute

DateTime 策略指定–to-datetime。

DateTime 允許你指定一個時間，然後將位移重置到該時間之後的最早位移處。常見的使用場景是，你想重新消費昨天的數據，那麼你可以使用該策略重設位移到昨天 0 點。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --to-datetime 2019-06-20T20:00:00.000 --execute

Duration 策略指定–by-duration。
Duration 策略則是指給定相對的時間間隔，然後將位移調整到距離當前給定時間間隔的位移處，具體格式是 PnDTnHnMnS。如果你熟悉 Java 8 引入的 Duration 類的話，你應該不會對這個格式感到陌生。它就是一個符合 ISO-8601 規範的 Duration 格式，以字母 P 開頭，後面由 4 部分組成，即 D、H、M 和 S，分別表示天、小時、分鐘和秒。舉個例子，如果你想將位移調回到 15 分鐘前，那麼你就可以指定 PT0H15M0S。

bin/kafka-consumer-groups.sh --bootstrap-server kafka-host:port --group test-group --reset-offsets --by-duration PT0H30M0S --execute

提交的位移都去哪了？

通過上面那幾部分的內容，我們已經搞懂了位移提交的方方面面，那麼提交的位移它保存在哪裡呢？這就要去位移主題的的世界里一探究竟了。kafka把位移保存在一個叫做__consumer_offsets的內部主題中，叫做位移主題。

注意：老版本的kafka其實是把位移保存在zookeeper中的，但是zookeeper並不適合這種高頻寫的場景。所以新版本已經是改進了這個方案，直接保存到kafka。畢竟kafka本身就適合高頻寫的場景，並且kafka也可以保證高可用性和高持久性。

既然它也是主題，那麼離不開分區和副本這兩個機制。我們並沒有手動創建這個主題並且指定，所以是kafka自動創建的， 分區的數量取決於Broker 端參數 offsets.topic.num.partitions，默認是50個分區，而副本參數取決於offsets.topic.replication.factor，默認是3。

既然也是主題，肯定會有消息，那麼消息格式是什麼呢？參考前面我們手動設計將位移寫入數據庫的方案，我們保存了topic,group_id,partition,offset四個字段。topic,group_id,partition無疑是數據表中的聯合主鍵，而offset是不斷更新的。無疑kafka的位移主題消息也是類似這種設計。key也是那三個字段，而消息體其實很複雜，你可以先簡單理解為就是offset。

既然也是主題，肯定也會有刪除策略，否則消息會無限膨脹。但是位移主題的刪除策略和其他主題刪除策略又不太一樣。我們知道普通主題的刪除是可以通過配置刪除時間或者大小的。而位移主題的刪除，叫做 Compaction。Kafka 使用Compact 策略來刪除位移主題中的過期消息，對於同一個 Key 的兩條消息 M1 和 M2，如果 M1 的發送時間早於 M2，那麼 M1 就是過期消息。Compact 的過程就是掃描日誌的所有消息，剔除那些過期的消息，然後把剩下的消息整理在一起。

Kafka 提供了專門的後台線程定期地巡檢待 Compact 的主題，看看是否存在滿足條件的可刪除數據。這個後台線程叫 Log Cleaner。很多實際生產環境中都出現過位移主題無限膨脹佔用過多磁盤空間的問題，如果你的環境中也有這個問題，我建議你去檢查一下 Log Cleaner 線程的狀態，通常都是這個線程掛掉了導致的。

總結

kafka的位移是個極其重要的概念，控制着消費進度，也即控制着消費的準確性，完整性，為了保證消息不重複和不丟失。我們最好做到以下幾點:

• 手動提交位移。

• 手動提交有異步提交和同步提交兩種方式，既然兩者有利也有弊，那麼我們可以結合起來使用。

• 細粒度的控制消費位移的提交，這樣可以避免重複消費的問題。

• 保守的將消費位移再記錄到了數據庫中，重新啟動消費端程序的時候從數據庫讀取位移。

獲取Kafka全套原創學習資料及思維導圖，關注【胖滾豬學編程】公眾號，回復”kafka”。

本文來源於公眾號：【胖滾豬學編程】。一枚集顏值與才華於一身，不算聰明卻足夠努力的女程序媛。用漫畫形式讓編程so easy and interesting！求關注！

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

【其他文章推薦】

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

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

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

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

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

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

2018 年可說是太陽能產業近年來波動最大的一年，歷經美國 201、301 條款，中國 531 新政，印度防衛性關稅，歐盟 MIP 結束等變動，從最上游的供應鏈到最下游的系統端都呈現極不穩定的狀態。由 EnergyTrend 所盤整的 2019 年五大趨勢來看，市況將會好轉，且產業也將在持續的變動中逐漸成熟。

趨勢一：2018 年低谷不低，2019 需求再創新高

中國的「531 新政」雖對市場造成衝擊，但因海外市場的需求走強，加上中國市場所受衝擊輕於預期，使 2018 年出現「低谷不低」的現象，預期全年新增併網量可達到 103GW（實際出貨量約95GW），年增4.9%。

展望 2019 年，在政策鼓勵與供應鏈價格持續下降的推波助瀾下，全球需求預計將繼續正成長，其中又以歐洲的成長幅度最大，最多可超過五成。2019 年預期新增併網量將來到 111.3GW，出現 7.7% 的成長，再次創下歷史新高。

趨勢二：市場持續分散，2019 年 GW 級市場增至 15 個

全球市場規模自 2018 年起預計會持穩在 100-120GW 之間，各年度需求量變化幅度將低於 10%。而根據 EnergyTrend 的最新需求報告統計，GW 級市場從 2016  年的 6  個成長到 2019 年將有 15 個，可見市場持續分散化的趨勢。

2016-2020 年 GW 級市場

中國、美國將持續穩居全球前二大市場，印度則從 2017 年起成為第三大需求國，日本次之。東南亞、北非、中東、拉丁美洲等新興市場自 2018 年崛起，如中東地區 2018 年全年需求預計將較 2017 年增加近 100%，2019  年還將增加 50% 左右。全球市場規模自 2019 年起將趨於穩定，印度最有可能出現較大幅度的需求成長。

2016-2023 年全球市場需求趨勢

趨勢三：供應鏈上游更為集中，單晶將逆轉市佔

雖然供應鏈整體在 2018 年陷於供過於求、低利潤的困境，但技術和成本優勢較強、全球布局較廣的一線大廠仍保有強勁的營運動能，既有的擴產計畫多能持續進行，使供應鏈廠家有持續集中化的現象。根據 EnergyTrend  的供給資料庫，中國前五大多晶矽廠的新產能預計在 2Q19 陸續開出，屆時前五大廠的產能將佔全球近 70%，且現金成本更具競爭力。在矽晶圓環節，則將呈現隆基與中環雙龍頭主宰市場的現象，單晶供應鏈也將因而變得更具主導性，有機會拉升全年單晶佔比來到 6 成，2017 年底展開的單多晶之戰逐漸落幕。

趨勢四：雙面產品產能倍增，P-PERC 效率還有成長空間

雙面電池技術已十分成熟，且可在幾乎不增加額外成本的前提下創造額外的發電收益，因此產能比例持續上升，預計 2019 年雙面電池的總產能將接近 40GW，且以雙面單晶 PERC 電池產能增加最多。另一方面，單晶 PERC 電池的量產效率仍有成長空間。據 EnergyTrend 調查，單晶 PERC 電池的平均量產效率在 2019 年上半年即可站上 22%，且還可導入更多技術，在 2019 年底效率可望上看 23%。而單晶 PERC 的強勢也壓縮了次世代 N 型技術的發展空間，2019 年 N 型產能預期僅會有小幅增加。

雙面電池產能成長趨勢（Unit: GW）

趨勢五：均化度電成本成為模組價格降價指標

供應鏈價格持續下探，使太陽能逐步朝擺脫補貼、平價上網的方向邁進；而無補貼系統的普及程度及其實際的均化度電成本（LCOE）將成為未來供應鏈的價格指標。

太陽能產業在 2018 年面臨強大考驗，但同時也進入產業盤整階段，預期長期發展將趨於穩定化與健康化，供應鏈的價格將以整體系統的度電成本為依歸。儲能系統與智慧電網技術的投入，將成為太陽能產業進一步市場化的關鍵。

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

【其他文章推薦】

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

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

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

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

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

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

南韓能源業者 SK Innovation 週一宣布，為鞏固美國客戶的訂單，正在考慮赴美設置電動車電池廠。美國為全球最大電動車市場之一，電池需求可期。

美國總統川普五月根據 232 條款，以國安之名就進口汽車與零件啟動調查，雖然還未決定是否開徵 25％ 的懲罰性關稅，但 SK Innovation 顯然已未雨綢繆，提早思考因應對策與未來投資佈局。

路透社報導，SK Innovation 發言人表示，設廠位置目前鎖定在美國南部地區，已有二至三個州列入考慮，但確切時間與其它設廠細節則還未敲定。

SK Innovation 競爭對手 LG Chem 目前已在美國設有生產據點，通用是主要客戶。

根據國際能源總署（International Energy Agency）五月底發佈報告顯示，2017 年全球電動車加油電混合動力車突破三百萬輛、年增 54%，中國為驅動成長主力，美國與北歐國家也快速成長中。（businessgreen.com）

（本文內容由 授權使用。首圖來源：）

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

【其他文章推薦】

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

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

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

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

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

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

原文鏈接：https://fuckcloudnative.io/posts/deploy-k3s-cross-public-cloud/

TCP 重置攻擊 是使用一個單一的數據包來執行的，只有幾個字節大小。攻擊者製作併發送一個偽造的 TCP 重置包來干擾用戶和網站的連接，欺騙通信雙方終止 TCP 連接。我們偉大的 xx 長城便運用了這個技術來進行 TCP 關鍵字阻斷。

理解 TCP 重置攻擊並不需要具備深厚的網絡知識功底，只需要一台筆記本就可以對自己進行模擬攻擊。本文將會帶你了解 TCP 重置攻擊的原理，同時會幫助你理解很多關於 TCP 協議的特性。本文主要內容：

• 回顧 TCP 協議的基礎知識
• 了解 TCP 重置攻擊的原理
• 使用一個簡單的 Python 腳本來模擬攻擊

下面開始分析 TCP 重置攻擊原理。

1. 偉大的 xx 長城是如何利用 TCP 重置攻擊的？

這一段略過，原因你懂得，感興趣的請直接看原文。

2. TCP 重置攻擊的工作原理

在 TCP 重置攻擊中，攻擊者通過向通信的一方或雙方發送偽造的消息，告訴它們立即斷開連接，從而使通信雙方連接中斷。正常情況下，如果客戶端收發現到達的報文段對於相關連接而言是不正確的，TCP 就會發送一個重置報文段，從而導致 TCP 連接的快速拆卸。

TCP 重置攻擊利用這一機制，通過向通信方發送偽造的重置報文段，欺騙通信雙方提前關閉 TCP 連接。如果偽造的重置報文段完全逼真，接收者就會認為它有效，並關閉 TCP 連接，防止連接被用來進一步交換信息。服務端可以創建一個新的 TCP 連接來恢復通信，但仍然可能會被攻擊者重置連接。萬幸的是，攻擊者需要一定的時間來組裝和發送偽造的報文，所以一般情況下這種攻擊只對長連接有殺傷力，對於短連接而言，你還沒攻擊呢，人家已經完成了信息交換。

從某種意義上來說，偽造 TCP 報文段是很容易的，因為 TCP/IP 都沒有任何內置的方法來驗證服務端的身份。有些特殊的 IP 擴展協議（例如 IPSec）確實可以驗證身份，但並沒有被廣泛使用。客戶端只能接收報文段，並在可能的情況下使用更高級別的協議（如 TLS）來驗證服務端的身份。但這個方法對 TCP 重置包並不適用，因為 TCP 重置包是 TCP 協議本身的一部分，無法使用更高級別的協議進行驗證。

儘管偽造 TCP 報文段很容易，但偽造正確的 TCP 重置報文段並完成攻擊卻並不容易。為了理解這項工作的難度，我們需要先了解一下 TCP 協議的工作原理。

3. TCP 協議工作原理

TCP 協議的目標是向客戶端發送一份完整的數據副本。例如，如果我的服務器通過 TCP 連接向你的計算機發送我的網站的 HTML，你的計算機的 TCP 協議棧應該能夠以我發送的形式和順序輸出 HTML。

然而現實生活中我的 HTML 內容並不是按順序發送的，它被分解成許多小塊(稱為 TCP 分組)，每個小塊在網絡上被單獨發送，並被重新組合成原來發送的順序。這種重新組合后的輸出被稱為 TCP 字節流。

將分組重建成字節流並不簡單，因為網絡是不可靠的。TCP分組可能會被丟棄，可能不按發送的順序到達客戶端，也可能會被重複發送、報文損壞等等。因此，TCP 協議的職責是在不可靠的網絡上提供可靠的通信。TCP 通過要求連接雙方保持密切聯繫，持續報告它們接收到了哪些數據來實現可靠通信，這樣服務端就能夠推斷出客戶端尚未接收到的數據，並重新發送丟失的數據。

為了進一步理解這個過程，我們需要了解服務端和客戶端是如何使用序列號（sequence numbers）來標記和跟蹤數據的。

TCP 序列號

TCP 協議的通信雙方， 都必須維護一個序列號（sequence numbers），對於客戶端來說，它會使用服務端的序列號來將接收到的數據按照發送的順序排列。

當通信雙方建立 TCP 連接時，客戶端與服務端都會向對方發送一個隨機的初始序列號，這個序列號標識了其發送數據流的第一個字節。TCP 報文段包含了 TCP 頭部，它是附加在報文段開頭的元數據，序列號就包含在 TCP 頭部中。由於 TCP 連接是雙向的，雙方都可以發送數據，所以 TCP 連接的雙方既是發送方也是接收方，每一方都必須分配和管理自己的序列號。

確認應答

當接收方收到一個 TCP 報文段時，它會向發送方返回一個 ACK 應答報文（同時將 TCP 頭部的 ACK 標誌位置 1），這個 ACK 號就表示接收方期望從發送方收到的下一個字節的序列號。發送方利用這個信息來推斷接收方已經成功接收到了序列號為 ACK 之前的所有字節。

TCP 頭部格式如下圖所示：

一個確認應答報文的 TCP 頭部必須包含兩個部分：

• ACK 標誌位置位 1
• 包含確認應答號（ACK number）

TCP 總共有 6 個標誌位，下文就會講到其中的 RST 標誌位。

TCP 頭部包含了多個選項，其中有一個選擇確認選項（SACK），如果使用該選項，那麼當接收方收到了某個範圍內的字節而不是連續的字節時，就會發送 SACK 告知對方。例如，只收到了字節 1000~3000 和 4000~5000，但沒有收到 3001~3999。為了簡單起見，下文討論 TCP 重置攻擊時將忽略選擇確認選項。

如果發送方發送了報文後在一段時間內沒有收到 ACK，就認為報文丟失了，並重新發送報文，用相同的序列號標記。這就意味着，如果接收方收到了重複的報文，可以使用序列號來判斷是否見過這個報文，如果見過則直接丟棄。網絡環境是錯綜複雜的，往往並不是如我們期望的一樣，先發送的數據包，就先到達目標主機，反而它很騷，可能會由於網絡擁堵等亂七八糟的原因，會使得舊的數據包，先到達目標主機。一般分兩種情況：

1. 發送的數據包丟失了
2. 發送的數據包被成功接收，但返回的 ACK 丟失了

這兩種情況對發送方來說其實是一樣的，發送方並不能區分是哪種情況，所以只能重新發送數據包。

只要不頻繁重複發送數據，額外的開銷基本可以忽略。

為偽造的重置包選擇序列號

構建偽造的重置包時需要選擇一個序列號。接收方可以接收序列號不按順序排列的報文段，但這種容忍是有限度的，如果報文段的序列號與它期望的相差甚遠，就會被直接丟棄。

因此，一個成功的 TCP 重置攻擊需要構建一個可信的序列號。但什麼才是可信的序列號呢？對於大多數報文段（除了重置包，即 RST 包）來說，序列號是由接收方的接收窗口大小決定的。

TCP 滑動窗口大小

想象一下，將一台上世紀 90 年代初的古老計算機，連接到現代千兆光纖網絡。閃電般快速的網絡可以以令人瞠目結舌的速度向這台古老的計算機傳送數據，速度遠遠超過該計算機的處理能力。但並沒有什麼卵用，因為只有接收方接收並處理了報文，才能認為這個報文已經被收到了。

TCP 協議棧有一個緩衝區，新到達的數據被放到緩衝區中等待處理。但緩衝區的大小是有限的，如果接收方的處理速度跟不上發送方的發送速度，緩衝區就會被填滿。一旦緩衝區被填滿，多餘的數據就會被直接丟棄，也不會返回 ACK。因此一旦接收方的緩衝區有了空位，發送方必須重新發送數據。也就是說，如果接收方的處理速度跟不上，發送方的發送速度再快也沒用。

緩衝區到底有多大？發送方如何才能知道什麼時候可以一次發送更多的數據，什麼時候該一次發送很少的數據？這就要靠 TCP 滑動窗口了。接收方的滑動窗口大小是指發送方無需等待確認應答，可以持續發送數據的最大值。 假設接收方的通告窗口大小為 100,000 字節，那麼發送方可以無需等待確認應答，持續發送 100,000 個字節。再假設當發送方發送第 100,000 個字節時，接收方已經發送了前 10,000 個字節的 ACK，這就意味着窗口中還有 90,000 個字節未被確認，發送方還可以再持續發送 10,000 個字節。如果發送了 10,000 個字節的過程中沒有收到任何的 ACK，那麼接收方的滑動窗口將被填滿，發送方將停止發送新數據（可以繼續發送之前丟失的數據），直到收到相關的 ACK 才可以繼續發送。

TCP 連接雙方會在建立連接的初始握手階段通告對方自己窗口的大小，後續還可以動態調整。TCP 緩衝區大的服務器可能會聲明一個大窗口，以便最大限度提高吞吐量。TCP 緩衝區小的服務器可能會被迫聲明一個小窗口，這樣做會犧牲一定的吞吐量，但為了防止接收方的 TCP 緩衝區溢出，還是很有必要的。

換個角度來看，TCP 滑動窗口大小是對網絡中可能存在的未確認數據量的硬性限制。我們可以用它來計算髮送方在某一特定時間內可能發送的最大序列號（max_seq_no）：

max_seq_no = max_acked_seq_no + window_size

其中 max_acked_seq_no 是接收方發送的最大 ACK 號，它表示發送方知道接收方已經成功接收的最大序列號。window_size 是窗口大小，它表示允許發送方最多發送的未被確認的字節。所以發送方可以發送的最大序列號是：max_acked_seq_no + window_size。

TCP 規範規定，接收方應該忽略任何序列號在接收窗口之外的數據。例如，如果接收方確認了所有序列號在 15,000 以下的字節，且接收窗口大小為 30,000，那麼接下來接收方只能接收序列號範圍在 15,000 ~ 45,000 之間的數據。如果一個報文段的部分數據在窗口內，另一部分數據在窗口外，那麼窗口內的數據將被接收確認，窗口外的數據將被丟棄。注意：這裏忽略了選擇確認選項，再強調一遍！

對於大多數 TCP 報文段來說，滑動窗口的規則告訴了發送方自己可以接收的序列號範圍。但對於重置報文來說，序列號的限制更加嚴格，這是為了抵禦一種攻擊叫做盲目 TCP 重置攻擊（blind TCP reset attack），下文將會解釋。

TCP 重置報文段的序列號

對於 TCP 重置報文段來說，接收方對序列號的要求更加嚴格，只有當其序列號正好等於下一個預期的序列號時才能接收。繼續搬出上面的例子，接收方發送了一個確認應答，ACK 號為 15,000。如果接下來收到了一個重置報文，那麼其序列號必須是 15,000 才能被接收。

如果重置報文的序列號超出了接收窗口範圍，接收方就會直接忽略該報文；如果其序列號在接收窗口範圍內，那麼接收方就會返回一個 challenge ACK，告訴發送方重置報文段的序列號是錯誤的，並告之正確的序列號，發送方可以利用 challenge ACK 中的信息來重新構建和發送重置報文。

其實在 2010 年之前，TCP 重置報文段和其他報文段的序列號限制規則一樣，但無法抵禦盲目 TCP 重置攻擊，後來才採取這些措施施加額外的限制。

盲目 TCP 重置攻擊

如果攻擊者能夠截獲通信雙方正在交換的信息，攻擊者就能讀取其數據包上的序列號和確認應答號，並利用這些信息得出偽裝的 TCP 重置報文段的序列號。相反，如果無法截獲通信雙方的信息，就無法確定重置報文段的序列號，但仍然可以批量發出盡可能多不同序列號的重置報文，以期望猜對其中一個序列號。這就是所謂的盲目 TCP 重置攻擊（blind TCP reset attack）。

在 2010 年之前 TCP 的原始版本中，攻擊者只需要猜對接收窗口內的隨便哪一個序列號即可，一般只需發送幾萬個報文段就能成功。採取額外限制的措施后，攻擊者需要發送數以百萬計的報文段才有可能猜對序列號，這幾乎是很難成功的。更多細節請參考 RFC-5963。

4. 模擬攻擊

以下實驗是在 OSX 系統中完成的，其他系統請自行測試。

現在來總結一下偽造一個 TCP 重置報文要做哪些事情：

• 嗅探通信雙方的交換信息。
• 截獲一個 ACK 標誌位置位 1 的報文段，並讀取其 ACK 號。
• 偽造一個 TCP 重置報文段（RST 標誌位置為 1），其序列號等於上面截獲的報文的 ACK 號。這隻是理想情況下的方案，假設信息交換的速度不是很快。大多數情況下為了增加成功率，可以連續發送序列號不同的重置報文。
• 將偽造的重置報文發送給通信的一方或雙方，時其中斷連接。

為了實驗簡單，我們可以使用本地計算機通過 localhost 與自己通信，然後對自己進行 TCP 重置攻擊。需要以下幾個步驟：

1. 在兩個終端之間建立一個 TCP 連接。
2. 編寫一個能嗅探通信雙方數據的攻擊程序。
3. 修改攻擊程序，偽造併發送重置報文。

下面正式開始實驗。

建立 TCP 連接

可以使用 netcat 工具來建立 TCP 連接，這個工很多操作系統都預裝了。打開第一個終端窗口，運行以下命令：

$ nc -nvl 8000

這個命令會啟動一個 TCP 服務，監聽端口為 8000。接着再打開第二個終端窗口，運行以下命令：

$ nc 127.0.0.1 8000

該命令會嘗試與上面的服務建立連接，在其中一個窗口輸入一些字符，就會通過 TCP 連接發送給另一個窗口並打印出來。

嗅探流量

編寫一個攻擊程序，使用 Python 網絡庫 scapy 來讀取兩個終端窗口之間交換的數據，並將其打印到終端上。完整的代碼參考我的 GitHub 倉庫，代碼的核心是調用 scapy 的嗅探方法：

t = sniff(
        iface='lo0',
        lfilter=is_packet_tcp_client_to_server(localhost_ip, localhost_server_port, localhost_ip),
        prn=log_packet,
        count=50)

這段代碼告訴 scapy 在 lo0 網絡接口上嗅探數據包，並記錄所有 TCP 連接的詳細信息。

• iface : 告訴 scapy 在 lo0（localhost）網絡接口上進行監聽。
• lfilter : 這是個過濾器，告訴 scapy 忽略所有不屬於指定的 TCP 連接（通信雙方皆為 localhost，且端口號為 8000）的數據包。
• prn : scapy 通過這個函數來操作所有符合 lfilter 規則的數據包。上面的例子只是將數據包打印到終端，下文將會修改函數來偽造重置報文。
• count : scapy 函數返回之前需要嗅探的數據包數量。

發送偽造的重置報文

下面開始修改程序，發送偽造的 TCP 重置報文來進行 TCP 重置攻擊。根據上面的解讀，只需要修改 prn 函數就行了，讓其檢查數據包，提取必要參數，並利用這些參數來偽造 TCP 重置報文併發送。

例如，假設該程序截獲了一個從（src_ip, src_port）發往 （dst_ip, dst_port）的報文段，該報文段的 ACK 標誌位已置為 1，ACK 號為 100,000。攻擊程序接下來要做的是：

• 由於偽造的數據包是對截獲的數據包的響應，所以偽造數據包的源 IP/Port 應該是截獲數據包的目的 IP/Port，反之亦然。
• 將偽造數據包的 RST 標誌位置為 1，以表示這是一個重置報文。
• 將偽造數據包的序列號設置為截獲數據包的 ACK 號，因為這是發送方期望收到的下一個序列號。
• 調用 scapy 的 send 方法，將偽造的數據包發送給截獲數據包的發送方。

對於我的程序而言，只需將這一行取消註釋，並註釋這一行的上面一行，就可以全面攻擊了。按照步驟 1 的方法設置 TCP 連接，打開第三個窗口運行攻擊程序，然後在 TCP 連接的其中一個終端輸入一些字符串，你會發現 TCP 連接被中斷了！

進一步實驗

1. 可以繼續使用攻擊程序進行實驗，將偽造數據包的序列號加減 1 看看會發生什麼，是不是確實需要和截獲數據包的 ACK 號完全相同。
2. 打開 Wireshark，監聽 lo0 網絡接口，並使用過濾器 ip.src == 127.0.0.1 && ip.dst == 127.0.0.1 && tcp.port == 8000 來過濾無關數據。你可以看到 TCP 連接的所有細節。
3. 在連接上更快速地發送數據流，使攻擊更難執行。

總的來說，TCP 重置攻擊既深奧又簡單，祝你實驗順利。

Kubernetes 1.18.2 1.17.5 1.16.9 1.15.12離線安裝包發布地址http://store.lameleg.com ，歡迎體驗。 使用了最新的sealos v3.3.6版本。 作了主機名解析配置優化，lvscare 掛載/lib/module解決開機啟動ipvs加載問題， 修復lvscare社區netlink與3.10內核不兼容問題,sealos生成百年證書等特性。更多特性 https://github.com/fanux/sealos 。歡迎掃描下方的二維碼加入釘釘群 ，釘釘群已經集成sealos的機器人實時可以看到sealos的動態。

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

【其他文章推薦】

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

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

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

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

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

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