之前的系列文章從 CPU 和內存方面簡單介紹了一下彙編語言，但是還沒有系統的了解一下彙編語言，彙編語言作為第二代計算機語言，會用一些容易理解和記憶的字母，單詞來代替一個特定的指令，作為高級編程語言的基礎，有必要系統的了解一下彙編語言，那麼本篇文章希望大家跟我一起來了解一下彙編語言。

彙編語言和本地代碼

我們在之前的文章中探討過，計算機 CPU 只能運行本地代碼(機器語言)程序，用 C 語言等高級語言編寫的代碼，需要經過編譯器編譯后，轉換為本地代碼才能夠被 CPU 解釋執行。

但是本地代碼的可讀性非常差，所以需要使用一種能夠直接讀懂的語言來替換本地代碼，那就是在各本地代碼中，附帶上表示其功能的英文縮寫，比如在加法運算的本地代碼加上add(addition) 的縮寫、在比較運算符的本地代碼中加上cmp(compare)的縮寫等，這些通過縮寫來表示具體本地代碼指令的標誌稱為 助記符，使用助記符的語言稱為彙編語言。這樣，通過閱讀彙編語言，也能夠了解本地代碼的含義了。

不過，即使是使用彙編語言編寫的源代碼，最終也必須要轉換為本地代碼才能夠運行，負責做這項工作的程序稱為編譯器，轉換的這個過程稱為彙編。在將源代碼轉換為本地代碼這個功能方面，彙編器和編譯器是同樣的。

用彙編語言編寫的源代碼和本地代碼是一一對應的。因而，本地代碼也可以反過來轉換成彙編語言編寫的代碼。把本地代碼轉換為彙編代碼的這一過程稱為反彙編，執行反彙編的程序稱為反彙編程序。

哪怕是 C 語言編寫的源代碼，編譯后也會轉換成特定 CPU 用的本地代碼。而將其反彙編的話，就可以得到彙編語言的源代碼，並對其內容進行調查。不過，本地代碼變成 C 語言源代碼的反編譯，要比本地代碼轉換成彙編代碼的反彙編要困難，這是因為，C 語言代碼和本地代碼不是一一對應的關係。

通過編譯器輸出彙編語言的源代碼

我們上面提到本地代碼可以經過反彙編轉換成為彙編代碼，但是只有這一種轉換方式嗎？顯然不是，C 語言編寫的源代碼也能夠通過編譯器編譯稱為彙編代碼，下面就來嘗試一下。

首先需要先做一些準備，需要先下載 Borland C++ 5.5 編譯器，為了方便，我這邊直接下載好了讀者直接從我的百度網盤提取即可 （鏈接:https://pan.baidu.com/s/19LqVICpn5GcV88thD2AnlA 密碼:hz1u）

下載完畢，需要進行配置，下面是配置說明 （https://wenku.baidu.com/view/22e2f418650e52ea551898ad.html），教程很完整跟着配置就可以，下面開始我們的編譯過程

首先用 Windows 記事本等文本編輯器編寫如下代碼

// 返回兩個參數值之和的函數
int AddNum(int a,int b){
  return a + b;
}

// 調用 AddNum 函數的函數
void MyFunc(){
  int c;
  c = AddNum(123,456);
}

編寫完成后將其文件名保存為 Sample4.c ，C 語言源文件的擴展名，通常用.c 來表示，上面程序是提供兩個輸入參數並返回它們之和。

在 Windows 操作系統下打開 命令提示符，切換到保存 Sample4.c 的文件夾下，然後在命令提示符中輸入

bcc32 -c -S Sample4.c

bcc32 是啟動 Borland C++ 的命令，-c 的選項是指僅進行編譯而不進行鏈接，-S 選項被用來指定生成彙編語言的源代碼

作為編譯的結果，當前目錄下會生成一個名為Sample4.asm 的彙編語言源代碼。彙編語言源文件的擴展名，通常用.asm 來表示，下面就讓我們用編輯器打開看一下 Sample4.asm 中的內容

    .386p
    ifdef ??version
    if    ??version GT 500H
    .mmx
    endif
    endif
    model flat
    ifndef  ??version
    ?debug  macro
    endm
    endif
    ?debug  S "Sample4.c"
    ?debug  T "Sample4.c"
_TEXT   segment dword public use32 'CODE'
_TEXT   ends
_DATA   segment dword public use32 'DATA'
_DATA   ends
_BSS    segment dword public use32 'BSS'
_BSS    ends
DGROUP  group   _BSS,_DATA
_TEXT   segment dword public use32 'CODE'
_AddNum proc    near
?live1@0:
   ;    
   ;    int AddNum(int a,int b){
   ;    
    push      ebp
    mov       ebp,esp
   ;    
   ;    
   ;        return a + b;
   ;    
@1:
    mov       eax,dword ptr [ebp+8]
    add       eax,dword ptr [ebp+12]
   ;    
   ;    }
   ;    
@3:
@2:
    pop       ebp
    ret 
_AddNum endp
_MyFunc proc    near
?live1@48:
   ;    
   ;    void MyFunc(){
   ;    
    push      ebp
    mov       ebp,esp
   ;    
   ;        int c;
   ;        c = AddNum(123,456);
   ;    
@4:
    push      456
    push      123
    call      _AddNum
    add       esp,8
   ;    
   ;    }
   ;    
@5:
    pop       ebp
    ret 
_MyFunc endp
_TEXT   ends
    public  _AddNum
    public  _MyFunc
    ?debug  D "Sample4.c" 20343 45835
    end

這樣，編譯器就成功的把 C 語言轉換成為了彙編代碼了。

不會轉換成本地代碼的偽指令

第一次看到彙編代碼的讀者可能感覺起來比較難，不過實際上其實比較簡單，而且可能比 C 語言還要簡單，為了便於閱讀彙編代碼的源代碼，需要注意幾個要點

彙編語言的源代碼，是由轉換成本地代碼的指令（後面講述的操作碼）和針對彙編器的偽指令構成的。偽指令負責把程序的構造以及彙編的方法指示給彙編器（轉換程序）。不過偽指令是無法彙編轉換成為本地代碼的。下面是上面程序截取的偽指令

_TEXT   segment dword public use32 'CODE'
_TEXT   ends
_DATA   segment dword public use32 'DATA'
_DATA   ends
_BSS    segment dword public use32 'BSS'
_BSS    ends
DGROUP  group   _BSS,_DATA

_AddNum proc    near
_AddNum endp

_MyFunc proc    near
_MyFunc endp

_TEXT   ends
    end

由偽指令 segment 和 ends 圍起來的部分，是給構成程序的命令和數據的集合體上加一個名字而得到的，稱為段定義。段定義的英文表達具有區域的意思，在這個程序中，段定義指的是命令和數據等程序的集合體的意思，一個程序由多個段定義構成。

上面代碼的開始位置，定義了3個名稱分別為 _TEXT、_DATA、_BSS 的段定義，_TEXT 是指定的段定義，_DATA 是被初始化（有初始值）的數據的段定義，_BSS 是尚未初始化的數據的段定義。這種定義的名稱是由 Borland C++ 定義的，是由 Borland C++ 編譯器自動分配的，所以程序段定義的順序就成為了 _TEXT、_DATA、_BSS ，這樣也確保了內存的連續性

_TEXT   segment dword public use32 'CODE'
_TEXT   ends
_DATA   segment dword public use32 'DATA'
_DATA   ends
_BSS    segment dword public use32 'BSS'
_BSS    ends

段定義( segment ) 是用來區分或者劃分範圍區域的意思。彙編語言的 segment 偽指令表示段定義的起始，ends 偽指令表示段定義的結束。段定義是一段連續的內存空間

而group 這個偽指令表示的是將 _BSS和_DATA 這兩個段定義匯總名為 DGROUP 的組

DGROUP  group   _BSS,_DATA

圍起 _AddNum 和 _MyFun 的 _TEXT segment 和 _TEXT ends ，表示_AddNum 和 _MyFun 是屬於 _TEXT 這一段定義的。

_TEXT   segment dword public use32 'CODE'
_TEXT   ends

因此，即使在源代碼中指令和數據是混雜編寫的，經過編譯和彙編后，也會轉換成為規整的本地代碼。

_AddNum proc 和 _AddNum endp 圍起來的部分，以及_MyFunc proc 和 _MyFunc endp 圍起來的部分，分別表示 AddNum 函數和 MyFunc 函數的範圍。

_AddNum proc    near
_AddNum endp

_MyFunc proc    near
_MyFunc endp

編譯后在函數名前附帶上下劃線_ ，是 Borland C++ 的規定。在 C 語言中編寫的 AddNum 函數，在內部是以 _AddNum 這個名稱處理的。偽指令 proc 和 endp 圍起來的部分，表示的是 過程(procedure) 的範圍。在彙編語言中，這種相當於 C 語言的函數的形式稱為過程。

末尾的 end 偽指令，表示的是源代碼的結束。

## 彙編語言的語法是 操作碼 + 操作數

在彙編語言中，一行表示一對 CPU 的一個指令。彙編語言指令的語法結構是 操作碼 + 操作數，也存在只有操作碼沒有操作數的指令。

操作碼錶示的是指令動作，操作數表示的是指令對象。操作碼和操作數一起使用就是一個英文指令。比如從英語語法來分析的話，操作碼是動詞，操作數是賓語。比如這個句子 Give me money這個英文指令的話，Give 就是操作碼，me 和 money 就是操作數。彙編語言中存在多個操作數的情況，要用逗號把它們分割，就像是 Give me,money 這樣。

能夠使用何種形式的操作碼，是由 CPU 的種類決定的，下面對操作碼的功能進行了整理。

本地代碼需要加載到內存后才能運行，內存中存儲着構成本地代碼的指令和數據。程序運行時，CPU會從內存中把數據和指令讀出來，然後放在 CPU 內部的寄存器中進行處理。

如果 CPU 和內存的關係你還不是很了解的話，請閱讀作者的另一篇文章 詳細了解。

寄存器是 CPU 中的存儲區域，寄存器除了具有臨時存儲和計算的功能之外，還具有運算功能，x86 系列的主要種類和角色如下圖所示

指令解析

下面就對 CPU 中的指令進行分析

最常用的 mov 指令

指令中最常使用的是對寄存器和內存進行數據存儲的 mov 指令，mov 指令的兩個操作數，分別用來指定數據的存儲地和讀出源。操作數中可以指定寄存器、常數、標籤(附加在地址前)，以及用方括號([]) 圍起來的這些內容。如果指定了沒有用([]) 方括號圍起來的內容，就表示對該值進行處理；如果指定了用方括號圍起來的內容，方括號的值則會被解釋為內存地址，然後就會對該內存地址對應的值進行讀寫操作。讓我們對上面的代碼片段進行說明

    mov       ebp,esp
    mov       eax,dword ptr [ebp+8]

mov ebp,esp 中，esp 寄存器中的值被直接存儲在了 ebp 中，也就是說，如果 esp 寄存器的值是100的話那麼 ebp 寄存器的值也是 100。

而在 mov eax,dword ptr [ebp+8] 這條指令中，ebp 寄存器的值 + 8 後會被解析稱為內存地址。如果 ebp

寄存器的值是100的話，那麼 eax 寄存器的值就是 100 + 8 的地址的值。dword ptr 也叫做 double word pointer 簡單解釋一下就是從指定的內存地址中讀出4字節的數據

對棧進行 push 和 pop

程序運行時，會在內存上申請分配一個稱為棧的數據空間。棧（stack）的特性是后入先出，數據在存儲時是從內存的下層（大的地址編號）逐漸往上層（小的地址編號）累積，讀出時則是按照從上往下進行讀取的。

棧是存儲臨時數據的區域，它的特點是通過 push 指令和 pop 指令進行數據的存儲和讀出。向棧中存儲數據稱為 入棧 ，從棧中讀出數據稱為 出棧，32位 x86 系列的 CPU 中，進行1次 push 或者 pop，即可處理 32 位（4字節）的數據。

函數的調用機制

下面我們一起來分析一下函數的調用機制，我們以上面的 C 語言編寫的代碼為例。首先，讓我們從MyFunc 函數調用AddNum 函數的彙編語言部分開始，來對函數的調用機制進行說明。棧在函數的調用中發揮了巨大的作用，下面是經過處理后的 MyFunc 函數的彙編處理內容

_MyFunc      proc    near
    push            ebp       ; 將 ebp 寄存器的值存入棧中              (1) 
    mov             ebp,esp ; 將 esp 寄存器的值存入 ebp 寄存器中        (2)
    push            456         ; 將 456 入棧                                                (3)
    push            123         ; 將 123 入棧                                                (4)
    call            _AddNum ; 調用 AddNum 函數                                       (5)
    add             esp,8       ; esp 寄存器的值 + 8                                     (6)
    pop             ebp         ; 讀出棧中的數值存入 esp 寄存器中                 (7)
    ret                             ; 結束 MyFunc 函數，返回到調用源                   (8)
_MyFunc         endp

代碼解釋中的(1)、(2)、(7)、(8)的處理適用於 C 語言中的所有函數，我們會在後面展示 AddNum 函數處理內容時進行說明。這裏希望大家先關注(3) – (6) 這一部分，這對了解函數調用機制至關重要。

(3) 和 (4) 表示的是將傳遞給 AddNum 函數的參數通過 push 入棧。在 C 語言源代碼中，雖然記述為函數 AddNum(123,456)，但入棧時則會先按照 456，123 這樣的順序。也就是位於後面的數值先入棧。這是 C 語言的規定。(5) 表示的 call 指令，會把程序流程跳轉到 AddNum 函數指令的地址處。在彙編語言中，函數名表示的就是函數所在的內存地址。AddNum 函數處理完畢后，程序流程必須要返回到編號(6) 這一行。call 指令運行后，call 指令的下一行(也就指的是 (6) 這一行)的內存地址(調用函數完畢后要返回的內存地址)會自動的 push 入棧。該值會在 AddNum 函數處理的最後通過 ret 指令 pop 出棧，然後程序會返回到 (6) 這一行。

(6) 部分會把棧中存儲的兩個參數 (456 和 123) 進行銷毀處理。雖然通過兩次的 pop 指令也可以實現，不過採用 esp 寄存器 + 8 的方式會更有效率(處理 1 次即可)。對棧進行數值的輸入和輸出時，數值的單位是4字節。因此，通過在負責棧地址管理的 esp 寄存器中加上4的2倍8，就可以達到和運行兩次 pop 命令同樣的效果。雖然內存中的數據實際上還殘留着，但只要把 esp 寄存器的值更新為數據存儲地址前面的數據位置，該數據也就相當於銷毀了。

我在編譯 Sample4.c 文件時，出現了下圖的這條消息

圖中的意思是指 c 的值在 MyFunc 定義了但是一直未被使用，這其實是一項編譯器優化的功能，由於存儲着 AddNum 函數返回值的變量 c 在後面沒有被用到，因此編譯器就認為 該變量沒有意義，進而也就沒有生成與之對應的彙編語言代碼。

下圖是調用 AddNum 這一函數前後棧內存的變化

函數的內部處理

上面我們用彙編代碼分析了一下 Sample4.c 整個過程的代碼，現在我們着重分析一下 AddNum 函數的源代碼部分，分析一下參數的接收、返回值和返回等機制

_AddNum         proc        near
    push            ebp                        -----------(1)
    mov             ebp,esp                -----------(2)
    mov             eax,dword ptr[ebp+8]   -----------(3)
    add             eax,dword ptr[ebp+12]  -----------(4)
    pop             ebp                                      -----------(5)
    ret             ----------------------------------(6)
_AddNum         endp

ebp 寄存器的值在(1)中入棧，在(5)中出棧，這主要是為了把函數中用到的 ebp 寄存器的內容，恢復到函數調用前的狀態。

(2) 中把負責管理棧地址的 esp 寄存器的值賦值到了 ebp 寄存器中。這是因為，在 mov 指令中方括號內的參數，是不允許指定 esp 寄存器的。因此，這裏就採用了不直接通過 esp，而是用 ebp 寄存器來讀寫棧內容的方法。

(3) 使用[ebp + 8] 指定棧中存儲的第1個參數123，並將其讀出到 eax 寄存器中。像這樣，不使用 pop 指令，也可以參照棧的內容。而之所以從多個寄存器中選擇了 eax 寄存器，是因為 eax 是負責運算的累加寄存器。

通過(4) 的 add 指令，把當前 eax 寄存器的值同第2個參數相加后的結果存儲在 eax 寄存器中。[ebp + 12] 是用來指定第2個參數456的。在 C 語言中，函數的返回值必須通過 eax 寄存器返回，這也是規定。也就是 函數的參數是通過棧來傳遞，返回值是通過寄存器返回的。

(6) 中 ret 指令運行后，函數返回目的地內存地址會自動出棧，據此，程序流程就會跳轉返回到(6) (Call _AddNum) 的下一行。這時，AddNum 函數入口和出口處棧的狀態變化，就如下圖所示

這是程序員需要了解的硬核知識之彙編語言(一) 第一篇文章，下一篇文章我們會着重討論局部變量和全局變量以及循環控制語句的彙編語言，防止斷更，請關注我

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

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

※高價3c回收,收購空拍機,收購鏡頭,收購 MACBOOK-更多收購平台討論專區

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

※收購3c瘋!各款手機、筆電、相機、平板,歡迎來詢價!

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

今天，我們將通過Apache Kafka主題構建一些彼此異步通信的微服務。我們使用Micronaut框架，它為與Kafka集成提供專門的庫。讓我們簡要介紹一下示例係統的體繫結構。我們有四個微型服務：訂單服務，行程服務，司機服務和乘客服務。這些應用程序的實現非常簡單。它們都有內存存儲，並連接到同一個Kafka實例。

我們系統的主要目標是為客戶安排行程。訂單服務應用程序還充當網關。它接收來自客戶的請求，保存歷史記錄並將事件發送到orders主題。所有其他微服務都在監聽orders這個主題，並處理order-service發送的訂單。每個微服務都有自己的專用主題，其中發送包含更改信息的事件。此類事件由其他一些微服務接收。架構如下圖所示。

在閱讀本文之前，有必要熟悉一下Micronaut框架。您可以閱讀之前的一篇文章，該文章描述了通過REST API構建微服務通信的過程:。

1 運行Kafka

要在本地機器上運行Apache Kafka，我們可以使用它的Docker映像。最新的鏡像是由共享的。在啟動Kafka容器之前，我們必須啟動kafka所用使用的ZooKeeper服務器。如果在Windows上運行Docker，其虛擬機的默認地址是192.168.99.100。它還必須設置為Kafka容器的環境。

Zookeeper和Kafka容器都將在同一個網絡中啟動。在docker中運行Zookeeper以zookeeper的名稱提供服務，並在暴露2181端口。Kafka容器需要在環境變量使用KAFKA_ZOOKEEPER_CONNECT的地址。

$ docker network create kafka
$ docker run -d --name zookeeper --network kafka -p 2181:2181 wurstmeister/zookeeper
$ docker run -d --name kafka -p 9092:9092 --network kafka --env KAFKA_ADVERTISED_HOST_NAME=192.168.99.100 --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 wurstmeister/kafka

2 添加micronaut-kafka依賴

使用Kafka構建的microaut應用程序可以在HTTP服務器存在的情況下啟動，也可以在不存在HTTP服務器的情況下啟動。要啟用Micronaut Kafka，需要添加micronaut-kafka庫到依賴項。如果您想暴露HTTP API，您還應該添加micronaut-http-server-netty:

<dependency>
    <groupId>io.micronaut.configuration</groupId>
    <artifactId>micronaut-kafka</artifactId>
</dependency>
<dependency>
    <groupId>io.micronaut</groupId>
    <artifactId>micronaut-http-server-netty</artifactId>
</dependency>

3 構建訂單微服務

訂單微服務是唯一一個啟動嵌入式HTTP服務器並暴露REST API的應用程序。這就是為什麼我們可以為Kafka提供內置Micronaut健康檢查。要做到這一點，我們首先應該添加micronaut-management依賴:

<dependency>
    <groupId>io.micronaut</groupId>
    <artifactId>micronaut-management</artifactId>
</dependency>

為了方便起見，我們將通過在application.yml中定義以下配置來啟用所有管理端點並禁用它們的HTTP身份驗證。

endpoints:
  all:
    enabled: true
    sensitive: false

現在，可以在地址欄下使用health check。我們的示例應用程序還將暴露添加新訂單和列出所有以前創建的訂單的簡單REST API。下面是暴露這些端點的Micronaut控制器實現:

@Controller("orders")
public class OrderController {

    @Inject
    OrderInMemoryRepository repository;
    @Inject
    OrderClient client;

    @Post
    public Order add(@Body Order order) {
        order = repository.add(order);
        client.send(order);
        return order;
    }

    @Get
    public Set<Order> findAll() {
        return repository.findAll();
    }

}

每個微服務都使用內存存儲庫實現。以下是訂單微服務（Order-Service）中的存儲庫實現：

@Singleton
public class OrderInMemoryRepository {

    private Set<Order> orders = new HashSet<>();

    public Order add(Order order) {
        order.setId((long) (orders.size() + 1));
        orders.add(order);
        return order;
    }

    public void update(Order order) {
        orders.remove(order);
        orders.add(order);
    }

    public Optional<Order> findByTripIdAndType(Long tripId, OrderType type) {
        return orders.stream().filter(order -> order.getTripId().equals(tripId) && order.getType() == type).findAny();
    }

    public Optional<Order> findNewestByUserIdAndType(Long userId, OrderType type) {
        return orders.stream().filter(order -> order.getUserId().equals(userId) && order.getType() == type)
                .max(Comparator.comparing(Order::getId));
    }

    public Set<Order> findAll() {
        return orders;
    }

}

內存存儲庫存儲Order對象實例。Order對象還被發送到名為orders的Kafka主題。下面是Order類的實現:

public class Order {

    private Long id;
    private LocalDateTime createdAt;
    private OrderType type;
    private Long userId;
    private Long tripId;
    private float currentLocationX;
    private float currentLocationY;
    private OrderStatus status;
    
    // ... GETTERS AND SETTERS
}

4 使用Kafka異步通信

現在，讓我們想一個可以通過示例係統實現的用例—— 添加新的行程。

我們創建了OrderType.NEW_TRIP類型的新訂單。在此之後，(1) 訂單服務創建一個訂單並將其發送到orders主題。訂單由三個微服務接收: 司機服務、乘客服務和行程服務。
(2)所有這些應用程序都處理這個新訂單。乘客服務應用程序檢查乘客帳戶上是否有足夠的資金。如果沒有，它就取消了行程，否則它什麼也做不了。司機服務正在尋找最近可用的司機，(3) 行程服務創建和存儲新的行程。司機服務和行程服務都將事件發送到它們的主題( drivers, trips),其中包含相關更改的信息。

每一個事件可以被其他microservices訪問,例如，(4) 行程服務偵聽來自司機服務的事件，以便為行程分配一個新的司機

下圖說明了在添加新的行程時，我們的微服務之間的通信過程。

現在，讓我們繼續討論實現細節。

4.1 發送訂單

首先，我們需要創建Kafka客戶端，負責向主題發送消息。我們創建的一個接口，命名為OrderClient，為它添加@KafkaClient並聲明用於發送消息的一個或多個方法。每個方法都應該通過@Topic註解設置目標主題名稱。對於方法參數，我們可以使用三個註解@KafkaKey、@Body或@Header。@KafkaKey用於分區，這是我們的示例應用程序所需要的。在下面可用的客戶端實現中，我們只使用@Body註解。

@KafkaClient
public interface OrderClient {

    @Topic("orders")
    void send(@Body Order order);

}

4.2 接收訂單

一旦客戶端發送了一個訂單，它就會被監聽orders主題的所有其他微服務接收。下面是司機服務中的監聽器實現。監聽器類OrderListener應該添加@KafkaListener註解。我們可以聲明groupId作為一個註解參數，以防止單個應用程序的多個實例接收相同的消息。然後，我們聲明用於處理傳入消息的方法。與客戶端方法相同，應該通過@Topic註解設置目標主題名稱，因為我們正在監聽Order對象，所以應該使用@Body註解——與對應的客戶端方法相同。

@KafkaListener(groupId = "driver")
public class OrderListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(OrderListener.class);

    private DriverService service;

    public OrderListener(DriverService service) {
        this.service = service;
    }

    @Topic("orders")
    public void receive(@Body Order order) {
        LOGGER.info("Received: {}", order);
        switch (order.getType()) {
            case NEW_TRIP -> service.processNewTripOrder(order);
        }
    }

}

4.3 發送到其他主題

現在，讓我們看一下司機服務中的processNewTripOrder方法。DriverService注入兩個不同的Kafka Client
bean: OrderClient和DriverClient。當處理新訂單時，它將試圖尋找與發送訂單的乘客最近的司機。找到他之後，將該司機的狀態更改為UNAVAILABLE，並將帶有Driver對象的事件發送到drivers主題。

@Singleton
public class DriverService {

    private static final Logger LOGGER = LoggerFactory.getLogger(DriverService.class);

    private DriverClient client;
    private OrderClient orderClient;
    private DriverInMemoryRepository repository;

    public DriverService(DriverClient client, OrderClient orderClient, DriverInMemoryRepository repository) {
        this.client = client;
        this.orderClient = orderClient;
        this.repository = repository;
    }

    public void processNewTripOrder(Order order) {
        LOGGER.info("Processing: {}", order);
        Optional<Driver> driver = repository.findNearestDriver(order.getCurrentLocationX(), order.getCurrentLocationY());
        driver.ifPresent(driverLocal -> {
            driverLocal.setStatus(DriverStatus.UNAVAILABLE);
            repository.updateDriver(driverLocal);
            client.send(driverLocal, String.valueOf(order.getId()));
            LOGGER.info("Message sent: {}", driverLocal);
        });
    }
    
    // ...
}

這是Kafka Client在司機服務中的實現，用於向driver主題發送消息。因為我們需要將Driver與Order關聯起來，所以我們使用@Header註解的orderId參數。沒有必要把它包括到Driver類中，將其分配給監聽器端的正確行程。

@KafkaClient
public interface DriverClient {

    @Topic("drivers")
    void send(@Body Driver driver, @Header("Order-Id") String orderId);

}

4.4 服務間通信

由DriverListener收到@KafkaListener在行程服務中聲明。它監聽傳入到trip主題。接收方法的參數和客戶端發送方法的類似，如下所示：

@KafkaListener(groupId = "trip")
public class DriverListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(OrderListener.class);

    private TripService service;

    public DriverListener(TripService service) {
        this.service = service;
    }

    @Topic("drivers")
    public void receive(@Body Driver driver, @Header("Order-Id") String orderId) {
        LOGGER.info("Received: driver->{}, header->{}", driver, orderId);
        service.processNewDriver(driver);
    }

}

最後一步，將orderId查詢到的行程Trip與driverId關聯，這樣整個流程就結束。

@Singleton
public class TripService {

    private static final Logger LOGGER = LoggerFactory.getLogger(TripService.class);

    private TripInMemoryRepository repository;
    private TripClient client;

    public TripService(TripInMemoryRepository repository, TripClient client) {
        this.repository = repository;
        this.client = client;
    }


    public void processNewDriver(Driver driver, String orderId) {
        LOGGER.info("Processing: {}", driver);
        Optional<Trip> trip = repository.findByOrderId(Long.valueOf(orderId));
        trip.ifPresent(tripLocal -> {
            tripLocal.setDriverId(driver.getId());
            repository.update(tripLocal);
        });
    }
    
    // ... OTHER METHODS

}

5 跟蹤

我們可以使用Micronaut Kafka輕鬆地啟用分佈式跟蹤。首先，我們需要啟用和配置Micronaut跟蹤。要做到這一點，首先應該添加一些依賴項：

<dependency>
    <groupId>io.micronaut</groupId>
    <artifactId>micronaut-tracing</artifactId>
</dependency>
<dependency>
    <groupId>io.zipkin.brave</groupId>
    <artifactId>brave-instrumentation-http</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.zipkin.reporter2</groupId>
    <artifactId>zipkin-reporter</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.opentracing.brave</groupId>
    <artifactId>brave-opentracing</artifactId>
</dependency>
<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-kafka-client</artifactId>
    <version>0.0.16</version>
    <scope>runtime</scope>
</dependency>

我們還需要在application.yml配置文件中，配置Zipkin的追蹤的地址等

tracing:
  zipkin:
    enabled: true
    http:
      url: http://192.168.99.100:9411
    sampler:
      probability: 1

在啟動應用程序之前，我們必須運行Zipkin容器：

$ docker run -d --name zipkin -p 9411:9411 openzipkin/zipkin

6 總結

在本文中，您將了解通過Apache Kafka使用異步通信構建微服務架構的過程。我已經向大家展示了Microaut Kafka庫最重要的特性，它允許您輕鬆地聲明Kafka主題的生產者和消費者，為您的微服務啟用健康檢查和分佈式跟蹤。我已經為我們的系統描述了一個簡單的場景的實現，包括根據客戶的請求添加一個新的行程。本示例係統的整體實現，請查看GitHub上的

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

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

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

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

※帶您來看台北網站建置，台北網頁設計，各種案例分享

cnn卷積神經網絡在前面已經有所了解了，目前博主也使用它進行了一個圖像分類問題，基於kaggle裏面的food-101進行的圖像識別，識別率有點感人，基於數據集的關係，大致來說還可行。
下面我就繼續學習rnn神經網絡。

rnn神經網絡（遞歸/循環神經網絡）模式如下：

我們在處理文字等問題的時候，我們的輸入會把上一個時間輸出的數據作為下一個時間的輸入數據進行處理。
例如：我們有一段話，我們將其分詞，得到t個數據，我們分別將每一個詞傳入到x0，x1….xt裏面，當x0傳入后，會得到一個結果h0，同時我們會將處理后的數據傳入到下個時間，到下個時間的時候，我們會再傳入一個數據x1，同時還有上一個時間處理后的數據，將這兩個數據進行整合計算，然後再向下傳輸，一直到結束。
rnn本質來說還是一個bp迴路，不過他只是比bp網絡多一個環節，即它可以反饋上一時間點處理后的數據。

上圖細化如下：

rnn實際上還是存在梯度消失的問題，因此如上圖所示，當我們在第一個時間輸入的數據，可能在很久之後他就已經梯度消失了（影響很小），因此我們使用lstm(long short trem memory)

上圖有三個門：輸入門    忘記門   輸出門
1.輸入門：通過input * g 來判斷是否輸入，如果不輸入就為0，輸入就是0，以此判斷信號是否輸入
2.忘記門：這個信號是否需要衰減多少，可能為50%，衰減是根據信號來判斷。
3.輸入門：通過判斷是否輸出，或者輸出多少，例如輸出50%。
因此上述圖可化為：

可以看出，這三個門，所有得影響都是關於輸入和上一個數據得輸出來進行計算的。

可以看下圖：

我們使用lstm得話，通過三個門決定信號是否向下傳輸，傳輸多少都可以控制，是否傳入信號，輸出信息都進行控制。

下面我們還是用tensorflow實現，數據集還是手寫数字，雖然rnn主要是用在文字和語言上，但是它依舊可以用在圖片上。
下面給出代碼：

```python
import tensorflow as tf
from tensorflow.contrib import rnn
from tensorflow.examples.tutorials.mnist import  input_data
mnist=input_data.read_data_sets("MNNIST_data",one_hot=True)

#輸入圖片為 28*28
n_inputs=28#輸入一行，一行有28個像素
max_time=28#一共28行，所以為28*28
lstm_size=100#100個隱藏單元
batch_size=50
n_classes=10
n_batch=mnist.train.num_examples//batch_size#計算一共多少批次

#這裏none表示第一個維度可以是任意長度
x=tf.placeholder(tf.float32,[None,784])

y=tf.placeholder(tf.float32,[None,10])

#初始化權值
weights=tf.Variable(tf.truncated_normal([lstm_size,n_classes],stddev=0.1))
#初始化偏置值
biases=tf.Variable(tf.constant(0.1,shape=[n_classes]))

##定義Rnn 網絡
def RNN(X,weights,biases):
    inputs=tf.reshape(X,[-1,max_time,n_inputs])
    #定義lstm基本cell
    lstm_cell = rnn.BasicLSTMCell(lstm_size)
    #lstm_cell=tf.contrib.rnn.core_rnn_cell.BasicLSTMCell(lstm_size)
    outputs,final_state=tf.nn.dynamic_rnn(lstm_cell,inputs,dtype=tf.float32)
    results=tf.nn.softmax(tf.matmul(final_state[1],weights)+biases)
    return results
prediction=RNN(x,weights,biases)
#損失函數
cross_entropy=tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=prediction,labels=y))
#優化器
train_step=tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)
#保存結果
correct_prediction=tf.equal(tf.argmax(y,1),tf.argmax(prediction,1))

accuracy=tf.reduce_mean(tf.cast(correct_prediction,tf.float32))

init=tf.global_variables_initializer()

with tf.Session() as sess:
    sess.run(init)
    for epoch in range(6):
        for batch in range(n_batch):
            batch_xs,batch_ys=mnist.train.next_batch(batch_size)
            sess.run(train_step,feed_dict={x:batch_xs,y:batch_ys})

        acc=sess.run(accuracy,feed_dict={x:mnist.test.images,y:mnist.test.labels})
        print("iter:"+str(epoch)+"testing accuracy"+str(acc))

“`
運行結果如下：

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

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

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

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

※帶您來看台北網站建置，台北網頁設計，各種案例分享

修改屏幕DPI，會觸發控件的Unloaded/Loaded

現象/重現案例

這裏簡單介紹下，修改屏幕DPI，觸發Unloaded/Loaded的神奇案例

1. 我們新建一個窗口，添加一個UserControl1，然後在UserControl1中添加UserControl2

 1 <Window x:Class="WPFUnloadedTriggerTest.MainWindow"
 2         xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
 3         xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
 4         xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
 5         xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
 6         xmlns:local="clr-namespace:WPFUnloadedTriggerTest"
 7         mc:Ignorable="d"
 8         Title="MainWindow" Height="450" Width="800">
 9     <local:UserControl1></local:UserControl1>
10 </Window>
11 ------------------------------我是分隔線-----------------------------------
12 <UserControl x:Class="WPFUnloadedTriggerTest.UserControl1"
13              xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
14              xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
15              xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006" 
16              xmlns:d="http://schemas.microsoft.com/expression/blend/2008" 
17              xmlns:local="clr-namespace:WPFUnloadedTriggerTest"
18              mc:Ignorable="d" 
19              d:DesignHeight="450" d:DesignWidth="800">
20     <local:UserControl2></local:UserControl2>
21 </UserControl>

View Code

2. 显示窗口后，修改DPI比例

3. 設置完后，會觸發Unloaded/Loaded重新加載

Unloaded的觸發順序是UserControl1–>UserControl2，Window並不會觸發Unloaded事件！

是不是詭異？我們繼續。。。

 4. Window我們添加一個ControlTemplate模塊

1     <Window.Template>
2         <ControlTemplate TargetType="Window">
3             <Border>
4                 <AdornerDecorator>
5                     <ContentPresenter />
6                 </AdornerDecorator>
7             </Border>
8         </ControlTemplate>
9     </Window.Template>

 再重複2、3步驟，Unloaded的觸發順序變了：

觸發UserControl2的Unloaded，Window、UserControl1並不會觸發Unloaded事件！

問題分析

第2步驟中修改DPI后，Unloaded事件不一定觸發。如何必現呢？

將窗口靠近到任務欄上方，再修改文本比例。

 我們查看調用堆棧，貌似是系統給窗口發送消息然後調用BroadcastUnloadedEvent事件，觸發Unload

 所以應該是修改DPI，窗口寬高超出了當前屏幕尺寸範圍，系統對UserControl的視覺樹進行重新加載布局。

至於窗口沒有觸發Unloaded、以及在窗口添加以上模塊後下一級子控件也沒有觸發Unloaded事件的原因，暫不了解

而對WPF-Unloaded/Loaded的已知情況如下：

• FrameworkElement， 第一次加載显示時，會觸發Loaded。元素被釋放時，會觸發Unloaded。窗口Show/Close時，視覺樹變化都會觸發加載事件
• MenuItem， 在FrameworkElement基礎上，每次和隱藏MenuItem時，會額外觸發Load/Unloaded
• TabControl，當你選中一個tabItem時會觸發Loaded，當你取消選中一個tabItem時會觸發Unloaded，所以切換Tab時必定有一個Loaded一個Unloaded。
• Expander，每次被Expanded擴展時會引發Loaded，但當隱藏時不會引發Unloaded。

 以上問題的解決方案？暫時沒有解決方案，只有規避措施，不要過於依賴於Unload/Loaded，而且使用了Unload/Loaded時也要添加註銷機制，防止重入

我在github提了個issue：

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

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

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

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

※帶您來看台北網站建置，台北網頁設計，各種案例分享

有點笨，參考了好幾篇大佬們寫的文章才整理出來的筆記….

字面意思上解釋，線程池就是裝有線程的池，我們可以把要執行的多線程交給線程池來處理，和連接池的概念一樣，通過維護一定數量的線程池來達到多個線程的復用。

好處

多線程產生的問題

一般我們使用到多線程的編程的時候，需要通過new Thread(xxRunnable).start()創建並開啟線程，我們可以使用多線程來達到最優效率（如多線程下載）。

但是，線程不是越多就越好，線程過多，創建和銷毀就會消耗系統的資源，也不方便管理。

除此之外，多線程還會造成併發問題，線程併發數量過多，搶佔系統資源從而導致阻塞。

線程池優點

我們將線程放入線程池，由線程池對線程進行管理，可以對線程池中緩衝的線程進行復用，這樣，就不會經常去創建和銷毀線程了，從而省下了系統的資源。

線程池能夠有效的控制線程併發的數量，能夠解決多線程造成的併發問題。

除此之外，線程池還能夠對線程進行一定的管理，如延時執行、定時循環執行的策略等

線程池實現

線程池的實現，主要是通過這個類ThreadPoolExecutor，其的構造參數非常長，我們先大概了解，之後再進行詳細的介紹。

public ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,long keepAliveTime,
    TimeUnit unit,BlockingQueue workQueue,
    RejectedExecutionHandler handler)

• corePoolSize：線程池核心線程數量
• maximumPoolSize：線程池最大線程數量
• keepAliverTime：當活躍線程數大於核心線程數時，空閑的多餘線程最大存活時間
• unit：存活時間的單位
• workQueue：存放線程的工作隊列
• handler：超出線程範圍和隊列容量的任務的處理程序（拒絕策略）

這裏大概簡單說明一下線程池的運行流程：

當線程被添加到線程池中，如果線程池中的當前的線程數量等於線程池定義的最大核心線程數量（corePoolSize)了，此線程就會別放入線程的工作隊列(workQueue)中，等待線程池的調用。

Java提供了一個工具類Excutors，方便我們快速創建線程池，其底層也是調用了ThreadPoolExecutor

不過阿里巴巴Java規範中強制要求我們應該通過ThreadPoolExecutor來創建自己的線程池，使用Excutors容易造成OOM問題。

所以，我們先從Excutors開始學習，之後在對ThreadPoolExecutor進行詳細的講解

Excutors

由於Excutors是工具類，所以下面的介紹的都是其的靜態方法，如果是比較線程數目比較少的小項目，可以使用此工具類來創建線程池

PS：把線程提交給線程池中，有兩種方法，一種是submit，另外一種則是execute

兩者的區別：

1. execute沒有返回值，如果不需要知道線程的結果就使用execute方法，性能會好很多。
2. submit返回一個Future對象，如果想知道線程結果就使用submit提交，而且它能在主線程中通過Future的get方法捕獲線程中的異常

線程池可以接收兩種的參數，一個為Runnable對象，另外則是Callable對象

Callable是JDK1.5時加入的接口，作為Runnable的一種補充，允許有返回值，允許拋出異常。

主要的幾個靜態方法：

1.newSingleThreadExecutor

創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行，我們可以使用它來達到控制線程順序執行。

控制進程順序執行：

Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("這是線程1");
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
Thread thread2 = new Thread(new Runnable() {
    @Override
    public void run() {
        try {
            System.out.println("這是線程2");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});
Thread thread3 = new Thread(new Runnable() {
    @Override
    public void run() {
        try {
            System.out.println("這是線程3");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});
//創建線程池對象
ExecutorService executorService = Executors.newSingleThreadExecutor();
//把線程添加到線程池中
executorService.submit(thread1);
executorService.submit(thread2);
executorService.submit(thread3);

之後出現的結果就是按照順序輸出

2.newFixedThreadPool

創建一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待。定長線程池的大小最好根據系統資源進行設置。如Runtime.getRuntime().availableProcessors()

3.newCachedThreadPool

創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程，線程池為無限大，當執行第二個任務時第一個任務已經完成，會復用執行第一個任務的線程，而不用每次新建線程。

代碼：

//創建了一個自定義的線程
public class MyThread extends Thread {
    private int index;

    public MyThread(int index) {
        this.index = index;
    }

    @Override
    public void run() {
        System.out.println(index+" 當前線程"+Thread.currentThread().getName());
    }
}

//創建緩存線程池
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
    executorService.execute(new MyThread(i));
    try {
        //這裏模擬等待時間，等待線程池復用回收線程
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

可以看到結果都是使用的同一個線程

4.newScheduledThreadPool

創建一個定長線程池，支持定時、周期性或延時任務執行

延遲1s后啟動線程：

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
scheduledExecutorService.schedule(new MyThread(1),1, TimeUnit.SECONDS);

ThreadPoolExecutor

構造方法

上面提到的那個構造方法其實只是ThreadPoolExecutor類中的一個，ThreadPoolExecutor類中存在有四種不同的構造方法，主要區別就是參數不同。

//五個參數的構造函數
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue)

//六個參數的構造函數-1
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory)

//六個參數的構造函數-2
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler)

//七個參數的構造函數
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

首先，有個概念需要明白，線程池的最大線程數（線程總數,maximumPoolSize）= 核心線程數(corePoolSize)+非核心線程數

• corePoolSize：線程池核心線程數量
• maximumPoolSize：線程池最大線程數量
• keepAliverTime：當活躍線程數大於核心線程數時，空閑的多餘線程最大存活時間
• unit：存活時間的單位
• workQueue：存放線程的工作隊列
• handler：超出線程範圍和隊列容量的任務的處理程序（拒絕策略）

核心線程和非核心線程有什麼區別呢？

核心線程是永遠不會被線程池丟棄回收（即使核心線程沒有工作），非核心線程則是超過一定時間（keepAliverTime)則就會被丟棄

workQueue

當所有的核心線程都在工作時，新添加的任務會被添加到這個隊列中等待處理，如果隊列滿了，則新建非核心線程執行任務

1.SynchronousQueue：這個隊列接收到任務的時候，會直接提交給線程處理，而不保留它，如果所有線程都在工作怎麼辦？那就新建一個線程來處理這個任務！所以為了保證不出現線程數達到了maximumPoolSize而不能新建線程的錯誤，使用這個類型隊列的時候，maximumPoolSize一般指定成Integer.MAX_VALUE，即無限大

2.LinkedBlockingQueue：這個隊列接收到任務的時候，如果當前線程數小於核心線程數，則新建線程(核心線程)處理任務；如果當前線程數等於核心線程數，則進入隊列等待。由於這個隊列沒有最大值限制，即所有超過核心線程數的任務都將被添加到隊列中，這也就導致了maximumPoolSize的設定失效，因為總線程數永遠不會超過corePoolSize

3.ArrayBlockingQueue：可以限定隊列的長度，接收到任務的時候，如果沒有達到corePoolSize的值，則新建線程(核心線程)執行任務，如果達到了，則入隊等候，如果隊列已滿，則新建線程(非核心線程)執行任務，又如果總線程數到了maximumPoolSize，並且隊列也滿了，則發生錯誤

4.DelayQueue：隊列內元素必須實現Delayed接口，這就意味着你傳進去的任務必須先實現Delayed接口。這個隊列接收到任務時，首先先入隊，只有達到了指定的延時時間，才會執行任務

拒絕策略：

兩種方法設置拒絕策略：

//ThreadPoolExecutor對象的setRejectedExecutionHandler方法設置
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue);
threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
//構造方法進行設置，省略

線程池默認的拒絕行為是AbortPolicy，也就是拋出RejectedExecutionHandler異常，該異常是非受檢異常，很容易忘記捕獲。

如果不關心任務被拒絕的事件，可以將拒絕策略設置成DiscardPolicy，這樣多餘的任務會悄悄的被忽略。

ThreadFactory

一個接口類，用來對線程進行設置，需要實現newThread(Runnable r)方法

官方的文檔說明:

newThread此方法一般來初始化線程的優先級(priority)，名字(name)，守護進程(daemon)或線程組(ThreadGroup)

簡單的例子（讓某個類實現ThreadFactory接口）：

@Override
public Thread newThread(Runnable r) {
    Thread thread = new Thread(r);
    thread.setDaemon(true);
    return thread;
}

線程池獲取執行結果

PS：把線程提交給線程池中，有兩種方法，一種是submit，另外一種則是execute

兩者的區別：

1. execute沒有返回值，如果不需要知道線程的結果就使用execute方法，性能會好很多。
2. submit返回一個Future對象，如果想知道線程結果就使用submit提交，而且它能在主線程中通過Future的get方法捕獲線程中的異常

線程池可以接收兩種的參數，一個為Runnable對象，另外則是Callable對象

Callable是JDK1.5時加入的接口，作為Runnable的一種補充，允許有返回值，允許拋出異常。

線程池的處理結果、以及處理過程中的異常都被包裝到Future中，並在調用Future.get()方法時獲取，執行過程中的異常會被包裝成ExecutionException，submit()方法本身不會傳遞結果和任務執行過程中的異常。

獲取執行結果的代碼可以這樣寫：

ExecutorService executorService = Executors.newFixedThreadPool(4);
Future<Object> future = executorService.submit(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            //該異常會在調用Future.get()時傳遞給調用者
            throw new RuntimeException("exception in call~");
        }
    });
    
try {
    //獲得返回結果
    Object result = future.get();
    
    
} catch (InterruptedException e) {
  // interrupt
} catch (ExecutionException e) {
  // exception in Callable.call()
  e.printStackTrace();
}

線程池運行流程

一個形象的比喻說明線程池的流程：

規定：

1. 線程池比作成一家公司
2. 公司的最大員工數為maximumPoolSize
3. 最大正式員工數為coolPoolSize（核心線程的總數）
4. 最大員工數（maximumPoolSize） = 最大正式員工（coolPoolSize）和臨時工（非核心線程）
5. 單子（任務）可看做為線程
6. 隊列使用的是ArrayBlockingQueue
7. 一個員工只能幹一個任務

最開始的時候，公司是沒有一名員工。之後，公司接到了單子（任務），這個時候，公司才去找員工（創建核心線程並讓線程開始執行），這個時候找到的員工就是正式員工了。

公司的聲譽越來越好，於是來了更多的單子，公司繼續招人，直到正式員工數量達到最大的正式員工的數量（核心線程數量已達到最大）

於是，多出來的單子就暫時地存放在了隊列中，都在排隊，等待正式員工們把手頭的工作做完之後，就從隊列中依次取出單子繼續工作。

某天，來了一個新單子，但是這個時候隊列已經滿了，公司為了自己的信譽和聲譽着想，不得已只能去找臨時工（創建非核心線程）來幫忙開始進行工作（負責新單子）

在此之後，又來了新單子，公司繼續去招臨時工為新來的單子工作，直到正式工和臨時工的數量已經達到了公司最大員工數。

這個時候，公司沒有辦法了，只能拒絕新來的單子了（拒絕策略）

此時，正式工和臨時工都是在加班加點去從隊列中取出任務來工作，終於某一天，隊列的已經沒有單子了，市場發展不好，單子越來越少，臨時工很久都不工作了（非核心線程超過了最大存活時間keepAliveTime)，公司就把這些臨時工解僱了，直到剩下只有正式員工。

PS:如果也想要解僱正式員工(銷毀核心線程)，可以設置ThreadPoolExecutor對象的的allowCoreThreadTimeOut這個屬性為true

個人理解，可能不是很正確，僅供參考！

線程池關閉

總結

如果是小的Java程序，可以使用Excutors，如果是服務器程序，則使用ThreadPoolExecutor進行自定義線程池的創建

參考鏈接：

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

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

※高價3c回收,收購空拍機,收購鏡頭,收購 MACBOOK-更多收購平台討論專區

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

※收購3c瘋!各款手機、筆電、相機、平板,歡迎來詢價!

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