作者: admin

一 Pod定義詳解

1.1 完整Pod定義文件

  1 apiVersion: v1			#必選，版本號，例如v1,版本號必須可以用 kubectl api-versions 查詢到
  2 kind: Pod				#必選，Pod
  3 metadata:				#必選，元數據
  4   name: string			#必選，Pod名稱，需符合RFC 1035規範
  5   namespace: string			#必選，Pod所屬的命名空間，默認為"default"
  6   labels:				#自定義標籤
  7     - name: string			#自定義標籤名字
  8   annotations:			#自定義註釋列表
  9     - name: string
 10 spec:				#必選，Pod中容器的詳細定義
 11   containers:			#必選，Pod中容器列表
 12   - name: string			#必選，容器名稱，需符合RFC 1035規範
 13     image: string			#必選，容器的鏡像名稱
 14     imagePullPolicy: [ Always|Never|IfNotPresent ]	#獲取鏡像的策略，Alawys表示每次都嘗試下載鏡像，IfnotPresent表示優先使用本地鏡像,否則下載鏡像，Nerver表示僅使用本地鏡像
 15     command: [string]		#容器的啟動命令列表，如不指定，使用打包時使用的啟動命令
 16     args: [string]			#容器的啟動命令參數列表
 17     workingDir: string		#容器的工作目錄
 18     volumeMounts:			#掛載到容器內部的存儲卷配置
 19     - name: string			#引用pod定義的共享存儲卷的名稱，需用volumes[]部分定義的的卷名
 20       mountPath: string		#存儲卷在容器內mount的絕對路徑，應少於512字符
 21       readOnly: boolean		#是否為只讀模式，默認為讀寫模式
 22     ports:				#需要暴露的端口庫號列表
 23     - name: string			#端口的名稱
 24       containerPort: int		#容器需要監聽的端口號
 25       hostPort: int		        #容器所在主機需要監聽的端口號，默認與Container相同
 26       protocol: string		#端口協議，支持TCP和UDP，默認TCP
 27     env:				#容器運行前需設置的環境變量列表
 28     - name: string			#環境變量名稱
 29       value: string		        #環境變量的值
 30     resources:			#資源限制和請求的設置
 31       limits:			#資源限制的設置
 32         cpu: string		        #CPU的限制，單位為core數，將用於docker run --cpu-shares參數
 33         memory: string		#內存限制，單位可以為Mib/Gib，將用於docker run --memory參數
 34       requests:			#資源請求的設置
 35         cpu: string		        #CPU請求，容器啟動的初始可用數量
 36         memory: string		#內存請求,容器啟動的初始可用數量
 37     livenessProbe:			#對Pod內各容器健康檢查的設置，當探測無響應幾次后將自動重啟該容器，檢查方法有exec、httpGet和tcpSocket，對一個容器只需設置其中一種方法即可
 38       exec:			        #對Pod容器內檢查方式設置為exec方式
 39         command: [string]		#exec方式需要制定的命令或腳本
 40       httpGet:			#對Pod內個容器健康檢查方法設置為HttpGet，需要制定Path、port
 41         path: string
 42         port: number
 43         host: string
 44         scheme: string
 45         HttpHeaders:
 46         - name: string
 47           value: string
 48       tcpSocket:			#對Pod內個容器健康檢查方式設置為tcpSocket方式
 49          port: number
 50        initialDelaySeconds: 0	#容器啟動完成后首次探測的時間，單位為秒
 51        timeoutSeconds: 0		#對容器健康檢查探測等待響應的超時時間，單位秒，默認1秒
 52        periodSeconds: 0		#對容器監控檢查的定期探測時間設置，單位秒，默認10秒一次
 53        successThreshold: 0
 54        failureThreshold: 0
 55        securityContext:
 56          privileged: false
 57     restartPolicy: [Always | Never | OnFailure]	#Pod的重啟策略，Always表示一旦不管以何種方式終止運行，kubelet都將重啟，OnFailure表示只有Pod以非0退出碼退出才重啟，Nerver表示不再重啟該Pod
 58     nodeSelector: obeject		#設置NodeSelector表示將該Pod調度到包含這個label的node上，以key：value的格式指定
 59     imagePullSecrets:		#Pull鏡像時使用的secret名稱，以key：secretkey格式指定
 60     - name: string
 61     hostNetwork: false		#是否使用主機網絡模式，默認為false，如果設置為true，表示使用宿主機網絡
 62     volumes:			#在該pod上定義共享存儲卷列表
 63     - name: string			#共享存儲卷名稱 （volumes類型有很多種）
 64       emptyDir: {}			#類型為emtyDir的存儲卷，與Pod同生命周期的一個臨時目錄。為空值
 65       hostPath: string		#類型為hostPath的存儲卷，表示掛載Pod所在宿主機的目錄
 66         path: string		#Pod所在宿主機的目錄，將被用於同期中mount的目錄
 67       secret:			#類型為secret的存儲卷，掛載集群與定義的secre對象到容器內部
 68         scretname: string
 69         items:
 70         - key: string
 71           path: string
 72       configMap:			#類型為configMap的存儲卷，掛載預定義的configMap對象到容器內部
 73         name: string
 74         items:
 75         - key: string
 76           path: string

二 Pod的基本用法

2.1 創建Pod

Pod可以由1個或多個容器組合而成，通常對於緊耦合的兩個應用，應該組合成一個整體對外提供服務，則應該將這兩個打包為一個pod。

屬於一個Pod的多個容器應用之間相互訪問只需要通過localhost即可通信，這一組容器被綁定在一個環境中。

  1 [root@k8smaster01 study]# vi frontend-localredis-pod.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: redis-php
  6   label:
  7     name: redis-php
  8 spec:
  9   containers:
 10   - name: frontend
 11     image: kubeguide/guestbook-php-frontend:localredis
 12     ports:
 13     - containersPort: 80
 14   - name: redis-php
 15     image: kubeguide/redis-master
 16     ports:
 17     - containersPort: 6379
 18 
 19 [root@k8smaster01 study]# kubectl create -f frontend-localredis-pod.yaml
 20 

2.2 查看Pod

  1 [root@k8smaster01 study]# kubectl get pods	                #READY為2/2，表示此Pod中運行了yaml定義的兩個容器
  2 NAME        READY   STATUS    RESTARTS   AGE
  3 redis-php   2/2     Running   0          7m45s
  4 [root@k8smaster01 study]# kubectl describe pod redis-php	#查看詳細信息
  5 

三 靜態Pod

3.1 靜態Pod概述

靜態pod是由kubelet進行管理的僅存在於特定Node的Pod上，他們不能通過API Server進行管理，無法與ReplicationController、Deployment或者DaemonSet進行關聯，並且kubelet無法對他們進行健康檢查。靜態Pod總是由kubelet進行創建，並且總是在kubelet所在的Node上運行。

創建靜態Pod有兩種方式：配置文件或者HTTP方式。

3.2 配置文件方式創建

  1 [root@k8snode01 ~]# mkdir -p /etc/kubelet.d
  2 [root@k8snode01 ~]# vi /etc/kubelet.d/static-web.yaml
  3 apiVersion: v1
  4 kind: Pod
  5 metadata:
  6   name: static-web
  7   label:
  8     name: static-web
  9 spec:
 10   containers:
 11   - name: static-web
 12     image: nginx
 13     ports:
 14     - name: web
 15       containersPort: 80
 16 
 17 [root@k8snode01 ~]# vi /etc/systemd/system/kubelet.service
 18 ……
 19   --config=/etc/kubelet.d/ \·				#加入此參數
 20 ……
 21 [root@k8snode01 ~]# systemctl daemon-reload
 22 [root@k8snode01 ~]# systemctl restart kubelet.service	#重啟kubelet
 23 [root@k8snode01 ~]# docker ps				#查看創建的pod

提示：由於靜態pod不能通過API Server進行管理，因此在Master節點執行刪除操作後會變為Pending狀態，且無法刪除。刪除該pod只能在其運行的node上，將定義POD的yaml刪除。

3.3 HTTP方式

通過設置kubelet的啟動參數–mainfest-url，會定期從該URL下載Pod的定義文件，並以.yaml或.json文件的格式進行解析，從而創建Pod。

四 Pod容器共享Volume

4.1 共享Volume

在同一個Pod中的多個容器能夠共享Pod級別的存儲就Volume。Volume可以被定義為各種類型，多個容器各自進行掛載操作，將一個Volume掛載為容器內部需要的目錄。


示例1：

Pod級別設置Volume “app-logs”，同時Pod包含兩個容器，Tomcat向該Volume寫日誌，busybox讀取日誌文件。

  1 [root@k8smaster01 study]# vi pod-volume-applogs.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: volume-pod
  6 spec:
  7   containers:
  8   - name: tomcat
  9     image: tomcat
 10     ports:
 11     - containerPort: 8080
 12     volumeMounts:
 13     - name: app-logs
 14       mountPath: /usr/local/tomcat/logs
 15   - name: logreader
 16     image: busybox
 17     command: ["sh","-c","tail -f /logs/catalina*.log"]
 18     volumeMounts:
 19     - name: app-logs
 20       mountPath: /logs
 21   volumes:
 22   - name: app-logs
 23     emptyDir: {}

解釋：

Volume名：app-logs；

emptyDir：為Pod分配到Node的時候創建。無需指定宿主機的目錄文件，為Kubernetes自動分配的目錄。

  1 [root@k8smaster01 study]# kubectl create -f pod-volume-applogs.yaml	#創建
  2 [root@k8smaster01 study]# kubectl get pods				#查看
  3 [root@k8smaster01 study]# kubectl logs volume-pod -c busybox	#讀取log

  1 [root@k8smaster01 study]# kubectl exec -it volume-pod -c tomcat -- ls /usr/local/tomcat/logs
  2 catalina.2019-06-29.log      localhost_access_log.2019-06-29.txt
  3 host-manager.2019-06-29.log  manager.2019-06-29.log
  4 localhost.2019-06-29.log
  5 [root@k8smaster01 study]# kubectl exec -it volume-pod -c tomcat -- tail /usr/local/tomcat/logs/catalina.2019-06-29.log

提示：通過tomcat容器可查看日誌，對比busybox通過共享Volume查看的日誌是否一致。

五 Pod配置管理

5.1 Pod配置概述

應用部署的一個最佳實踐是將應用所需的配置信息與程序進行分離，使程序更加靈活。將相應的應用打包為鏡像，可以通過環境變量或者外掛volume的方式在創建容器的時候進行配置注入，從而實現更好的復用。

Kubernetes提供一種統一的應用配置管理方案：ConfigMap。

5.2 ConfigMap概述

ConfigMap供容器使用的主要場景：

• 生成容器內部的環境變量；
• 設置容器的啟動命令的參數（需設置為環境變量）；
• 以volume的形式掛載為容器內部的文件或者目錄。

ConfigMap以一個或多個key:value的形式定義。value可以是string也可以是一個文件內容，可以通過yaml配置文件或者通過kubectl create configmap 的方式創建configMap。

5.3 創建ConfigMap資源對象——yaml方式

  1 [root@k8smaster01 study]# vi cm-appvars.yaml
  2 apiVersion: v1
  3 kind: ConfigMap
  4 metadata:
  5   name: cm-appvars
  6 data:
  7   apploglevel: info
  8   appdatadir: /var/data
  9 
 10 [root@k8smaster01 study]# kubectl create -f cm-appvars.yaml
 11 configmap/cm-appvars created
 12 [root@k8smaster01 study]# kubectl get configmaps
 13 NAME         DATA   AGE
 14 cm-appvars   2      8s
 15 [root@k8smaster01 study]# kubectl describe configmaps cm-appvars

  1 [root@k8smaster01 study]# kubectl get configmaps cm-appvars -o yaml

5.4 創建ConfigMap資源對象——命令行方式

語法1

  1 # kubectl create configmap NAME --from-file=[key=]source --from-file=[key=]source

解釋：通過–from-file參數從文件中創建，可以指定key名稱，也可以制定多個key。

語法2

  1 # kubectl create configmap NAME --from-file=config-files-dir

解釋：通過–from-file參數從目錄中創建，該目錄下的每個配置文件名都被設置為key，文件的內容被設置為value。

語法3

  1 # kubectl create configmap NAME --from-literal=key1=value1 --from-literal=key2=value2

解釋：通過–from-literal參數從文本中創建，直接將指定的key#=value#創建為ConfigMap的內容。

5.5 Pod使用ConfigMap

容器應用使用ConfigMap有兩種方式：

• 通過環境變量獲取ConfigMap中的內容；
• 通過Volume掛載的方式將ConfigMap中的內容掛載為容器內容的文件或目錄。

  1 [root@k8smaster01 study]# vi cm-test-pod.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: cm-test-pod
  6 spec:
  7   containers:
  8   - name: cm-test
  9     image: busybox
 10     command: ["/bin/sh","-c","env|grep APP"]	#容器里執行查看環境變量的命令
 11     env:
 12     - name: APPLOGLEVEL				#定義容器環境變量名稱
 13       valueFrom:
 14         configMapKeyRef:			#環境變量的值來自ConfigMap
 15           name: cm-appvars			#指定來自cm-appvars的ConfigMap
 16           key: apploglevel			#key為apploglevel
 17     - name: APPDATADIR
 18       valueFrom:
 19         configMapKeyRef:
 20           name: cm-appvars
 21           key: appdatadir
 22 
 23 [root@k8smaster01 study]# kubectl create -f cm-test-pod.yaml
 24 [root@k8smaster01 study]# kubectl get pods
 25 NAME          READY   STATUS      RESTARTS   AGE
 26 cm-test-pod   0/1     Completed   2          24s

【掛載形式-待補充】

5.6 ConfigMap限制

• Configmap必須在pod創建之間創建；
• ConfigMap受到namespace的限制，只有同一個命名空間下才能引用；
• ConfigMap暫時無法配置配額；
• 靜態的pod無法使用ConfigMap；
• 在使用volumeMount掛載的時候，configMap基於items創建的文件會整體的將掛載數據卷的容器的目錄下的文件全部覆蓋。

六 Pod獲取自身信息

6.1 Downward API

pod擁有唯一的名字、IP地址，並且處於某個Namespace中。pod的容器內獲取pod的信息科通過Downward API實現。具體有以下兩種方式：

• 環境變量：用於單個變量，可以將pod信息和container信息注入容器內部；
• volume掛載：將數組類信息生成為文件，掛載至容器內部。

舉例1：通過Downward API將Pod的IP、名稱和所在的Namespace注入容器的環境變量。

  1 [root@k8smaster01 study]# vi dapi-test-pod.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: dapi-test-pod
  6 spec:
  7   containers:
  8     - name: test-container
  9       image: busybox
 10       command: [ "/bin/sh", "-c", "env" ]
 11       env:
 12         - name: MY_POD_NAME
 13           valueFrom:
 14             fieldRef:
 15               fieldPath: metadata.name
 16         - name: MY_POD_NAMESPACE
 17           valueFrom:
 18             fieldRef:
 19               fieldPath: metadata.namespace
 20         - name: MY_POD_IP
 21           valueFrom:
 22             fieldRef:
 23               fieldPath: status.podIP
 24   restartPolicy: Never

提示：Downward API提供如下變量：

metadata.name：Pod的名稱，當Pod通過RC生成時，其名稱是RC隨機產生的唯一名稱；

status.podIP：Pod的IP地址，POd的IP屬於狀態數據，而非元數據；

metadata.namespace：Pod所在的namespace。

  1 [root@k8smaster01 study]# kubectl create -f dapi-test-pod.yaml
  2 [root@k8smaster01 study]# kubectl logs dapi-test-pod | grep MY_POD
  3 MY_POD_NAMESPACE=default
  4 MY_POD_IP=172.30.240.4
  5 MY_POD_NAME=dapi-test-pod
  6 

舉例2：通過Downward API將Container的自願請求和限制信息注入容器的環境變量。

  1 [root@k8smaster01 study]# vi dapi-test-pod-container-vars.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: dapi-test-pod-container-vars
  6 spec:
  7   containers:
  8     - name: test-container
  9       image: busybox
 10       imagePullPolicy: Never
 11       command: [ "/bin/sh", "-c" ]
 12       args:
 13       - while true; do
 14           echo -en '\n';
 15           printenv MY_CPU_REQUEST MY_CPU_LIMIT;
 16           printenv MY_MEM_REQUEST MY_MEM_LIMIT;
 17           sleep 3600;
 18         done;
 19       resources:
 20         requests:
 21           memory: "32Mi"
 22           cpu: "125m"
 23         limits:
 24           memory: "64Mi"
 25           cpu: "250m"
 26       env:
 27         - name: MY_CPU_REQUEST
 28           valueFrom:
 29             resourceFieldRef:
 30               containerName: test-container
 31               resource: requests.cpu
 32         - name: MY_CPU_LIMIT
 33           valueFrom:
 34             resourceFieldRef:
 35               containerName: test-container
 36               resource: limits.cpu
 37         - name: MY_MEM_REQUEST
 38           valueFrom:
 39             resourceFieldRef:
 40               containerName: test-container
 41               resource: requests.memory
 42         - name: MY_MEM_LIMIT
 43           valueFrom:
 44             resourceFieldRef:
 45               containerName: test-container
 46               resource: limits.memory
 47   restartPolicy: Never

提示：Downward API提供如下變量：

requests.cpu：容器的CPU請求值；

limits.cpu：容器的CPU限制值；

requests.memory：容器的內存請求值；

limits.memory：容器的內存限制值。

  1 [root@k8smaster01 study]# kubectl create -f dapi-test-pod-container-vars.yaml
  2 [root@k8smaster01 study]# kubectl logs dapi-test-pod-container-vars
  3 1
  4 1
  5 33554432
  6 67108864

舉例3：通過Downward API將Pod的Label、Annotation列表通過Volume掛載為容器內的一個文件。

  1 [root@k8smaster01 study]# vi dapi-test-pod-volume.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: dapi-test-pod-volume
  6   labels:
  7     zone: us-est-coast
  8     cluster: test-cluster1
  9     rack: rack-22
 10   annotations:
 11     build: two
 12     builder: john-doe
 13 spec:
 14   containers:
 15     - name: test-container
 16       image: busybox
 17       imagePullPolicy: Never
 18       command: [ "/bin/sh", "-c" ]
 19       args:
 20       - while true; do
 21           if [[ -e /etc/labels ]]; then
 22             echo -en '\n\n'; cat /etc/labels; fi;
 23           if [[ -e /etc/annotations ]]; then
 24             echo -en '\n\n'; cat /etc/annotations; fi;
 25           sleep 3600;
 26         done;
 27       volumeMounts:
 28         - name: podinfo
 29           mountPath: /etc
 30           readOnly: false
 31   volumes:
 32     - name: podinfo
 33       downwardAPI:
 34         items:
 35           - path: "labels"
 36             fieldRef:
 37               fieldPath: metadata.labels
 38           - path: "annotations"
 39             fieldRef:
 40               fieldPath: metadata.annotations

注意：Volume中的ddownwardAPI的items語法，將會以path的名稱生成文件。如上所示，會在容器內生產/etc/labels和/etc/annotations兩個文件，分別包含metadata.labels和metadata.annotations的全部Label。

  1 [root@k8smaster01 study]# kubectl create -f dapi-test-pod-volume.yaml
  2 [root@k8smaster01 study]# kubectl logs dapi-test-pod-volume
  3 

提示：DownwardAPI意義：

在某些集群中，集群中的每個節點需要將自身的標識（ID）及進程綁定的IP地址等信息事先寫入配置文件中，進程啟動時讀取此類信息，然後發布到某個類似註冊服務中心。此時可通過DowanwardAPI，將一個預啟動腳本或Init Container，通過環境變量或文件方式獲取Pod自身的信息，然後寫入主程序配置文件中，最後啟動主程序。本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

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

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

※試算大陸海運運費!

前言

學習Netty也有一段時間了，Netty作為一個高性能的異步框架，很多RPC框架也運用到了Netty中的知識，在rpc框架中豐富的數據協議及編解碼可以讓使用者更加青睞；
Netty支持豐富的編解碼框架，其本身內部提供的編解碼也可以應對各種業務場景；
今天主要就是學習下Netty中提供的編、解碼類，之前只是簡單的使用了下Netty提供的解碼類，今天更加深入的研究下Netty中編、解碼的源碼及部分使用。

編、解碼的概念

• 編碼（Encoder）

  編碼就是將我們發送的數據編碼成字節數組方便在網絡中進行傳輸，類似Java中的序列化，將對象序列化成字節傳輸

• 解碼（Decoder）

  解碼和編碼相反，將傳輸過來的字節數組轉化為各種對象來進行展示等，類似Java中的反序列化
  如：
  // 將字節數組轉化為字符串
  new String(byte bytes[], Charset charset)

編、解碼超類

ByteToMessageDecoder: 解碼超類，將字節轉換成消息

解碼解碼一般用於將獲取到的消息解碼成系統可識別且自己需要的數據結構；因此ByteToMessageDecoder需要繼承ChannelInboundHandlerAdapter入站適配器來獲取到入站的數據，在handler使用之前通過channelRead獲取入站數據進行一波解碼；

ByteToMessageDecoder類圖

源碼分析

通過channelRead獲取入站數據，將數據緩存至cumulation數據緩衝區，最後在傳給decode進行解碼，在read完成之後清空緩存的數據

1. 獲取入站數據

/**
*  通過重寫channelRead方法來獲取入站數據
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 檢測是否是byteBuf對象格式數據
    if (msg instanceof ByteBuf) {
        // 實例化字節解碼成功輸出集合 即List<Object> out
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            // 獲取到的請求的數據
            ByteBuf data = (ByteBuf) msg;
            // 如果緩衝數據區為空則代表是首次觸發read方法
            first = cumulation == null;
            if (first) {
                // 如果是第一次read則當前msg數據為緩衝數據
                cumulation = data;
            } else {
                // 如果不是則觸發累加，將緩衝區的舊數據和新獲取到的數據通過        expandCumulation 方法累加在一起存入緩衝區cumulation
                // cumulator 累加類，將緩衝池中數據和新數據進行組合在一起
                // private Cumulator cumulator = MERGE_CUMULATOR;
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            // 將緩衝區數據cumulation進行解碼
            callDecode(ctx, cumulation, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable t) {
            throw new DecoderException(t);
        } finally {
            // 在解碼完畢后釋放引用和清空全局字節緩衝區
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
                // discardAfterReads為netty中設置的讀取多少次后開始丟棄字節 默認值16
                // 可通過setDiscardAfterReads(int n)來設置值不設置默認16次
            } else if (++ numReads >= discardAfterReads) {
                // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                // 在我們讀取了足夠的數據可以嘗試丟棄一些字節已保證不出現內存溢出的異常
                // 
                // See https://github.com/netty/netty/issues/4275
                // 讀取次數重置為0
                numReads = 0;
                // 重置讀寫指針或丟棄部分已讀取的字節
                discardSomeReadBytes();
            }
            // out為解碼成功的傳遞給下一個handler
            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            // 結束當前read傳遞到下個ChannelHandler
            fireChannelRead(ctx, out, size);
            // 回收響應集合 將insertSinceRecycled設置為false；
            // insertSinceRecycled用於channelReadComplete判斷使用
            out.recycle();
        }
    } else {
        // 不是的話直接fire傳遞給下一個handler
        ctx.fireChannelRead(msg);
    }
}

2. 初始化字節緩衝區計算器: Cumulator主要用於全局字節緩衝區和新讀取的字節緩衝區組合在一起擴容

public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
    
    /**
    * alloc ChannelHandlerContext分配的字節緩衝區
    * cumulation 當前ByteToMessageDecoder類全局的字節緩衝區
    * in 入站的字節緩衝區
    **/
    @Override
    public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        final ByteBuf buffer;
        // 如果全局ByteBuf寫入的字節+當前入站的字節數據大於全局緩衝區最大的容量或者全局緩衝區的引用數大於1個或全局緩衝區只讀
        if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
            || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
            // Expand cumulation (by replace it) when either there is not more room in the buffer
            // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
            // duplicate().retain() or if its read-only.
            //
            // See:
            // - https://github.com/netty/netty/issues/2327
            // - https://github.com/netty/netty/issues/1764
            // 進行擴展全局字節緩衝區（容量大小 = 新數據追加到舊數據末尾組成新的全局字節緩衝區）
            buffer = expandCumulation(alloc, cumulation, in.readableBytes());
        } else {
            buffer = cumulation;
        }
        // 將新數據寫入緩衝區
        buffer.writeBytes(in);
        // 釋放當前的字節緩衝區的引用
        in.release();
        
        return buffer;
    }
};


/**
* alloc 字節緩衝區操作類
* cumulation 全局累加字節緩衝區
* readable 讀取到的字節數長度
*/
// 字節緩衝區擴容方法
static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
    // 舊數據
    ByteBuf oldCumulation = cumulation;
    // 通過ByteBufAllocator將緩衝區擴大到oldCumulation + readable大小
    cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
    // 將舊數據重新寫入到新的字節緩衝區
    cumulation.writeBytes(oldCumulation);
    // 舊字節緩衝區引用-1
    oldCumulation.release();
    return cumulation;
}

3. ByteBuf釋放當前字節緩衝區的引用: 通過調用ReferenceCounted接口中的release方法來釋放

@Override
public boolean release() {
    return release0(1);
}

@Override
public boolean release(int decrement) {
    return release0(checkPositive(decrement, "decrement"));
}

/**
* decrement 減量
*/
private boolean release0(int decrement) {
    for (;;) {
        int refCnt = this.refCnt;
        // 當前引用小於減量
        if (refCnt < decrement) {
            throw new IllegalReferenceCountException(refCnt, -decrement);
        }
        // 這裏就利用里線程併發中的知識CAS，線程安全的設置refCnt的值
        if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
            // 如果減量和引用量相等
            if (refCnt == decrement) {
                // 全部釋放
                deallocate();
                return true;
            }
            return false;
        }
    }
}

4. 將全局字節緩衝區進行解碼

/**
* ctx ChannelHandler的上下文，用於傳輸數據與下一個handler來交互
* in 入站數據
* out 解析之後的出站集合 （此出站不是返回給客戶端的而是傳遞給下個handler的）
*/
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    try {
        // 如果入站數據還有沒解析的
        while (in.isReadable()) {
            // 解析成功的出站集合長度
            int outSize = out.size();
            // 如果大於0則說明解析成功的數據還沒被消費完，直接fire掉給通道中的後續handler繼續                消費
            if (outSize > 0) {
                fireChannelRead(ctx, out, outSize);
                out.clear();

                // Check if this handler was removed before continuing with decoding.
                // 在這個handler刪除之前檢查是否還在繼續解碼
                // If it was removed, it is not safe to continue to operate on the buffer.
                // 如果移除了，它繼續操作緩衝區是不安全的
                //
                // See:
                // - https://github.com/netty/netty/issues/4635
                if (ctx.isRemoved()) {
                    break;
                }
                outSize = 0;
            }
            // 入站數據字節長度
            int oldInputLength = in.readableBytes();
            // 開始解碼數據
            decodeRemovalReentryProtection(ctx, in, out);

            // Check if this handler was removed before continuing the loop.
            // 
            // If it was removed, it is not safe to continue to operate on the buffer.
            //
            // See https://github.com/netty/netty/issues/1664
            if (ctx.isRemoved()) {
                break;
            }

            // 解析完畢跳出循環
            if (outSize == out.size()) {
                if (oldInputLength == in.readableBytes()) {
                    break;
                } else {
                    continue;
                }
            }

            if (oldInputLength == in.readableBytes()) {
                throw new DecoderException(
                    StringUtil.simpleClassName(getClass()) +
                    ".decode() did not read anything but decoded a message.");
            }

            if (isSingleDecode()) {
                break;
            }
        }
    } catch (DecoderException e) {
        throw e;
    } catch (Throwable cause) {
        throw new DecoderException(cause);
    }
}

final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 設置解碼狀態為正在解碼  STATE_INIT = 0; STATE_CALLING_CHILD_DECODE = 1;             STATE_HANDLER_REMOVED_PENDING = 2; 分別為初始化； 解碼； 解碼完畢移除
        decodeState = STATE_CALLING_CHILD_DECODE;
        try {
            // 具體的解碼邏輯（netty提供的解碼器或自定義解碼器中重寫的decode方法）
            decode(ctx, in, out);
        } finally {
            // 此時decodeState為正在解碼中 值為1，返回false
            boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
            // 在設置為初始化等待解碼
            decodeState = STATE_INIT;
            // 解碼完成移除當前ChannelHandler標記為不處理
            // 可以看看handlerRemoved源碼。如果緩衝區還有數據直接傳遞給下一個handler
            if (removePending) {
                handlerRemoved(ctx);
            }
        }
    }

5. 執行channelReadComplete

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    // 讀取次數重置
    numReads = 0;
    // 重置讀寫index
    discardSomeReadBytes();
    // 在channelRead meth中定義賦值 decodeWasNull = !out.insertSinceRecycled();
    // out指的是解碼集合List<Object> out; 咱們可以點進
    if (decodeWasNull) {
        decodeWasNull = false;
        if (!ctx.channel().config().isAutoRead()) {
            ctx.read();
        }
    }
    // fire掉readComplete傳遞到下一個handler的readComplete
    ctx.fireChannelReadComplete();
}

/**
*  然後我們可以搜索下insertSinceRecucled在什麼地方被賦值了
* Returns {@code true} if any elements where added or set. This will be reset once {@link #recycle()} was called.
*/
boolean insertSinceRecycled() {
    return insertSinceRecycled;
}


// 搜索下insert的調用我們可以看到是CodecOutputList類即為channelRead中的out集合，眾所周知在    decode完之後，解碼數據就會被調用add方法，此時insertSinceRecycled被設置為true
private void insert(int index, Object element) {
    array[index] = element;
    insertSinceRecycled = true;
}


/**
* 清空回收數組內部的所有元素和存儲空間
* Recycle the array which will clear it and null out all entries in the internal storage.
*/
// 搜索recycle的調用我么可以知道在channelRead的finally邏輯中 調用了out.recycle();此時        insertSinceRecycled被設置為false
void recycle() {
    for (int i = 0 ; i < size; i ++) {
        array[i] = null;
    }
    clear();
    insertSinceRecycled = false;
    handle.recycle(this);
}

至此ByteToMessageDecoder解碼類應該差不多比較清晰了！！！

MessageToByteEncoder: 編碼超類，將消息轉成字節進行編碼發出

何謂編碼，就是將發送數據轉化為客戶端和服務端約束好的數據結構和格式進行傳輸，我們可以在編碼過程中將消息體body的長度和一些頭部信息有序的設置到ByteBuf字節緩衝區中；方便解碼方靈活的運用來判斷（是否完整的包等）和處理業務；解碼是繼承入站數據，反之編碼應該繼承出站的數據；接下來我們看看編碼類是怎麼進行編碼的；

MessageToByteEncoder類圖如下

源碼分析

既然是繼承出站類，我們直接看看write方法是怎麼樣的

/**
* 通過write方法獲取到出站的數據即要發送出去的數據
* ctx channelHandler上下文
* msg 發送的數據 Object可以通過繼承類指定的泛型來指定
* promise channelPromise異步監聽，類似ChannelFuture，只不過promise可以設置監聽的結果，future只能通過獲取監聽的成功失敗結果；可以去了解下promise和future的區別
*/
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        // 檢測發送數據的類型 通過TypeParameterMatcher類型匹配器
        if (acceptOutboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            I cast = (I) msg;
            // 分配字節緩衝區 preferDirect默認為true
            buf = allocateBuffer(ctx, cast, preferDirect);
            try {
                // 進行編碼
                encode(ctx, cast, buf);
            } finally {
                // 完成編碼后釋放對象的引用
                ReferenceCountUtil.release(cast);
            }
            // 如果緩衝區有數據則通過ctx發送出去，promise可以監聽數據傳輸並設置是否完成
            if (buf.isReadable()) {
                ctx.write(buf, promise);
            } else {
                // 如果沒有數據則釋放字節緩衝區的引用併發送一個empty的空包
                buf.release();
                ctx.write(Unpooled.EMPTY_BUFFER, promise);
            }
            buf = null;
        } else {
            // 非TypeParameterMatcher類型匹配器匹配的類型直接發送出去
            ctx.write(msg, promise);
        }
    } catch (EncoderException e) {
        throw e;
    } catch (Throwable e) {
        throw new EncoderException(e);
    } finally {
        if (buf != null) {
            buf.release();
        }
    }
}

// 初始化設置preferDirect為true
protected MessageToByteEncoder() {
    this(true);
}
protected MessageToByteEncoder(boolean preferDirect) {
    matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");
    this.preferDirect = preferDirect;
}

編碼： 重寫encode方法，根據實際業務來進行數據編碼

// 此處就是我們需要重寫的編碼方法了，我們和根據約束好的或者自己定義好想要的數據格式發送給對方

// 下面是我自己寫的demo的編碼方法；頭部設置好body的長度，服務端可以根據長度來判斷是否是完整的包，僅僅自學寫的簡單的demo非正常線上運營項目的邏輯
public class MyClientEncode extends MessageToByteEncoder<String> {

    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        if (null != msg) {
            byte[] request = msg.getBytes(Charset.forName("UTF-8"));
            out.writeInt(request.length);
            out.writeBytes(request);
        }
    }
}

編碼類相對要簡單很多，因為只需要將發送的數據序列化，按照一定的格式進行發送數據！！！

項目實戰

項目主要簡單的實現下自定義編解碼器的運用及LengthFieldBasedFrameDecoder的使用

• 項目結構如下

  │  hetangyuese-netty-06.iml
  │  pom.xml
  │
  ├─src
  │  ├─main
  │  │  ├─java
  │  │  │  └─com
  │  │  │      └─hetangyuese
  │  │  │          └─netty
  │  │  │              ├─client
  │  │  │              │      MyClient06.java
  │  │  │              │      MyClientChannelInitializer.java
  │  │  │              │      MyClientDecoder.java
  │  │  │              │      MyClientEncode.java
  │  │  │              │      MyClientHandler.java
  │  │  │              │      MyMessage.java
  │  │  │              │
  │  │  │              └─server
  │  │  │                      MyChannelInitializer.java
  │  │  │                      MyServer06.java
  │  │  │                      MyServerDecoder.java
  │  │  │                      MyServerDecoderLength.java
  │  │  │                      MyServerEncoder.java
  │  │  │                      MyServerHandler.java
  │  │  │
  │  │  └─resources
  │  └─test
  │      └─java
  

• 服務端

  Serverhandler: 只是簡單的將解碼的內容輸出

  public class MyServerHandler extends ChannelInboundHandlerAdapter {
  
      @Override
      public void channelActive(ChannelHandlerContext ctx) throws Exception {
          System.out.println("客戶端連接成功 time: " + new Date().toLocaleString());
      }
  
      @Override
      public void channelInactive(ChannelHandlerContext ctx) throws Exception {
          System.out.println("客戶端斷開連接 time: " + new Date().toLocaleString());
      }
  
      @Override
      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
          String body = (String) msg;
          System.out.println("content:" + body);
      }
  
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
          // 出現異常關閉通道
          cause.printStackTrace();
          ctx.close();
      }
  }

  解碼器

  public class MyServerDecoder extends ByteToMessageDecoder {
  
      // 此處我頭部只塞了長度字段佔4個字節，別問為啥我知道，這是要客戶端和服務端約束好的
      private static int min_head_length = 4;
  
      @Override
      protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
          // 解碼的字節長度
          int size = in.readableBytes();
          if(size < min_head_length) {
              System.out.println("解析的數據長度小於頭部長度字段的長度");
              return ;
          }
          // 讀取的時候指針已經移位到長度字段的尾端
          int length = in.readInt();
          if (size < length) {
              System.out.println("解析的數據長度與長度不符合");
              return ;
          }
  
          // 上面已經讀取到了長度字段，後面的長度就是body
          ByteBuf decoderArr = in.readBytes(length);
          byte[] request = new byte[decoderArr.readableBytes()];
          // 將數據寫入空數組
          decoderArr.readBytes(request);
          String body = new String(request, Charset.forName("UTF-8"));
          out.add(body);
      }
  }

  將解碼器加入到channelHandler中：記得加到業務handler的前面否則無效

  public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
  
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline()
  //                .addLast(new MyServerDecoderLength(10240, 0, 4, 0, 0))
  //                .addLast(new LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0))
                  .addLast(new MyServerDecoder())
                  .addLast(new MyServerHandler())
          ;
      }
  }

• 客戶端

  ClientHandler

  public class MyClientHandler extends ChannelInboundHandlerAdapter {
  
      @Override
      public void channelActive(ChannelHandlerContext ctx) throws Exception {
          System.out.println("與服務端連接成功");
          for (int i = 0; i<10; i++) {
              ctx.writeAndFlush("hhhhh" + i);
          }
      }
  
      @Override
      public void channelInactive(ChannelHandlerContext ctx) throws Exception {
          System.out.println("與服務端斷開連接");
      }
  
      @Override
      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
          System.out.println("收到服務端消息:" +msg+ " time: " + new Date().toLocaleString());
      }
  
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
          cause.printStackTrace();
          ctx.close();
      }
  }

  編碼器

  public class MyClientEncode extends MessageToByteEncoder<String> {
  
      @Override
      protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
          if (null != msg) {
              byte[] request = msg.getBytes(Charset.forName("UTF-8"));
              out.writeInt(request.length);
              out.writeBytes(request);
          }
      }
  }

  將編碼器加到ClientHandler的前面

  public class MyClientChannelInitializer extends ChannelInitializer<SocketChannel> {
  
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline()
                  .addLast(new MyClientDecoder())
                  .addLast(new MyClientEncode())
                  .addLast(new MyClientHandler())
          ;
  
      }
  }

• 服務端運行結果

  MyServer06 is start ...................
  客戶端連接成功 time: 2019-11-19 16:35:47
  content:hhhhh0
  content:hhhhh1
  content:hhhhh2
  content:hhhhh3
  content:hhhhh4
  content:hhhhh5
  content:hhhhh6
  content:hhhhh7
  content:hhhhh8
  content:hhhhh9

• 如果不用自定義的解碼器怎麼獲取到body內容呢

  將自定義編碼器換成LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0)

  public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
  
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline()
  //                .addLast(new MyServerDecoderLength(10240, 0, 4, 0, 0))
                  .addLast(new LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0))
  //                .addLast(new MyServerDecoder())
                  .addLast(new MyServerHandler())
          ;
      }
  }
  
  // 怕忘記的各個參數的含義在這在說明一次，自己不斷的修改每個值觀察結果就可以更加深刻的理解
  /**
  * maxFrameLength：消息體的最大長度，好像默認最大值為1024*1024
  * lengthFieldOffset 長度字段所在字節數組的下標 （我這是第一個write的所以下標是0）
  * lengthFieldLength 長度字段的字節長度（int類型佔4個字節）
  * lengthAdjustment 長度字段補償的數值 （lengthAdjustment =  數據包長度 - lengthFieldOffset - lengthFieldLength - 長度域的值），解析需要減去對應的數值
  * initialBytesToStrip 是否去掉長度字段（0不去除，對應長度域字節長度）
  */
  public LengthFieldBasedFrameDecoder(
              int maxFrameLength,
              int lengthFieldOffset, int lengthFieldLength,
              int lengthAdjustment, int initialBytesToStrip)

  結果: 前都帶上了長度

  MyServer06 is start ...................
  客戶端連接成功 time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh0, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh1, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh2, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh3, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh4, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh5, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh6, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh7, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh8, time: 2019-11-19 17:53:42
  收到客戶端發來的消息:   hhhhh9, time: 2019-11-19 17:53:42

  如果我們在客戶端的長度域中做手腳 LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 0)

  舊： out.writeInt(request.length);
  新： out.writeInt(request.length + 1);

  // 看結果就不正常，0後面多了一個0；但是不知道為啥只解碼了一次？？？ 求解答
  MyServer06 is start ...................
  客戶端連接成功 time: 2019-11-19 17:56:55
  收到客戶端發來的消息:   hhhhh0 , time: 2019-11-19 17:56:55
  
  // 正確修改為 LengthFieldBasedFrameDecoder(10240, 0, 4, -1, 0)
  // 結果：
  MyServer06 is start ...................
  客戶端連接成功 time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh0, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh1, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh2, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh3, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh4, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh5, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh6, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh7, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh8, time: 2019-11-19 18:02:18
  收到客戶端發來的消息:   hhhhh9, time: 2019-11-19 18:02:18

  捨棄長度域 ：LengthFieldBasedFrameDecoder(10240, 0, 4, 0, 4)

  // 結果
  MyServer06 is start ...................
  客戶端連接成功 time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh0, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh1, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh2, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh3, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh4, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh5, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh6, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh7, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh8, time: 2019-11-19 18:03:44
  收到客戶端發來的消息:hhhhh9, time: 2019-11-19 18:03:44

  分析源碼示例中的 lengthAdjustment = 消息字節長度 – lengthFieldOffset-lengthFieldLength-長度域中的值

• 源碼中的示例

   * <pre>
   * lengthFieldOffset   =  0
   * lengthFieldLength   =  2
   * <b>lengthAdjustment</b>    = <b>-2</b> (= the length of the Length field)
   * initialBytesToStrip =  0
   *
   * BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
   * +--------+----------------+      +--------+----------------+
   * | Length | Actual Content |----->| Length | Actual Content |
   * | 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
   * +--------+----------------+      +--------+----------------+
   * </pre>

  長度域中0x000E為16進制，轉換成10進制是14，說明消息體長度為14；根據公式：14-0-2-14 = -2

  * <pre>
   * lengthFieldOffset   = 0
   * lengthFieldLength   = 3
   * <b>lengthAdjustment</b>    = <b>2</b> (= the length of Header 1)
   * initialBytesToStrip = 0
   *
   * BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
   * +----------+----------+----------------+      +----------+----------+----------------+
   * |  Length  | Header 1 | Actual Content |----->|  Length  | Header 1 | Actual Content |
   * | 0x00000C |  0xCAFE  | "HELLO, WORLD" |      | 0x00000C |  0xCAFE  | "HELLO, WORLD" |
   * +----------+----------+----------------+      +----------+----------+----------------+
   * </pre>

  從上的例子可以知道；lengthAdjustment(2) = 17- 12（00000C）-lengthFieldOffset(0) - lengthFieldLength(3);

  …….等等

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

【其他文章推薦】

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

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

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

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

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

※試算大陸海運運費!