Alibaba Nacos 學習(五):K8S Nacos搭建,使用nfs

 

準備環境

Centos7  192.168.50.21 k8s-master 2G
Centos7  192.168.50.22 k8s-node01 2G
Centos7  192.168.50.23 k8s-node02 2G

K8S集群搭建參考 

 

master安裝好Git ,yum install git

master,node01,node02  安裝 nfs-utils

yum install nfs-utils

master,node01,node02添加nfs exports配置,為了解決後續的nfs報錯異常

/data/mysql-slave *(insecure,fsid=0,rw,async,no_root_squash)
/data/mysql-master *(insecure,fsid=0,rw,async,no_root_squash)
/data/nfs-share *(rw,fsid=0,sync,no_root_squash)
mysql-slave 數據庫從庫 
mysql-master 數據庫主庫
nfs-share nocas文件掛在目錄

後面的yml中會提到
master,node01,node02創建目錄
mkdir /data/mysql-slave
mkdir /data/mysql-master
mkdir /data/nfs-share 

 master 克隆代碼

   git clone https://github.com/nacos-group/nacos-k8s.git

克隆完成進入以下目錄

 cd /opt/nacos-k8s/deploy/

 

1.nfs安裝

kubectl create -f nfs/rbac.yaml 
kubectl create -f nfs/class.yaml 

修改nfs/deployment.yaml IP配置

 

 

 

kubectl create -f nfs/deployment.yaml

查看安裝狀態

kubectl get pod -l app=nfs-client-provisioner

 

 

 

2.mysql部署

cd /opt/nacos-k8s/deploy/mysql/

修改數據配置文件ip

vi mysql-master-nfs.yaml

 

 

 部署主庫

kubectl create -f mysql-master-nfs.yaml 

修改存庫ip

vi mysql-slave-nfs.yaml
kubectl create -f mysql-slave-nfs.yaml 

主從部署非常慢 耐心等待,如果報nfs相關的錯,重啟nfs即可

service nfs restart

 

 

3. 部署nacos

cd /opt/nacos-k8s/deploy/nacos/

 

 

 

 

 

kubectl create -f nacos-pvc-nfs.yaml 

 查看訪問端口

kubectl get svc|grep nacos

 

 

 

 

 查看K8S集群狀態

 

 Failed to pull image “nacos/nacos-server:latest”: rpc error: code = Unknown desc = context canceled

進去對應節點機器 ,拉取鏡像后,重新應用即可

kubectl apply -f

 4. 部署問題

部署過程中大部分都是NFS問題

可以參考

mount.nfs: No route to host
Warning FailedMount 100s (x5 over 10m) kubelet, node2 Unable to mount volumes for pod “nfs-client-provisioner-594f778474-whhb5_default(56aef93a-9d31-11e9-a4c4-00163e069f44)”: timeout expired waiting for volumes to attach or mount for pod “default”/”nfs-client-provisioner-594f778474-whhb5”. list of unmounted volumes=[nfs-client-root]. list of unattached volumes=[nfs-client-root nfs-client-provisioner-token-8dcrx]

修改deployment.yaml中server的IP地址為某個node節點的內網IP地址,圖1已標註

 

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

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

如何使用偽類選擇器

偽類選擇器介紹

  • 偽類選擇器就是用來給超級鏈接設置不同的狀態樣式。
  • 超級鏈接分為4種狀態如:正常狀態、訪問過後狀態、鼠標放上狀態、激活狀態。

偽類選擇器說明表

選擇器 描述
:link 向未被訪問的超級鏈接添加樣式,正常狀態。
:visited 向已經被訪問的超級鏈接添加樣式,訪問過後狀態。
:hover 當鼠標懸浮在超級鏈接上方時,向超級鏈接添加樣式,鼠標放上狀態。
:active 鼠標放在超級鏈接上並且點擊的一瞬間,向超級鏈接添加樣式,激活狀態。

偽類選擇器實踐

  • 讓我們進入偽類選擇器實踐,實踐內容將超級鏈接4種狀態進行演示,演示效果如:將向未被訪問的超級鏈接文本顏色設置為紅色、已經被訪問的超級鏈接文本顏色設置為綠色、當鼠標懸浮在超級鏈接上文本顏色設置為紫色、用鼠標點擊超級鏈接的一瞬間文本顏色設置為藍色

  • 代碼塊

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>偽類選擇器</title>
    <style>
        a:link{
            color:red;
        }
        a:visited{
            color: lime;
        }
        a:hover{
            color: purple;
        }
        a:active{
            color: blue;
        }
    </style>
</head>
  
<body>
    <a href="https://www.cnblogs.com/lq000122/">微笑是最初的信仰</a>
</body>
</html>
  • 正常狀態結果圖

  • 鼠標放上狀態結果圖

  • 激活狀態結果圖

  • 訪問過後狀態

總結

  • 超級鏈接的不同狀態他其實是由順序,也就是說偽類選擇器設置其實是順序的,如果按照偽類選擇器的順序,那麼設置的樣式就不會被渲染。
  • 順序:linkvisitedhoveractive

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

3c收購,鏡頭 收購有可能以全新價回收嗎?

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

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

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

賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

PHP 的 self 關鍵字用法

之前有人詢問 self 關鍵字的用法,答案是比較明顯的:靜態成員函數內不能用 this 調用非成員函數,但可以用 self 調用靜態成員函數/變量/常量;其他成員函數可以用 self 調用靜態成員函數以及非靜態成員函數。隨着討論的深入,發現 self 並沒有那麼簡單。鑒於此,本文先對幾個關鍵字做對比和區分,再總結 self 的用法。

parentstatic 以及 this 的區別

要想將徹底搞懂 self ,要與 parentstatic 以及 this 區分開。以下分別做對比。

parent

selfparent 的區分比較容易: parent 引用父類/基類被隱蓋的方法(或變量), self則引用自身方法(或變量)。例如構造函數中調用父類構造函數:

class Base {
    public function __construct() {
        echo "Base contructor!", PHP_EOL;
    }
}

class Child {
    public function __construct() {
        parent::__construct();
        echo "Child contructor!", PHP_EOL;
    }
}

new Child;
// 輸出:
// Base contructor!
// Child contructor!

 

static

static 常規用途是修飾函數或變量使其成為類函數和類變量,也可以修飾函數內變量延長其生命周期至整個應用程序的生命周期。但是其與 self 關聯上是PHP 5.3以來引入的新用途:靜態延遲綁定。

有了 static 的靜態延遲綁定功能,可以在運行時動態確定歸屬的類。例如:

class Base {
    public function __construct() {
        echo "Base constructor!", PHP_EOL;
    }

    public static function getSelf() {
        return new self();
    }

    public static function getInstance() {
        return new static();
    }

    public function selfFoo() {
        return self::foo();
    }

    public function staticFoo() {
        return static::foo();
    }

    public function thisFoo() {
        return $this->foo();
    }

    public function foo() {
        echo  "Base Foo!", PHP_EOL;
    }
}

class Child extends Base {
    public function __construct() {
        echo "Child constructor!", PHP_EOL;
    }

    public function foo() {
        echo "Child Foo!", PHP_EOL;
    }
}

$base = Child::getSelf();
$child = Child::getInstance();

$child->selfFoo();
$child->staticFoo();
$child->thisFoo();

 

程序輸出結果如下:

Base constructor!
Child constructor!
Base Foo!
Child Foo!
Child Foo!

 

在函數引用上, selfstatic 的區別是:對於靜態成員函數, self 指向代碼當前類, static 指向調用類;對於非靜態成員函數, self 抑制多態,指向當前類的成員函數, static 等同於 this ,動態指向調用類的函數。

parentselfstatic 三個關鍵字聯合在一起看挺有意思,分別指向父類、當前類、子類,有點“過去、現在、未來”的味道。

this

selfthis 是被討論最多,也是最容易引起誤用的組合。兩者的主要區別如下:

  1. this 不能用在靜態成員函數中, self 可以;
  2. 對靜態成員函數/變量的訪問, 建議 用 self ,不要用 $this::$this-> 的形式;
  3. 對非靜態成員變量的訪問,不能用 self ,只能用 this ;
  4. this 要在對象已經實例化的情況下使用, self 沒有此限制;
  5. 在非靜態成員函數內使用, self 抑制多態行為,引用當前類的函數;而 this 引用調用類的重寫(override)函數(如果有的話)。

self 的用途

看完與上述三個關鍵字的區別, self 的用途是不是呼之即出?一句話總結,那就是: self總是指向“當前類(及類實例)”。詳細說則是:

  1. 替代類名,引用當前類的靜態成員變量和靜態函數;
  2. 抑制多態行為,引用當前類的函數而非子類中覆蓋的實現;

槽點

  1. 這幾個關鍵字中,只有 this 要加 $ 符號且必須加,強迫症表示很難受;
  2. 靜態成員函數中不能通過 $this-> 調用非靜態成員函數,但是可以通過 self:: 調用,且在調用函數中未使用 $this-> 的情況下還能順暢運行。此行為貌似在不同PHP版本中表現不同,在當前的7.3中ok;
  3. 在靜態函數和非靜態函數中輸出 self ,猜猜結果是什麼?都是 string(4) "self" ,迷之輸出;
  4. return $this instanceof static::class; 會有語法錯誤,但是以下兩種寫法就正常:
    $class = static::class;
    return $this instanceof $class;
    // 或者這樣:
    return $this instanceof static;

 

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

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

串燒 JavaCAS相關知識

JMM與問題引入

為啥先說JMM,因為CAS的實現類中維護的變量都被volatile修飾, 這個volatile 是遵循JMM規範(不是百分百遵循,下文會說)實現的保證多線程併發訪問某個變量實現線程安全的手段

一連串的知識點慢慢縷

首先說什麼是JMM, JMM就是大家所說的java的內存模型, 它是人們在邏輯上做出的劃分, 或者可以將JMM當成是一種規範, 有哪些規範呢? 如下

  1. 可見性: 某一個線程對內存中的變量做出改動后,要求其他的線程在第一事件內馬上馬得到通知,在CAS的實現中, 可見性其實是通過不斷的while循環讀取而得到的通知, 而不是被動的得到通知
  2. 原子性: 線程在執行某個操作的時,要麼一起成功,要麼就一起失敗
  3. 有序性: 為了提高性能, 編譯器處理器會進行指令的重排序, 源碼-> 編譯器優化重排 -> 處理器優化重排 -> 內存系統重排 -> 最終執行的命令

JVM運行的實體是線程, 每一個線程在創建之後JVM都會為其創建一個工作空間, 這個工作空間是每一個線程之間的私有空間, 並且任何兩條線程之間的都不能直接訪問到對方的工作空間, 線程之間的通信,必須通過共享空間來中轉完成

JMM規定所有的變量全部存在主內存中,主內存是一塊共享空間,那麼如果某個線程相對主內存中共享變量做出修改怎麼辦呢? 像下面這樣:

  1. 將共享變量的副本拷貝到工作空間中
  2. 對變量進行賦值修改
  3. 將工作空間中的變量寫回到內存中

JMM還規定如下:

  • 任何線程在解鎖前必須將工作空間的共享變量立即刷新進內存中
  • 線程在加鎖前必須讀取主內存中的值更新到自己的工作空間中
  • 加鎖和解鎖是同一把鎖

問題引入

這時候如果多個線程併發按照上面的三步走去訪問主內存中的共享變量的話就會出現線程安全性的問題, 比如說 現在主內存中的共享變量是c=1, 有AB兩個線程去併發訪問這個c變量, 都想進行c++, 現在A將c拷貝到自己的工作空間進行c++, 於是c=2 , 於此同時線程B也進行c++, c在B的工作空間中=2, AB線程將結果寫回工作空間最終的結果就是2, 而不是我們預期的3

相信怎麼解決大家都知道, 就是使用JUC,中的原子類就能規避這個問題

而原子類的底層實現使用的就是CAS技術

什麼是CAS

CAS(compare and swap) 顧名思義: 比較和交換,在JUC中原子類的底層使用的都是CAS無鎖實現線程安全,是一門很炫的技術

如下面兩行代碼, 先比較再交換, 即: 如果從主內存中讀取到的值為4就將它更新為2019

        AtomicInteger atomicInteger = new AtomicInteger(4);
        atomicInteger.compareAndSet(4,2019);

跟進AtomicInteger的源碼如下, 底層維護着一個int 類型的 變量, (當然是因為我選擇的原來類是AtomicInteger類型), 並且這個int類型的值被 volatile 修飾

    private volatile int value;

    /**
     * Creates a new AtomicInteger with the given initial value.
     *
     * @param initialValue the initial value
     */
    public AtomicInteger(int initialValue) {
        value = initialValue;
    }

什麼是volatile

volatile是JVM提供的輕量的同步機制, 為什麼是輕量界別呢? , 剛才在上面說了JMM規範中提到了三條特性, 而JVM提供的volatile僅僅滿足上面的規範中的 2/3, 如下:

  1. 保證可見性
  2. 不保證原子性
  3. 禁止指令重排序

單獨的volatile是不能滿足原子性的,即如下代碼在多線程併發訪問的情況下依然會出現線程安全性問題

private volatile int value;
 
public void add(){
  value++;   
}

那麼JUC的原子類是如何實現的 可以滿足原子性呢? 於是就不得不說本片博文的主角, CAS

CAS源碼跟進

我們跟進AtomicInteger中的先遞增再獲取的方法 incrementAndGet()

    public final int incrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
    }

通過代碼我們看到調用了Unsafe類來實現

什麼是Unsafe類?

進入Unsafe類,可以看到他裏面存在大量的 native方法,這些native方法全部是空方法,

這個unsafe類其實相當於一個後門,他是java去訪問調用系統上 C C++ 函數類庫的方法 如下圖

繼續跟進這個方法incrementAndGet() 於是我們就來到了我們的主角方法, 關於這個方法倒是不難理解,主要是搞清楚方法中的var12345到底代表什麼就行, 如下代碼+註釋

var1: 上一個方法傳遞進來的: this,即當前對象
var2: 上一個方法傳遞進來的valueOffset, 就是內存地址偏移量
      通過這個內存地址偏移量我能精確的找到要操作的變量在內存中的地址
      
var4: 上一個方法傳遞進來的1, 就是每次增長的值
var5: 通過this和內存地址偏移量讀取出來的當前內存中的目標值
public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

        return var5;
    }

注意它用的是while循環, 相對if(flag){} 這種寫法會多一次判斷, 整體的思路就是 在進行修改之前先進行一次比較,如果讀取到的當前值和預期值是相同的,就自增,否則的話就繼續輪詢修改

小總結

通過上面的過程, 其實就能總結出CAS的底層實現原理

  • volatile
  • 自旋鎖
  • unsafe類

補充: CAS通過Native方法的底層實現,本質上是操作系統層面上的CPU的併發原語,JVM會直接實現出彙編層面的指令,依賴於硬件去實現, 此外, 對於CPU的原語來說, 有兩條特性1,必定連續, 2.不被中斷

CAS的優缺點

優點:

它的底層我們看到了通過do-while 實現的自旋鎖來實現, 就省去了在多個線程之間進行切換所帶來的額外的上下文切換的開銷

缺點:

  1. 通過while循環不斷的嘗試獲取, 省去了上下文切換的開銷,但是佔用cpu的資源
  2. CAS只能保證一個共享變量的原子性, 如果存在多個共享變量的話不得不加鎖實現
  3. 存在ABA問題

ABA問題

什麼是ABA問題

我們這樣玩, 還是AB兩個線程, 給AtomicInteger賦初始值0

A線程中的代碼如下:

        Thread.sleep(3000);
        atomicInteger.compareAndSet(0,2019);

B線程中的代碼如下:

        atomicInteger.compareAndSet(0,1);
        atomicInteger.compareAndSet(1,0);

AB線程同時啟動, 雖然最終的結果A線程能成果的將值修改成2019,,但是它不能感知到在他睡眠過程中B線程對數據進行過改變, 換句話說就是A線程被B線程欺騙了

ABA問題的解決— AtomicStampedRefernce.java

帶時間戳的原子引用, 實現的機制就是通過 原子引用+版本號來完成, 每次對指定值的修改相應的版本號會加1, 實例如下

        // 0表示初始化, 1表示初始版本號
        AtomicStampedReference<Integer> reference = new AtomicStampedReference<>(0, 1);
        reference.getStamp(); // 獲取版本號
        reference.attemptStamp(1,2); // 期待是1, 如果是1就更新為2

原子引用

JUC中我們可以找到像AtomicInteger這樣已經定義好了實現類, 但是JUC沒有給我們提供類似這樣 AtomicUser或者 AtomicProduct 這樣自定義類型的原子引用類型啊, 不過java仍然是提供了後門就是 原子引用類型

使用實例:

        User user  = getUserById(1);
        AtomicReference<User> userAtomicReference = new AtomicReference<User>();
        user.setUsername("張三");
        userAtomicReference.compareAndSet(user,user);

歡迎關注我, 會繼續更新筆記

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

3c收購,鏡頭 收購有可能以全新價回收嗎?

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

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

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

賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

帶你漲姿勢的認識一下 Kafka 消費者

之前我們介紹過了 Kafka 整體架構,Kafka 生產者,Kafka 生產的消息最終流向哪裡呢?當然是需要消費了,要不只產生一系列數據沒有任何作用啊,如果把 Kafka 比作餐廳的話,那麼生產者就是廚師的角色,消費者就是客人,只有廚師的話,那麼炒出來的菜沒有人吃也沒有意義,如果只有客人沒有廚師的話,誰會去這個店吃飯呢?!所以如果你看完前面的文章意猶未盡的話,可以繼續讓你爽一爽。如果你沒看過前面的文章,那就從現在開始讓你爽。

Kafka 消費者概念

應用程序使用 KafkaConsumer 從 Kafka 中訂閱主題並接收來自這些主題的消息,然後再把他們保存起來。應用程序首先需要創建一個 KafkaConsumer 對象,訂閱主題並開始接受消息,驗證消息並保存結果。一段時間后,生產者往主題寫入的速度超過了應用程序驗證數據的速度,這時候該如何處理?如果只使用單個消費者的話,應用程序會跟不上消息生成的速度,就像多個生產者像相同的主題寫入消息一樣,這時候就需要多個消費者共同參与消費主題中的消息,對消息進行分流處理。

Kafka 消費者從屬於消費者群組。一個群組中的消費者訂閱的都是相同的主題,每個消費者接收主題一部分分區的消息。下面是一個 Kafka 分區消費示意圖

上圖中的主題 T1 有四個分區,分別是分區0、分區1、分區2、分區3,我們創建一個消費者群組1,消費者群組中只有一個消費者,它訂閱主題T1,接收到 T1 中的全部消息。由於一個消費者處理四個生產者發送到分區的消息,壓力有些大,需要幫手來幫忙分擔任務,於是就演變為下圖

這樣一來,消費者的消費能力就大大提高了,但是在某些環境下比如用戶產生消息特別多的時候,生產者產生的消息仍舊讓消費者吃不消,那就繼續增加消費者。

如上圖所示,每個分區所產生的消息能夠被每個消費者群組中的消費者消費,如果向消費者群組中增加更多的消費者,那麼多餘的消費者將會閑置,如下圖所示

向群組中增加消費者是橫向伸縮消費能力的主要方式。總而言之,我們可以通過增加消費組的消費者來進行水平擴展提升消費能力。這也是為什麼建議創建主題時使用比較多的分區數,這樣可以在消費負載高的情況下增加消費者來提升性能。另外,消費者的數量不應該比分區數多,因為多出來的消費者是空閑的,沒有任何幫助。

Kafka 一個很重要的特性就是,只需寫入一次消息,可以支持任意多的應用讀取這個消息。換句話說,每個應用都可以讀到全量的消息。為了使得每個應用都能讀到全量消息,應用需要有不同的消費組。對於上面的例子,假如我們新增了一個新的消費組 G2,而這個消費組有兩個消費者,那麼就演變為下圖這樣

在這個場景中,消費組 G1 和消費組 G2 都能收到 T1 主題的全量消息,在邏輯意義上來說它們屬於不同的應用。

總結起來就是如果應用需要讀取全量消息,那麼請為該應用設置一個消費組;如果該應用消費能力不足,那麼可以考慮在這個消費組裡增加消費者

消費者組和分區重平衡

消費者組是什麼

消費者組(Consumer Group)是由一個或多個消費者實例(Consumer Instance)組成的群組,具有可擴展性和可容錯性的一種機制。消費者組內的消費者共享一個消費者組ID,這個ID 也叫做 Group ID,組內的消費者共同對一個主題進行訂閱和消費,同一個組中的消費者只能消費一個分區的消息,多餘的消費者會閑置,派不上用場。

我們在上面提到了兩種消費方式

  • 一個消費者群組消費一個主題中的消息,這種消費模式又稱為點對點的消費方式,點對點的消費方式又被稱為消息隊列
  • 一個主題中的消息被多個消費者群組共同消費,這種消費模式又稱為發布-訂閱模式

消費者重平衡

我們從上面的消費者演變圖中可以知道這麼一個過程:最初是一個消費者訂閱一個主題並消費其全部分區的消息,後來有一個消費者加入群組,隨後又有更多的消費者加入群組,而新加入的消費者實例分攤了最初消費者的部分消息,這種把分區的所有權通過一個消費者轉到其他消費者的行為稱為重平衡,英文名也叫做 Rebalance 。如下圖所示

重平衡非常重要,它為消費者群組帶來了高可用性伸縮性,我們可以放心的添加消費者或移除消費者,不過在正常情況下我們並不希望發生這樣的行為。在重平衡期間,消費者無法讀取消息,造成整個消費者組在重平衡的期間都不可用。另外,當分區被重新分配給另一個消費者時,消息當前的讀取狀態會丟失,它有可能還需要去刷新緩存,在它重新恢復狀態之前會拖慢應用程序。

消費者通過向組織協調者(Kafka Broker)發送心跳來維護自己是消費者組的一員並確認其擁有的分區。對於不同不的消費群體來說,其組織協調者可以是不同的。只要消費者定期發送心跳,就會認為消費者是存活的並處理其分區中的消息。當消費者檢索記錄或者提交它所消費的記錄時就會發送心跳。

如果過了一段時間 Kafka 停止發送心跳了,會話(Session)就會過期,組織協調者就會認為這個 Consumer 已經死亡,就會觸發一次重平衡。如果消費者宕機並且停止發送消息,組織協調者會等待幾秒鐘,確認它死亡了才會觸發重平衡。在這段時間里,死亡的消費者將不處理任何消息。在清理消費者時,消費者將通知協調者它要離開群組,組織協調者會觸發一次重平衡,盡量降低處理停頓。

重平衡是一把雙刃劍,它為消費者群組帶來高可用性和伸縮性的同時,還有有一些明顯的缺點(bug),而這些 bug 到現在社區還無法修改。

重平衡的過程對消費者組有極大的影響。因為每次重平衡過程中都會導致萬物靜止,參考 JVM 中的垃圾回收機制,也就是 Stop The World ,STW,(引用自《深入理解 Java 虛擬機》中 p76 關於 Serial 收集器的描述):

更重要的是它在進行垃圾收集時,必須暫停其他所有的工作線程。直到它收集結束。Stop The World 這個名字聽起來很帥,但這項工作實際上是由虛擬機在後台自動發起並完成的,在用戶不可見的情況下把用戶正常工作的線程全部停掉,這對很多應用來說都是難以接受的。

也就是說,在重平衡期間,消費者組中的消費者實例都會停止消費,等待重平衡的完成。而且重平衡這個過程很慢……

創建消費者

上面的理論說的有點多,下面就通過代碼來講解一下消費者是如何消費的

在讀取消息之前,需要先創建一個 KafkaConsumer 對象。創建 KafkaConsumer 對象與創建 KafkaProducer 對象十分相似 — 把需要傳遞給消費者的屬性放在 properties 對象中,後面我們會着重討論 Kafka 的一些配置,這裏我們先簡單的創建一下,使用3個屬性就足矣,分別是 bootstrap.serverkey.deserializervalue.deserializer

這三個屬性我們已經用過很多次了,如果你還不是很清楚的話,可以參考

還有一個屬性是 group.id 這個屬性不是必須的,它指定了 KafkaConsumer 是屬於哪個消費者群組。創建不屬於任何一個群組的消費者也是可以的

Properties properties = new Properties();
        properties.put("bootstrap.server","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");
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

主題訂閱

創建好消費者之後,下一步就開始訂閱主題了。subscribe() 方法接受一個主題列表作為參數,使用起來比較簡單

consumer.subscribe(Collections.singletonList("customerTopic"));

為了簡單我們只訂閱了一個主題 customerTopic,參數傳入的是一個正則表達式,正則表達式可以匹配多個主題,如果有人創建了新的主題,並且主題的名字與正則表達式相匹配,那麼會立即觸發一次重平衡,消費者就可以讀取新的主題。

要訂閱所有與 test 相關的主題,可以這樣做

consumer.subscribe("test.*");

輪詢

我們知道,Kafka 是支持訂閱/發布模式的,生產者發送數據給 Kafka Broker,那麼消費者是如何知道生產者發送了數據呢?其實生產者產生的數據消費者是不知道的,KafkaConsumer 採用輪詢的方式定期去 Kafka Broker 中進行數據的檢索,如果有數據就用來消費,如果沒有就再繼續輪詢等待,下面是輪詢等待的具體實現

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
    for (ConsumerRecord<String, String> record : records) {
      int updateCount = 1;
      if (map.containsKey(record.value())) {
        updateCount = (int) map.get(record.value() + 1);
      }
      map.put(record.value(), updateCount);
    }
  }
}finally {
  consumer.close();
}
  • 這是一個無限循環。消費者實際上是一個長期運行的應用程序,它通過輪詢的方式向 Kafka 請求數據。
  • 第三行代碼非常重要,Kafka 必須定期循環請求數據,否則就會認為該 Consumer 已經掛了,會觸發重平衡,它的分區會移交給群組中的其它消費者。傳給 poll() 方法的是一個超市時間,用 java.time.Duration 類來表示,如果該參數被設置為 0 ,poll() 方法會立刻返回,否則就會在指定的毫秒數內一直等待 broker 返回數據。
  • poll() 方法會返回一個記錄列表。每條記錄都包含了記錄所屬主題的信息,記錄所在分區的信息、記錄在分區中的偏移量,以及記錄的鍵值對。我們一般會遍歷這個列表,逐條處理每條記錄。
  • 在退出應用程序之前使用 close() 方法關閉消費者。網絡連接和 socket 也會隨之關閉,並立即觸發一次重平衡,而不是等待群組協調器發現它不再發送心跳並認定它已經死亡。

線程安全性

在同一個群組中,我們無法讓一個線程運行多個消費者,也無法讓多個線程安全的共享一個消費者。按照規則,一個消費者使用一個線程,如果一個消費者群組中多個消費者都想要運行的話,那麼必須讓每個消費者在自己的線程中運行,可以使用 Java 中的 ExecutorService 啟動多個消費者進行進行處理。

消費者配置

到目前為止,我們學習了如何使用消費者 API,不過只介紹了幾個最基本的屬性,Kafka 文檔列出了所有與消費者相關的配置說明。大部分參數都有合理的默認值,一般不需要修改它們,下面我們就來介紹一下這些參數。

  • fetch.min.bytes

該屬性指定了消費者從服務器獲取記錄的最小字節數。broker 在收到消費者的數據請求時,如果可用的數據量小於 fetch.min.bytes 指定的大小,那麼它會等到有足夠的可用數據時才把它返回給消費者。這樣可以降低消費者和 broker 的工作負載,因為它們在主題使用頻率不是很高的時候就不用來回處理消息。如果沒有很多可用數據,但消費者的 CPU 使用率很高,那麼就需要把該屬性的值設得比默認值大。如果消費者的數量比較多,把該屬性的值調大可以降低 broker 的工作負載。

  • fetch.max.wait.ms

我們通過上面的 fetch.min.bytes 告訴 Kafka,等到有足夠的數據時才會把它返回給消費者。而 fetch.max.wait.ms 則用於指定 broker 的等待時間,默認是 500 毫秒。如果沒有足夠的數據流入 kafka 的話,消費者獲取的最小數據量要求就得不到滿足,最終導致 500 毫秒的延遲。如果要降低潛在的延遲,就可以把參數值設置的小一些。如果 fetch.max.wait.ms 被設置為 100 毫秒的延遲,而 fetch.min.bytes 的值設置為 1MB,那麼 Kafka 在收到消費者請求后,要麼返回 1MB 的數據,要麼在 100 ms 后返回所有可用的數據。就看哪個條件首先被滿足。

  • max.partition.fetch.bytes

該屬性指定了服務器從每個分區里返回給消費者的最大字節數。它的默認值時 1MB,也就是說,KafkaConsumer.poll() 方法從每個分區里返回的記錄最多不超過 max.partition.fetch.bytes 指定的字節。如果一個主題有20個分區和5個消費者,那麼每個消費者需要至少4 MB的可用內存來接收記錄。在為消費者分配內存時,可以給它們多分配一些,因為如果群組裡有消費者發生崩潰,剩下的消費者需要處理更多的分區。max.partition.fetch.bytes 的值必須比 broker 能夠接收的最大消息的字節數(通過 max.message.size 屬性配置大),否則消費者可能無法讀取這些消息,導致消費者一直掛起重試。 在設置該屬性時,另外一個考量的因素是消費者處理數據的時間。消費者需要頻繁的調用 poll() 方法來避免會話過期和發生分區再平衡,如果單次調用poll() 返回的數據太多,消費者需要更多的時間進行處理,可能無法及時進行下一個輪詢來避免會話過期。如果出現這種情況,可以把 max.partition.fetch.bytes 值改小,或者延長會話過期時間。

  • session.timeout.ms

這個屬性指定了消費者在被認為死亡之前可以與服務器斷開連接的時間,默認是 3s。如果消費者沒有在 session.timeout.ms 指定的時間內發送心跳給群組協調器,就會被認定為死亡,協調器就會觸發重平衡。把它的分區分配給消費者群組中的其它消費者,此屬性與 heartbeat.interval.ms 緊密相關。heartbeat.interval.ms 指定了 poll() 方法向群組協調器發送心跳的頻率,session.timeout.ms 則指定了消費者可以多久不發送心跳。所以,這兩個屬性一般需要同時修改,heartbeat.interval.ms 必須比 session.timeout.ms 小,一般是 session.timeout.ms 的三分之一。如果 session.timeout.ms 是 3s,那麼 heartbeat.interval.ms 應該是 1s。把 session.timeout.ms 值設置的比默認值小,可以更快地檢測和恢復崩憤的節點,不過長時間的輪詢或垃圾收集可能導致非預期的重平衡。把該屬性的值設置得大一些,可以減少意外的重平衡,不過檢測節點崩潰需要更長的時間。

  • auto.offset.reset

該屬性指定了消費者在讀取一個沒有偏移量的分區或者偏移量無效的情況下的該如何處理。它的默認值是 latest,意思指的是,在偏移量無效的情況下,消費者將從最新的記錄開始讀取數據。另一個值是 earliest,意思指的是在偏移量無效的情況下,消費者將從起始位置處開始讀取分區的記錄。

  • enable.auto.commit

我們稍後將介紹幾種不同的提交偏移量的方式。該屬性指定了消費者是否自動提交偏移量,默認值是 true,為了盡量避免出現重複數據和數據丟失,可以把它設置為 false,由自己控制何時提交偏移量。如果把它設置為 true,還可以通過 auto.commit.interval.ms 屬性來控制提交的頻率

  • partition.assignment.strategy

我們知道,分區會分配給群組中的消費者。PartitionAssignor 會根據給定的消費者和主題,決定哪些分區應該被分配給哪個消費者,Kafka 有兩個默認的分配策略RangeRoundRobin

  • client.id

該屬性可以是任意字符串,broker 用他來標識從客戶端發送過來的消息,通常被用在日誌、度量指標和配額中

  • max.poll.records

該屬性用於控制單次調用 call() 方法能夠返回的記錄數量,可以幫你控制在輪詢中需要處理的數據量。

  • receive.buffer.bytes 和 send.buffer.bytes

socket 在讀寫數據時用到的 TCP 緩衝區也可以設置大小。如果它們被設置為 -1,就使用操作系統默認值。如果生產者或消費者與 broker 處於不同的數據中心內,可以適當增大這些值,因為跨數據中心的網絡一般都有比較高的延遲和比較低的帶寬。

提交和偏移量的概念

特殊偏移

我們上面提到,消費者在每次調用poll() 方法進行定時輪詢的時候,會返回由生產者寫入 Kafka 但是還沒有被消費者消費的記錄,因此我們可以追蹤到哪些記錄是被群組裡的哪個消費者讀取的。消費者可以使用 Kafka 來追蹤消息在分區中的位置(偏移量)

消費者會向一個叫做 _consumer_offset 的特殊主題中發送消息,這個主題會保存每次所發送消息中的分區偏移量,這個主題的主要作用就是消費者觸發重平衡後記錄偏移使用的,消費者每次向這個主題發送消息,正常情況下不觸發重平衡,這個主題是不起作用的,當觸發重平衡后,消費者停止工作,每個消費者可能會分到對應的分區,這個主題就是讓消費者能夠繼續處理消息所設置的。

如果提交的偏移量小於客戶端最後一次處理的偏移量,那麼位於兩個偏移量之間的消息就會被重複處理

如果提交的偏移量大於最後一次消費時的偏移量,那麼處於兩個偏移量中間的消息將會丟失

既然_consumer_offset 如此重要,那麼它的提交方式是怎樣的呢?下面我們就來說一下

提交方式

KafkaConsumer API 提供了多種方式來提交偏移量

自動提交

最簡單的方式就是讓消費者自動提交偏移量。如果 enable.auto.commit 被設置為true,那麼每過 5s,消費者會自動把從 poll() 方法輪詢到的最大偏移量提交上去。提交時間間隔由 auto.commit.interval.ms 控制,默認是 5s。與消費者里的其他東西一樣,自動提交也是在輪詢中進行的。消費者在每次輪詢中會檢查是否提交該偏移量了,如果是,那麼就會提交從上一次輪詢中返回的偏移量。

提交當前偏移量

auto.commit.offset 設置為 false,可以讓應用程序決定何時提交偏移量。使用 commitSync() 提交偏移量。這個 API 會提交由 poll() 方法返回的最新偏移量,提交成功后馬上返回,如果提交失敗就拋出異常。

commitSync() 將會提交由 poll() 返回的最新偏移量,如果處理完所有記錄后要確保調用了 commitSync(),否則還是會有丟失消息的風險,如果發生了在均衡,從最近一批消息到發生在均衡之間的所有消息都將被重複處理。

異步提交

異步提交 commitAsync() 與同步提交 commitSync() 最大的區別在於異步提交不會進行重試,同步提交會一致進行重試。

同步和異步組合提交

一般情況下,針對偶爾出現的提交失敗,不進行重試不會有太大的問題,因為如果提交失敗是因為臨時問題導致的,那麼後續的提交總會有成功的。但是如果在關閉消費者或再均衡前的最後一次提交,就要確保提交成功。

因此,在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量

提交特定的偏移量

消費者API允許調用 commitSync() 和 commitAsync() 方法時傳入希望提交的 partition 和 offset 的 map,即提交特定的偏移量。

文章參考:

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

《Kafka 權威指南》

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

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

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

特斯拉發表電動卡車 Cybertruck,一台從科幻電影走出來的鋼鐵車

今天的 Elon Musk 看起來不像鋼鐵人,更像蝙蝠俠,因為他們的新車 Cybertruck,不但外型酷似蝙蝠車,同時還能防彈防撞,並且擁有超越保時捷的加速度,跟勝過市面上卡車的拖吊能力,更驚人的是,售價只要 39,900 美元起。

眾所期待的特斯拉新車 Cybertruck 今日正式發表,和之前流出的影像不同,Cybertruck 酷似隱形戰機 F-117 的設計,讓人聯想到蝙蝠車,甚至懷疑這是不是一台防雷達偵測的戰車?

Tesla Cybertruck 全車採用冷鑄鋼板,能夠抵擋 9mm 口徑手槍的射擊,現場展示用重鎚敲擊也毫髮無傷;車窗玻璃同樣採用防彈設計,然而有趣的是,現場展示時,被大鐵球砸出了一片雪花。「至少,它沒被打穿,你坐在裡面很安全。」Elon Musk 笑著說。

Cybertruck 為了因應負重,搭載了適應性氣壓懸吊系統,針對高速公路,或是越野泥巴路,能夠自動調整懸吊高度,同時也順便使用這個氣壓系統,做了一個高壓出力裝置,使用者可以自行加裝不同氣壓工具,像是高壓水槍或是電鑽等。

當重裝電動機車開上後廂時,懸吊系統會自動調整車尾高度,讓車身保持平衡。

車尾與其他皮卡車開放式貨斗不同,Cybertruck 採用封閉式貨斗,並有升降式尾門,現場展示時,將這台電動機車 ATV 直接騎上貨斗後,還能直接充電,顯然是在致敬蝙蝠車跟蝙蝠機車。

Cybertruck 如同其他皮卡車,車尾裝有釣鉤,能夠充當拖車使用,而歸功於它的強力馬達,拖車能力屌打了皮卡車霸主 Ford F-150,在現場展示的影片中,特斯拉讓 Cybertruck 跟 F-150 互相拖住對方,進行拔河測試,結果 F-150 整台被 Cybertruck 拖走。

F-150 慘遭 Cybertruck 拖走。

馬斯克強調,一般皮卡車需要另外裝載發電機才能使用電動工具,Cybertruck 直接提供了電源,因此省下不少空間,同時還提供強大的拖力。

此外,做為一台卡車,Cybertruck 莫名其妙地擁有超越保時捷的加速度,根據現場公布數據,最頂級版的 0-100 公里加速時間不到 3 秒。現場展示了 Cybertruck 與 Porsche 911 賽跑的影片,起步雖然小輸一點,但隨後就超越了 911。

現場展示競速影片,大約 1 秒後,Cybertruck 就超過了 911。

Tesla Cybertruck 共有 3 種版本,依照馬達數量來分別,最低價 39,900 美元起,最高 69,900 美元。Cybertruck 從今天起在美國開放預購,實際交車時間預計要等到 2021 年底。頂級的三馬達款,更預計要等到 2022 年底才會開始生產。

如同馬斯克開場所說,卡車在過去幾十年來都長得差不多,特斯拉要打造一台完全不一樣的卡車,同時還要保持零排放,跟超高性能,從今天的現場展示來看,特斯拉再次完成一個不可能的任務。在興奮之餘也別忘了,這一切都是現場展示,實際上如何,就有待實際交車後驗證了!

(合作媒體:。圖片來源:)

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

【其他文章推薦】

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

【其他文章推薦】

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

自己動手實現分佈式任務調度框架(續)

  之前寫過一篇:本來是用來閑來分享一下自己的思維方式,時至今日發現居然有些人正在使用了,本着對代碼負責人的態度,對代碼部分已知bug進行了修改,並增加了若干功能,如立即啟動,實時停止等功能,新增加的功能會在這一篇做詳細的說明。

  提到分佈式任務調度,市面上本身已經有一些框架工具可以使用,但是個人覺得功能做的都太豐富,架構都過於複雜,所以才有了我重複造輪子。個人喜歡把複雜的問題簡單化,利用有限的資源實現竟可能多的功能。因為有幾個朋友問部署方式,這裏再次強調下:我的這個服務可以直接打成jar放在自己本地倉庫,然後依賴進去,或者直接copy代碼過去,當成自己項目的一部分就可以了。也就是說跟隨你們自己的項目啟動,所以我這裏也沒有寫界面。下面先談談怎麼基於上次的代碼實現任務立即啟動吧!

  調度和自己服務整合後部署圖抽象成如下:

  

 

 

   用戶在前端點擊立即請求按鈕,通過各種負載均衡軟件或者設備,到達某台機器的某個帶有本調度框架的服務,然後進行具體的執行,也就是說這個立即啟動就是一個最常見最簡單的請求,沒有過多複雜的問題(比如多節點會不會重複執行這些)。最簡單的辦法,當用戶請求過來直接用一個線程或者線程池執行用戶點的那個任務的邏輯代碼就行了,當然我這裏沒有那麼粗暴,現有的調度代碼資源如下:

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 創建任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 可以明確知道最多只會運行2個線程,直接使用系統自帶工具就可以了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工作線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池,初始線程數量corePoolSize,線程池等待隊列大小queueSize,當初始線程都有任務,並且等待隊列滿后
         * 線程數量會自動擴充最大線程數maxSize,當新擴充的線程空閑60s后自動回收.自定義線程池是因為Executors那幾個線程工具
         * 各有各的弊端,不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)才開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不該自己拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態,如果更新成功,其他節點就不會更新成功。如果其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務,version必然和查出來時候的version不一樣了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是因為上面樂觀鎖已經更新了版本,會導致後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就可以從延時隊列拿出任務對象,然後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了,這裏沒必要了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完后刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務,如果父任務失敗了,子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改成執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態,不然這裏可能和恢複線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下,避免上面update修改后version不同步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務后再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

   
}

  上面主要就是三組線程,Loader負責加載將要執行的任務放入本地的任務隊列,Boss線程負責取出任務隊列的任務,然後分配Worker線程池的一個線程去執行。由上面的代碼可以看到如果要立即執行,其實只需要把一個延時為0的任務放入任務隊列,等着Boss線程去取然後分配給worker執行就可以實現了,代碼如下:

    /**
     * 立即執行任務,就是設置一下延時為0加入任務隊列就好了,這個可以外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

  啟動不用再多說,下面介紹一下停止任務,根據面向對象的思維,用戶要想停止一個任務,最終執行停止任務的就是正在執行任務的那個節點。停止任務有兩種情況,第一種任務沒有正在運行如何停止,第二種是任務正在運行如何停止。第一種其實直接改變一下任務對象的狀態為停止就行了,不必多說。下面主要考慮如何停止正在運行的任務,細心的朋友可能已經發現上面代碼和之前那一篇代碼有點區別,之前用的Runnble作為線程實現接口,這個用了Callable,其實在java中停止線程池中正在運行的線程最常用的就是直接調用future的cancel方法了,要想獲取到這個future對象就需要將以前實現Runnbale改成實現Callable,然後提交到線程池由execute改成submit就可以了,然後每次提交到線程池得到的future對象使用taskId一起保存在一個map中,方便根據taskId隨時找到。當然任務執行完后要及時刪除這個map里的任務,以免常駐其中導致內存溢出。停止任務的請求流程如下

  

 

 

  圖還是原來的圖,但是這時候情況不一樣了,因為停止任務的時候假如當前正在執行這個任務的節點處於服務1,負載均衡是不知道要去把你引到服務1的,他可能會引入到服務2,那就悲劇了,所以通用的做法就是停止請求過來不管落到哪個節點上,那個節點就往一個公用的mq上發一個帶有停止任務業務含義的消息,各個節點訂閱這個消息,然後判斷都判斷任務在不在自己這裏執行,如果在就執行停止操作。但是這樣勢必讓我們的調度服務又要依賴一個外部的消息隊列服務,就算很方便的就可以引入一個外部的消息隊列,但是你真的可以駕馭的了嗎,消息丟了咋辦,重複發送了咋辦,消息服務掛了咋辦,網絡斷了咋辦,又引入了一大堆問題,那我是不是又要寫n篇文章來分別解決這些問題。往往現實卻是就是這麼殘酷,你解決了一個問題,引入了更多的問題,這就是為什麼bug永遠改不完的道理了。當然這不是我的風格,我的風格是利用有限的資源做盡可能多的事情(可能是由於我工作的企業都是那種資源貧瘠的,養成了我這種習慣,土豪公司的程序員請繞道,哈哈)。

  簡化一下問題:目前的問題就是如何讓正在執行任務的節點知道,然後停止正在執行的這個任務,其實就是這個停止通知如何實現。這不免讓我想起了12306網站上買票,其實我們作為老百姓多麼希望12306可以在有票的時候發個短信通知一下我們,然後我們上去搶,但是現實卻是,你要麼使用軟件一直刷,要麼是自己隔一段時間上去瞄一下有沒有票。如果把有票了給我們發短信通知定義為異步通知,那麼這種我們要隔一段時間自己去瞄一下的方式就是同步輪訓。這兩種方式都能達到告知的目的,關鍵的區別在於你到底有沒有時間去一直去瞄,不過相比於可以回家,這些時間都是值得的。個人認為軟件的設計其實就是一個權衡是否值得的過程。如果約定了不使用外部消息隊列這種異步通知的方式,那麼我們只能使用同步輪訓的方式了。不過正好我們的任務調度本身已經有一個心跳機制,沒隔一段時間就去更新一下節點狀態,如果我們把用戶的停止請求作為命令信息更新到每個節點的上,然後隨着心跳獲取到這個節點的信息,然後判斷這個命令,做相應的處理是不是就可以完美解決這個問題。值得嗎?很明顯是值得的,我們只是在心跳邏輯上加一個小小的副作用就實現了通知功能了。代碼如下

package com.rdpaas.task.common;

/**
 * @author rongdi
 * @date 2019/11/26
 */
public enum NotifyCmd {

    //沒有通知,默認狀態
    NO_NOTIFY(0),
    //開啟任務(Task)
    START_TASK(1),
    //修改任務(Task)
    EDIT_TASK(2),
    //停止任務(Task)
    STOP_TASK(3);

    int id;

    NotifyCmd(int id) {
        this.id = id;
    }

    public int getId() {
        return id;
    }

    public static NotifyCmd valueOf(int id) {
        switch (id) {
            case 1:
                return START_TASK;
            case 2:
                return EDIT_TASK;
            case 3:
                return STOP_TASK;
            default:
                return NO_NOTIFY;
        }
    }

}
package com.rdpaas.task.handles;

import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.utils.SpringContextUtil;

/**
 * @author: rongdi
 * @date:
 */
public interface NotifyHandler<T> {

    static NotifyHandler chooseHandler(NotifyCmd notifyCmd) {
        return SpringContextUtil.getByTypeAndName(NotifyHandler.class,notifyCmd.toString());
    }

    public void update(T t);

}
package com.rdpaas.task.handles;

import com.rdpaas.task.scheduler.TaskExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author: rongdi
 * @date:
 */
@Component("STOP_TASK")
public class StopTaskHandler implements NotifyHandler<Long> {

    @Autowired
    private TaskExecutor taskExecutor;

    @Override
    public void update(Long taskId) {
        taskExecutor.stop(taskId);
    }

}
class HeartBeat implements Runnable {
        @Override
        public void run() {
            for(;;) {
                try {
                    /**
                     * 時間到了就可以從延時隊列拿出節點對象,然後更新時間和序號,
                     * 最後再新建一個超時時間為心跳時間的節點對象放入延時隊列,形成循環的心跳
                     */
                    DelayItem<Node> item = heartBeatQueue.take();
                    if(item != null && item.getItem() != null) {
                        Node node = item.getItem();
                        handHeartBeat(node);
                    }
                    heartBeatQueue.offer(new DelayItem<>(config.getHeartBeatSeconds() * 1000,new Node(config.getNodeId())));
                } catch (Exception e) {
                    logger.error("task heart beat error,cause by:{} ",e);
                }
            }
        }
    }

    /**
     * 處理節點心跳
     * @param node
     */
    private void handHeartBeat(Node node) {
        if(node == null) {
            return;
        }
        /**
         * 先看看數據庫是否存在這個節點
         * 如果不存在:先查找下一個序號,然後設置到node對象中,最後插入
         * 如果存在:直接根據nodeId更新當前節點的序號和時間
         */
        Node currNode= nodeRepository.getByNodeId(node.getNodeId());
        if(currNode == null) {
            node.setRownum(nodeRepository.getNextRownum());
            nodeRepository.insert(node);
        } else  {
            nodeRepository.updateHeartBeat(node.getNodeId());
            NotifyCmd cmd = currNode.getNotifyCmd();
            String notifyValue = currNode.getNotifyValue();
            if(cmd != null && cmd != NotifyCmd.NO_NOTIFY) {
                /**
                 * 藉助心跳做一下通知的事情,比如及時停止正在執行的任務
                 * 根據指令名稱查找Handler
                 */
                NotifyHandler handler = NotifyHandler.chooseHandler(currNode.getNotifyCmd());
                if(handler == null || StringUtils.isEmpty(notifyValue)) {
                    return;
                }
                /**
                 * 執行操作
                 */
                handler.update(Long.valueOf(notifyValue));
            }
            
        }


    }

  最終的任務調度代碼如下:

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 創建任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 可以明確知道最多只會運行2個線程,直接使用系統自帶工具就可以了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工作線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池,初始線程數量corePoolSize,線程池等待隊列大小queueSize,當初始線程都有任務,並且等待隊列滿后
         * 線程數量會自動擴充最大線程數maxSize,當新擴充的線程空閑60s后自動回收.自定義線程池是因為Executors那幾個線程工具
         * 各有各的弊端,不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)才開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不該自己拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態,如果更新成功,其他節點就不會更新成功。如果其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務,version必然和查出來時候的version不一樣了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是因為上面樂觀鎖已經更新了版本,會導致後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就可以從延時隊列拿出任務對象,然後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了,這裏沒必要了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完后刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務,如果父任務失敗了,子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改成執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態,不然這裏可能和恢複線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下,避免上面update修改后version不同步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務后再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

    /**
     * 立即執行任務,就是設置一下延時為0加入任務隊列就好了,這個可以外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

    /**
     * 立即停止正在執行的任務,留給外部調用的方法
     * @param taskId
     * @return
     */
    public boolean stopNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        if(task == null) {
            return false;
        }
        /**
         * 該任務不是正在執行,直接修改task狀態為已完成即可
         */
        if(task.getStatus() != TaskStatus.DOING) {
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
            return true;
        }
        /**
         * 該任務正在執行,使用節點配合心跳發布停用通知
         */
        int n = nodeRepository.updateNotifyInfo(NotifyCmd.STOP_TASK,String.valueOf(taskId));
        return n > 0;
    }

    /**
     * 立即停止正在執行的任務,這個不需要自己調用,是給心跳線程調用
     * @param taskId
     * @return
     */
    public boolean stop(Long taskId) {
        Task task = taskRepository.get(taskId);
        /**
         * 不是自己節點的任務,本節點不能執行停用
         */
        if(task == null || !config.getNodeId().equals(task.getNodeId())) {
            return false;
        }
        /**
         * 拿到正在執行任務的future,然後強制停用,並刪除doingFutures的任務
         */
        Future future = doingFutures.get(taskId);
        boolean flag =  future.cancel(true);
        if(flag) {
            doingFutures.remove(taskId);
            /**
             * 修改狀態為已停用
             */
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
        }
        /**
         * 重置通知信息,避免重複執行停用通知
         */
        nodeRepository.resetNotifyInfo(NotifyCmd.STOP_TASK);
        return flag;
    }
}

  好吧,其實實現很簡單,關鍵在於思路,不BB了,詳細代碼見: 在下告辭!

  

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

3c收購,鏡頭 收購有可能以全新價回收嗎?

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

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

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

賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

[apue] 神奇的 Solaris pipe

說到 pipe 大家可能都不陌生,經典的pipe調用配合fork進行父子進程通訊,簡直就是Unix程序的標配。

然而Solaris上的pipe卻和Solaris一樣是個奇葩(雖然Solaris前途黯淡,但是不妨礙我們從它裏面挖掘一些有價值的東西),

有着和一般pipe諸多的不同之處,本文就來說說Solaris上神奇的pipe和一般pipe之間的異同。

 

1.solaris pipe 是全雙工的

一般系統上的pipe調用是半雙工的,只能單向傳遞數據,如果需要雙向通訊,我們一般是建兩個pipe分別讀寫。像下面這樣:

 1     int n, fd1[2], fd2[2]; 
 2     if (pipe (fd1) < 0 || pipe(fd2) < 0)
 3         err_sys ("pipe error"); 
 4 
 5     char line[MAXLINE]; 
 6     pid_t pid = fork (); 
 7     if (pid < 0) 
 8         err_sys ("fork error"); 
 9     else if (pid > 0)
10     {
11         close (fd1[0]);  // write on pipe1 as stdin for co-process
12         close (fd2[1]);  // read on pipe2 as stdout for co-process
13         while (fgets (line, MAXLINE, stdin) != NULL) { 
14             n = strlen (line); 
15             if (write (fd1[1], line, n) != n)
16                 err_sys ("write error to pipe"); 
17             if ((n = read (fd2[0], line, MAXLINE)) < 0)
18                 err_sys ("read error from pipe"); 
19 
20             if (n == 0) { 
21                 err_msg ("child closed pipe"); 
22                 break;
23             }
24             line[n] = 0; 
25             if (fputs (line, stdout) == EOF)
26                 err_sys ("fputs error"); 
27         }
28 
29         if (ferror (stdin))
30             err_sys ("fputs error"); 
31 
32         return 0; 
33     }
34     else { 
35         close (fd1[1]); 
36         close (fd2[0]); 
37         if (fd1[0] != STDIN_FILENO) { 
38             if (dup2 (fd1[0], STDIN_FILENO) != STDIN_FILENO)
39                 err_sys ("dup2 error to stdin"); 
40             close (fd1[0]); 
41         }
42 
43         if (fd2[1] != STDOUT_FILENO) { 
44             if (dup2 (fd2[1], STDOUT_FILENO) != STDOUT_FILENO)
45                 err_sys ("dup2 error to stdout"); 
46             close (fd2[1]); 
47         }
48 
49         if (execl (argv[1], "add2", (char *)0) < 0)
50             err_sys ("execl error"); 
51     }

這個程序創建兩個管道,fd1用來寫請求,fd2用來讀應答;對子進程而言,fd1重定向到標準輸入,fd2重定向到標準輸出,讀取stdin中的數據相加然後寫入stdout完成工作。父進程在取得應答後向標準輸出寫入結果。

如果在Solaris上,可以直接用一個pipe同時讀寫,代碼可以重寫成這樣:

 1 int fd[2];
 2 if (pipe(fd) < 0) 
 3     err_sys("pipe error\n");
 4 
 5 char line[MAXLINE];
 6 pid_t pid = fork();
 7 if (pid < 0)
 8     err_sys("fork error\n");
 9 else if (pid > 0)
10 {
11     close(fd[1]);
12     while (fgets(line, MAXLINE, stdin) != NULL) {
13         n = strlen(line);
14         if (write(fd[0], line, n) != n)
15             err_sys("write error to pipe\n")
16         if ((n = read(fd[0], line, MAXLINE)) < 0) 
17             err_sys("read error from pipe\n");
18 
19         if (n == 0) 
20             err_sys("child closed pipe\n");
21         line[n] = 0;
22         if (fputs(line, stdout) == EOF) 
23             err_sys("fputs error\n");
24     }
25 
26     if (ferror(stdin))
27         err_sys("fputs error\n");
28 
29     return 0;
30 }
31 else {
32     close(fd[0]);
33     if (fd[1] != STDIN_FILENO)
34         if (dup2(fd[1], STDIN_FILENO) != STDIN_FILENO)
35             err_sys("dup2 error to stdin\n");
36 
37     if (fd[1] != STDOUT_FILENO) {
38         if (dup2(fd[1], STDOUT_FILENO) != STDOUT_FILENO)
39             err_sys("dup2 error to stdout\n");
40         close(fd[1]);
41     }
42 
43     if (execl(argv[1], argv[2], (char *)0) < 0)
44         err_sys("execl error\n");
45 
46 }

代碼清爽多了,不用去考慮fd1[0]和fd2[1]是啥意思是一件很養腦的事。

不過這樣的代碼只能在Solaris上運行(聽說BSD也支持?),如果考慮到可移植性,還是寫上面的比較穩妥。

 

測試程序

 

 

2. solaris pipe 可以脫離父子關係建立

pipe 好用但是沒法脫離fork使用,一般的pipe如果想讓任意兩個進程通訊,得藉助它的變身fifo來實現。

關於FIFO,詳情可參考我之前寫的一篇文章:

 

而Solaris上的pipe沒這麼多事,加入兩個調用:fattach / fdetach,你就可以像使用FIFO一樣使用pipe了:

 1 int fd[2];
 2 if (pipe(fd) < 0)
 3     err_sys("pipe error\n");
 4 
 5 if (fattach(fd[1], "./pipe") < 0)
 6     err_sys("fattach error\n");
 7 
 8 printf("attach to file pipe ok\n");
 9 
10 close(fd[1]);
11 char line[MAXLINE];
12 while (fgets(line, MAXLINE, stdin) != NULL) {
13     n = strlen(line);
14     if (write(fd[0], line, n) != n)
15         err_sys("write error to pipe\n");
16     if ((n = read(fd[0], line, MAXLINE)) < 0)
17         err_sys("read error from pipe\n");
18 
19     if (n == 0) 
20         err_sys("child closed pipe\n");
21 
22     line[n] = 0;
23     if (fputs(line, stdout) == EOF)
24         err_sys("fputs error\n");
25 }
26 
27 if (ferror(stdin))
28     err_sys("fputs error\n");
29 
30 if (fdetach("./pipe") < 0)
31     err_sys("fdetach error\n");
32 
33 printf("detach from file pipe ok\n");

在pipe調用之後立即加入fattach調用,可以將管道關聯到文件系統的一個文件名上,該文件必需事先存在,且可讀可寫。

在fattach調用之前這個文件(./pipe)是個普通文件,打開讀寫都是磁盤IO;

在fattach調用之後,這個文件就變身成為一個管道了,打開讀寫都是內存流操作,且管道的另一端就是attach的那個進程。

子進程也需要改造一下,以便使用pipe通訊:

 1 int fd, n, int1, int2;
 2 char line[MAXLINE];
 3 fd = open("./pipe", O_RDWR);
 4 if (fd < 0)
 5     err_sys("open file pipe failed\n");
 6 
 7 printf("open file pipe ok, fd = %d\n", fd);
 8 while ((n = read(fd, line, MAXLINE)) > 0) {
 9     line[n] = 0;
10     if (sscanf(line, "%d%d", &int1, &int2) == 2) {
11         sprintf(line, "%d\n", int1 + int2);
12         n = strlen(line);
13         if (write(fd, line, n) != n)
14             err_sys("write error\n");
15 
16         printf("i am working on %s\n", line);
17     }
18     else {
19         if (write(fd, "invalid args\n", 13) != 13)
20             err_sys("write msg error\n");
21     }
22 }
23 
24 close(fd);

打開pipe就如同打開普通文件一樣,open直接搞定。當然前提是attach進程必需已經在運行。

當attach進程detach后,管道文件又將恢復它的本來面目。

 

脫離了父子關係的pipe其實可以建立多對一關係(多對多關係不可以,因為只能有一個進程attach)。

例如開4個cmd窗口,分別執行以下命令:

./padd2 abc
./add2
./add2
./add2

 向attach進程(padd2)發送9個計算請求后,可以看到輸出結果如下:

-bash-3.2$ ./padd2 abc
attach to file pipe ok
1 1
2
2 2
4
3 3 
6
4 4
8
5 5
10
6 6 
12
7 7 
14
8 8
16
9 9
18

 再回來看各個open管道的進程,輸出分別如下:

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 1 1
i am working on 2
source: 4 4
i am working on 8
source: 7 7 
i am working on 14 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 2 2
i am working on 4
source: 5 5
i am working on 10
source: 9 9
i am working on 18 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 2 2
i am working on 4
source: 5 5
i am working on 10
source: 9 9
i am working on 18 

 

-bash-3.2$ ./add2
open file pipe ok, fd = 3
source: 3 3
i am working on 6
source: 6 6
i am working on 12
source: 8 8 
i am working on 16

 

可以發現一個很有趣的現象,就是各個add2進程基本是輪着來獲取請求的,可以猜想底層的pipe可能有一個進程排隊機制。

但是反過來使用pipe就不行了。就是說當啟動一個add3(區別於上例的add2與padd2)作為fattach端打開pipe,啟動多個padd3作為open端使用pipe,

然後通過命令行給padd3傳遞要相加的值,可以寫一個腳本同時啟動多個padd3,來查看效果:

#! /bin/sh
./padd3 1 1 &
./padd3 2 2 &
./padd3 3 3 &
./padd3 4 4 &

 這個腳本中啟動了4個加法進程,同時向add3發送4個加法請求,腳本中四個進程輸出如下:

-bash-3.2$ ./padd3.sh
-bash-3.2$ open file pipe ok, fd = 3
1 1 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
open file pipe ok, fd = 3
4 4 = 37

 可以看到3+3的請求被忽略了,轉到add3查看輸出:

-bash-3.2$ ./add3
attach to file pipe ok
source: 1 1
i am working on 1 + 1 = 2
source: 2 2
i am working on 2 + 2 = 4
source: 3 34 4
i am working on 3 + 34 = 37

 原來是3+3與4+4兩個請求粘連了,導致add3識別成一個3+34的請求,所以出錯了。

多運行幾遍腳本后,發現還有這樣的輸出:

-bash-3.2$ ./padd3.sh
-bash-3.2$ open file pipe ok, fd = 3
4 4 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
3 3 = 6
open file pipe ok, fd = 3
1 1 = 8

  4+4=2?1+1=8?再看add3這頭的輸出:

-bash-3.2$ ./add3
attach to file pipe ok
source: 1 1
i am working on 1 + 1 = 2
source: 2 2
i am working on 2 + 2 = 4
source: 3 3
i am working on 3 + 3 = 6
source: 4 4
i am working on 4 + 4 = 8

 完全正常呢。

經過一番推理,發現是4+4的請求取得了1+1請求的應答;1+1的請求取得了4+4的應答。

可見這樣的結構還有一個弊端,同時請求的進程可能無法得到自己的應答,應答與請求之間相互錯位。

所以想用fattach來實現多路請求的人還是洗洗睡吧,畢竟它就是一個pipe不是,還能給它整成tcp么?

而之前的例子可以,是因為請求是順序發送的,上個請求得到應答后才發送下個請求,所以不存在這個例子的問題(但是實用性也不高)。

 

測試程序

 

 

3. solaris pipe 可以通過connld模塊實現類似tcp的多路連接

第2條剛說不能實現多路連接,第3條就接着來打臉了,這是由於Solaris上的pipe都是基於STREAMS技術構建,

而STREAMS是支持靈活的PUSH、POP流處理模塊的,再加上STREAMS傳遞文件fd的能力,就可以支持類似tcp中accept的能力。

即每個open pipe文件的進程,得到的不是原來管道的fd,而是新創建管道的fd,而管道的另一側fd則通過已有的管道發送到attach進程,

後者使用這個新的fd與客戶進程通訊。為了支持多路連接,我們的代碼需要重新整理一下,首先看客戶端:

1 int fd;
2 char line[MAXLINE];
3 fd = cli_conn("./pipe");
4 if (fd < 0)
5     return 0;

這裏將open相關邏輯封裝到了cli_conn函數中,以便之後復用:

 1 int cli_conn(const char *name)
 2 {
 3     int fd;
 4     if ((fd = open(name, O_RDWR)) < 0) {
 5         printf("open pipe file failed\n");
 6         return -1;
 7     }
 8 
 9     if (isastream(fd) == 0) {
10         close(fd);
11         return -2;
12     }
13 
14     return fd;
15 }

可以看到與之前幾乎沒有變化,只是增加了isastream調用防止attach進程沒有啟動。

再來看下服務端:

 1 int listenfd = serv_listen("./pipe");
 2 if (listenfd < 0)
 3     return 0;
 4 
 5 int acceptfd = 0;
 6 int n = 0, int1 = 0, int2 = 0;
 7 char line[MAXLINE];
 8 uid_t uid = 0;
 9 while ((acceptfd = serv_accept(listenfd, &uid)) >= 0)
10 {
11     printf("accept a client, fd = %d, uid = %ld\n", acceptfd, uid);
12     while ((n = read(acceptfd, line, MAXLINE)) > 0) {
13         line[n] = 0;
14         printf("source: %s\n", line);
15         if (sscanf(line, "%d%d", &int1, &int2) == 2) {
16             sprintf(line, "%d\n", int1 + int2);
17             n = strlen(line);
18             if (write(acceptfd, line, n) != n) {
19                 printf("write error\n");
20                 return 0;
21             }
22             printf("i am working on %d + %d = %s\n", int1, int2, line);
23         }
24         else {
25             if (write(acceptfd, "invalid args\n", 13) != 13) {
26                 printf("write msg error\n");
27                 return 0;
28             }
29         }
30     }
31 
32     close(acceptfd);
33 }
34 
35 if (fdetach("./pipe") < 0) {
36     printf("fdetach error\n");
37     return 0;
38 }
39 
40 printf("detach from file pipe ok\n");
41 close(listenfd);

首先調用serv_listen建立基本pipe,然後不斷在該pipe上調用serv_accept來獲取獨立的客戶端連接。之後的邏輯與以前一樣。

現在重點看下封裝的這兩個方法:

 1 int serv_listen(const char *name)
 2 {
 3     int tempfd;
 4     int fd[2];
 5     unlink(name);
 6     tempfd = creat(name, FIFO_MODE);
 7     if (tempfd < 0) {
 8         printf("creat failed\n");
 9         return -1;
10     }
11 
12     if (close(tempfd) < 0) {
13         printf("close temp fd failed\n");
14         return -2;
15     }
16 
17     if (pipe(fd) < 0) {
18         printf("pipe error\n");
19         return -3;
20     }
21 
22     if (ioctl(fd[1], I_PUSH, "connld") < 0) {
23         printf("I_PUSH connld failed\n");
24         close(fd[0]);
25         close(fd[1]);
26         return -4;
27     }
28 
29     printf("push connld ok\n");
30     if (fattach(fd[1], name) < 0) {
31         printf("fattach error\n");
32         close(fd[0]);
33         close(fd[1]);
34         return -5;
35     }
36 
37     printf("attach to file pipe ok\n");
38     close(fd[1]);
39     return fd[0];
40 }

serv_listen封裝了與建立基本pipe相關的代碼,首先確保pipe文件存在且可讀寫,然後創建普通的pipe,在fattach調用之前必需先PUSH一個connld模塊到該pipe STREAM中。這樣就大功告成!

 1 int serv_accept(int listenfd, uid_t *uidptr)
 2 {
 3     struct strrecvfd recvfd;
 4     if (ioctl(listenfd, I_RECVFD, &recvfd) < 0) {
 5         printf("I_RECVFD from listen fd failed\n");
 6         return -1;
 7     }
 8 
 9     if (uidptr)
10         *uidptr = recvfd.uid;
11 
12     return recvfd.fd;
13 }

當有客戶端連接上來的時候,使用I_RECVFD接收connld返回的另一個pipe的fd。之後的數據將在該pipe進行。

看了看,感覺和tcp的listen與accept別無二致,看來天下武功,至精深處都是英雄所見略同。

之前的多個客戶端同時運行的例子再跑一遍,觀察attach端輸出:

-bash-3.2$ ./add4
push connld ok
attach to file pipe ok
accept a client, fd = 4, uid = 101
source: 1 1
i am working on 1 + 1 = 2
accept a client, fd = 4, uid = 101
source: 2 2
i am working on 2 + 2 = 4
accept a client, fd = 4, uid = 101
source: 3 3
i am working on 3 + 3 = 6
accept a client, fd = 4, uid = 101
source: 4 4
i am working on 4 + 4 = 8

 一切正常。再看下腳本中四個進程的輸出:

-bash-3.2$ ./padd4.sh
-bash-3.2$ open file pipe ok, fd = 3
1 1 = 2
open file pipe ok, fd = 3
2 2 = 4
open file pipe ok, fd = 3
3 3 = 6
open file pipe ok, fd = 3
4 4 = 8

 也是沒問題的,既沒有出現多個請求粘連的情況,也沒有出現請求與應答錯位的情況。

 

測試程序

 

 

4.結論

Solaris 上的pipe不僅可以全雙工通訊、不依賴父子進程關係,還可以實現類似tcp那樣分離多個客戶端通訊連接的能力。

雖然Solaris前途未卜,但是希望一些好的東西還是能流傳下來,就比如這個神奇的pipe。

 

看完今天的文章,你是不是對特立獨行的Solaris又加深了一層了解?歡迎留言區說說你認識的Solaris。

 

 

 

 

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

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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

嵌入式、C語言位操作的一些技巧匯總

下面分享關於位操作的一些筆記:

一、位操作簡單介紹

首先,以下是按位運算符:

嵌入式編程中,常常需要對一些寄存器進行配置,有的情況下需要改變一個字節中的某一位或者幾位,但是又不想改變其它位原有的值,這時就可以使用按位運算符進行操作。下面進行舉例說明,假如有一個8位的TEST寄存器:

當我們要設置第0位bit0的值為1時,可能會這樣進行設置:

TEST = 0x01;

但是,這樣設置是不夠準確的,因為這時候已經同時操作到了高7位:bit1~bit7,如果這高7位沒有用到的話,這麼設置沒有什麼影響;但是,如果這7位正在被使用,結果就不是我們想要的了。

在這種情況下,我們就可以借用按位操作運算符進行配置。

對於二進制位操作來說,不管該位原來的值是0還是1,它跟0進行&運算,得到的結果都是0,而跟1進行&運算,將保持原來的值不變;不管該位原來的值是0還是1,它跟1進行|運算,得到的結果都是1,而跟0進行|運算,將保持原來的值不變。

所以,此時可以設置為:

TEST = TEST | 0x01;

其意義為:TEST寄存器的高7位均不變,最低位變成1了。在實際編程中,常改寫為:

TEST |= 0x01;

這種寫法可以一定程度上簡化代碼,是 C 語言常用的一種編程風格。設置寄存器的某一位還有另一種操作方法,以上的等價方法如:

TEST |= (0x01 << 0);

第幾位要置1就左移幾位。

同樣的,要給TEST的低4位清0,高4位保持不變,可以進行如下配置:

TEST &= 0xF0;

二、嵌入式中位操作一些常見用法

1、一個32bit數據的位、字節讀取操作

(1)獲取單字節:

#define GET_LOW_BYTE0(x)    ((x >>  0) & 0x000000ff)    /* 獲取第0個字節 */
#define GET_LOW_BYTE1(x)    ((x >>  8) & 0x000000ff)    /* 獲取第1個字節 */
#define GET_LOW_BYTE2(x)    ((x >> 16) & 0x000000ff)    /* 獲取第2個字節 */
#define GET_LOW_BYTE3(x)    ((x >> 24) & 0x000000ff)    /* 獲取第3個字節 */

示例:

(2)獲取某一位:

#define GET_BIT(x, bit) ((x & (1 << bit)) >> bit)   /* 獲取第bit位 */

示例:

2、一個32bit數據的位、字節清零操作

(1)清零某個字節:

#define CLEAR_LOW_BYTE0(x)  (x &= 0xffffff00)   /* 清零第0個字節 */
#define CLEAR_LOW_BYTE1(x)  (x &= 0xffff00ff)   /* 清零第1個字節 */
#define CLEAR_LOW_BYTE2(x)  (x &= 0xff00ffff)   /* 清零第2個字節 */
#define CLEAR_LOW_BYTE3(x)  (x &= 0x00ffffff)   /* 清零第3個字節 */

示例:

(2)清零某一位:

#define CLEAR_BIT(x, bit)   (x &= ~(1 << bit))  /* 清零第bit位 */

示例:

3、一個32bit數據的位、字節置1操作

(1)置某個字節為1:

#define SET_LOW_BYTE0(x)    (x |= 0x000000ff)   /* 第0個字節置1 */   
#define SET_LOW_BYTE1(x)    (x |= 0x0000ff00)   /* 第1個字節置1 */   
#define SET_LOW_BYTE2(x)    (x |= 0x00ff0000)   /* 第2個字節置1 */   
#define SET_LOW_BYTE3(x)    (x |= 0xff000000)   /* 第3個字節置1 */

示例:

(2)置位某一位:

#define SET_BIT(x, bit) (x |= (1 << bit))   /* 置位第bit位 */

4、判斷某一位或某幾位連續位的值

(1)判斷某一位的值

舉例說明:判斷0x68第3位的值。

也就是說,要判斷第幾位的值,if里就左移幾位(當然別過頭了)。在嵌入式編程中,可通過這樣的方式來判斷寄存器的狀態位是否被置位。

(2)判斷某幾位連續位的值

/* 獲取第[n:m]位的值 */
#define BIT_M_TO_N(x, m, n)  ((unsigned int)(x << (31-(n))) >> ((31 - (n)) + (m)))

示例:

這是一個查詢連續狀態位的例子,因為有些情況不止有0、1兩種狀態,可能會有多種狀態,這種情況下就可以用這種方法來取出狀態位,再去執行相應操作。

以上是對32bit數據的一些操作進行總結,其它位數的數據類似,可根據需要進行修改。

三、STM32寄存器配置

STM32有幾套固件庫,這些固件庫函數以函數的形式進行1層或者多層封裝(軟件開發中很重要的思想之一:分層思想),但是到了最裏面的一層就是對寄存器的配置。我們平時都比較喜歡固件庫來開發,大概是因為固件庫用起來比較簡單,用固件庫寫出來的代碼比較容易閱讀。最近一段時間一直在配置寄存器,越發地發現使用寄存器來進行一些外設的配置也是很容易懂的。使用寄存器的方式編程無非就是往寄存器的某些位置1、清零以及對寄存器一些狀態位進行判斷、讀取寄存器的內容等。

這些基本操作在上面的例子中已經有介紹,我們依舊以實例來鞏固上面的知識點(以STM32F1xx為例):

(1)寄存器配置

看一下GPIO功能的端口輸出數據寄存器 (GPIOx_ODR) (x=A..E) :

假設我們要讓PA10引腳輸出高、輸出低,可以這麼做:

方法一:

GPIOA->ODR |= 1 << 10;      /* PA10輸出高(置1操作) */
GPIOA->ODR &= ~(1 << 10);  /* PA10輸出低(清0操作) */

也可用我們上面的置位、清零的宏定義:

SET_BIT(GPIOA->ODR, 10);    /* PA10輸出高(置1操作) */
CLEAR_BIT(GPIOA->ODR, 10);  /* PA10輸出低(清0操作) */

方法二:

GPIOA->ODR |= (uint16_t)0x0400;   /* PA10輸出高(置1操作) */
GPIOA->ODR &= ~(uint16_t)0x0400;  /* PA10輸出低(清0操作) */

貌似第二種方法更麻煩?還得去細心地去構造一個數據。

但是,其實第二種方法其實是ST推薦我們用的方法,為什麼這麼說呢?因為ST官方已經把這些我們要用到的值給我們配好了,在stm32f10x.h中:

這個頭文件中存放的就是外設寄存器的一些位配置。

所以我們的方法二等價於:

GPIOA->ODR |= GPIO_ODR_ODR10;   /* PA10輸出高(置1操作) */
GPIOA->ODR &= ~GPIO_ODR_ODR10;  /* PA10輸出低(清0操作) */

兩種方法都是很好的方法,但方法一似乎更好理解。

配置連續幾位的方法也是一樣的,就不介紹了。簡單介紹配置不連續位的方法,以TIM1的CR1寄存器為例:

設置CEN位為1、設置CMS[1:0]位為01、設置CKD[1:0]位為10:

TIM1->CR1 |= (0x1 << 1)| (0x1 << 5) |(0x2 << 8);

這是組合的寫法。當然,像上面一樣拆開來寫也是可以的。

(2)判斷標誌位

以狀態寄存器(USART_SR) 為例:

判斷RXNE是否被置位:

/* 數據寄存器非空,RXNE標誌置位 */
if (USART1->SR & (1 << 5))
{
    /* 其它代碼 */
    
    USART1->SR &= ~(1 << 5);  /* 清零RXNE標誌 */
}

或者:

/* 數據寄存器非空,RXNE標誌置位 */
if (USART1->SR & USART_SR_RXNE)
{
    /* 其它代碼 */
    
    USART1->SR &= ~USART_SR_RXNE;  /* 清零RXNE標誌 */
}

四、總結

以上就是本次關於位操作的一點總結筆記,有必要掌握。雖然說在用STM32的時候有庫函數可以用,但是最接近芯片內部原理的還是寄存器。有可能之後有用到其它芯片沒有像ST這樣把寄存器相關配置封裝得那麼好,那就不得不直接操控寄存器了。

此外,使用庫函數的方式代碼佔用空間大,用寄存器的話,代碼佔用空間小。之前有個需求,我們能用的Flash的空間大小隻有4KB,遇到類似這樣的情況就不能那麼隨性的用庫函數了。

最後,應用的時候當然是怎麼簡單就怎麼用。學從“難”處學,用從易處用,與君共勉~

END:以上筆記中如有錯誤,歡迎指出!謝謝

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

3c收購,鏡頭 收購有可能以全新價回收嗎?

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

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

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

賣IPHONE,iPhone回收,舊換新!教你怎麼賣才划算?

[ch02-03] 梯度下降

系列博客,原文在筆者所維護的github上:,
點擊star加星不要吝嗇,星越多筆者越努力。

2.3 梯度下降

2.3.1 從自然現象中理解梯度下降

在大多數文章中,都以“一個人被困在山上,需要迅速下到谷底”來舉例,這個人會“尋找當前所處位置最陡峭的地方向下走”。這個例子中忽略了安全因素,這個人不可能沿着最陡峭的方向走,要考慮坡度。

在自然界中,梯度下降的最好例子,就是泉水下山的過程:

  1. 水受重力影響,會在當前位置,沿着最陡峭的方向流動,有時會形成瀑布(梯度下降);
  2. 水流下山的路徑不是唯一的,在同一個地點,有可能有多個位置具有同樣的陡峭程度,而造成了分流(可以得到多個解);
  3. 遇到坑窪地區,有可能形成湖泊,而終止下山過程(不能得到全局最優解,而是局部最優解)。

2.3.2 梯度下降的數學理解

梯度下降的數學公式:

\[\theta_{n+1} = \theta_{n} – \eta \cdot \nabla J(\theta) \tag{1}\]

其中:

  • \(\theta_{n+1}\):下一個值;
  • \(\theta_n\):當前值;
  • \(-\):減號,梯度的反向;
  • \(\eta\):學習率或步長,控制每一步走的距離,不要太快以免錯過了最佳景點,不要太慢以免時間太長;
  • \(\nabla\):梯度,函數當前位置的最快上升點;
  • \(J(\theta)\):函數。

梯度下降的三要素

  1. 當前點;
  2. 方向;
  3. 步長。

為什麼說是“梯度下降”?

“梯度下降”包含了兩層含義:

  1. 梯度:函數當前位置的最快上升點;
  2. 下降:與導數相反的方向,用數學語言描述就是那個減號。

亦即與上升相反的方向運動,就是下降。

圖2-9 梯度下降的步驟

圖2-9解釋了在函數極值點的兩側做梯度下降的計算過程,梯度下降的目的就是使得x值向極值點逼近。

2.3.3 單變量函數的梯度下降

假設一個單變量函數:

\[J(x) = x ^2\]

我們的目的是找到該函數的最小值,於是計算其微分:

\[J'(x) = 2x\]

假設初始位置為:

\[x_0=1.2\]

假設學習率:

\[\eta = 0.3\]

根據公式(1),迭代公式:

\[x_{n+1} = x_{n} – \eta \cdot \nabla J(x)= x_{n} – \eta \cdot 2x\tag{1}\]

假設終止條件為J(x)<1e-2,迭代過程是:

x=0.480000, y=0.230400
x=0.192000, y=0.036864
x=0.076800, y=0.005898
x=0.030720, y=0.000944

上面的過程如圖2-10所示。

圖2-10 使用梯度下降法迭代的過程

2.3.4 雙變量的梯度下降

假設一個雙變量函數:

\[J(x,y) = x^2 + \sin^2(y)\]

我們的目的是找到該函數的最小值,於是計算其微分:

\[{\partial{J(x,y)} \over \partial{x}} = 2x\]
\[{\partial{J(x,y)} \over \partial{y}} = 2 \sin y \cos y\]

假設初始位置為:

\[(x_0,y_0)=(3,1)\]

假設學習率:

\[\eta = 0.1\]

根據公式(1),迭代過程是的計算公式:
\[(x_{n+1},y_{n+1}) = (x_n,y_n) – \eta \cdot \nabla J(x,y)\]
\[ = (x_n,y_n) – \eta \cdot (2x,2 \cdot \sin y \cdot \cos y) \tag{1}\]

根據公式(1),假設終止條件為\(J(x,y)<1e-2\),迭代過程如表2-3所示。

表2-3 雙變量梯度下降的迭代過程

迭代次數 x y J(x,y)
1 3 1 9.708073
2 2.4 0.909070 6.382415
15 0.105553 0.063481 0.015166
16 0.084442 0.050819 0.009711

迭代16次后,J(x,y)的值為0.009711,滿足小於1e-2的條件,停止迭代。

上面的過程如表2-4所示,由於是雙變量,所以需要用三維圖來解釋。請注意看兩張圖中間那條隱隱的黑色線,表示梯度下降的過程,從紅色的高地一直沿着坡度向下走,直到藍色的窪地。

表2-4 在三維空間內的梯度下降過程

觀察角度1 觀察角度2

2.3.5 學習率η的選擇

在公式表達時,學習率被表示為\(\eta\)。在代碼里,我們把學習率定義為learning_rate,或者eta。針對上面的例子,試驗不同的學習率對迭代情況的影響,如表2-5所示。

表2-5 不同學習率對迭代情況的影響

學習率 迭代路線圖 說明
1.0 學習率太大,迭代的情況很糟糕,在一條水平線上跳來跳去,永遠也不能下降。
0.8 學習率大,會有這種左右跳躍的情況發生,這不利於神經網絡的訓練。
0.4 學習率合適,損失值會從單側下降,4步以後基本接近了理想值。
0.1 學習率較小,損失值會從單側下降,但下降速度非常慢,10步了還沒有到達理想狀態。

代碼位置

ch02, Level3, Level4, Level5

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

收購3c,收購IPHONE,收購蘋果電腦-詳細收購流程一覽表

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

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

※公開收購3c價格,不怕被賤賣!

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