特斯拉超級電池工廠Gigafactory 7月正式開業 目前仍在建

據外媒體報導,特斯拉稱其位於內華達州里諾市郊的超級電池工廠Gigafactory將於7月29日晚上舉行開業典禮,目前該工廠仍處於建設之中。

特斯拉在發給客戶的一封電子郵件中披露,為特斯拉成功推薦的客戶超過5位元的人將會受邀參加此次開業典禮。特斯拉已經通過電子郵件向這些符合資格的客戶通知了開業典禮時間。

事實上,在未來2個月內Gigafactory工廠不可能完全竣工,特斯拉僅僅是決定在已建成區域舉行小型典禮。在今年3月份,特斯拉、SolarCity在該廠區接待了內華達州議員,探討了太陽能業務在該州的發展。

據悉,Gigafactory工廠建設造價逾50億美元,到今年5月初僅建成14%。此外,松下向特斯拉的超級電池工廠供應電池生產設備,並提供部分資金。

儘管電池廠還在建設之中,但是工廠已經投產。工廠目前生產的太陽能電池被用於公共設施、企業以及各個地產業主。

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

【其他文章推薦】

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

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

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

小三通海運與一般國際貿易有何不同?

小三通快遞通關作業有哪些?

Zabbix-(五)監控Docker容器與自定義jvm監控項

Zabbix-(五)監控Docker容器與自定義jvm監控項

一.前言

前文中講述了Zabbix對服務器硬件方面的監控功能,本文將講述利用Zabbix監控Docker容器中的Java Web服務,並通過自定義監控項,監控JVM老年代使用情況以及GC信息。Zabbix其實提供了,自帶了JMX模板能夠直接監控JVM信息,本文主要側重於自定義參數與自定義監控項,關於JMX會在之後的文章中介紹。

準備

  • Zabbix Server (Zabbix 4.4) (ip:192.168.152.140)
  • 運行Java應用的主機 以下簡稱Server-A (已被Zabbix監控) (ip:192.168.152.142)

二.開啟agent用戶自定義參數配置

  1. 修改配置

    使用自定義參數時,首先需要修改Server-A的agent配置

    # vim /etc/zabbix/zabbix_agentd.conf

    修改配置 UnsafeUserParameters=1

    UnsafeUserParameters=1
  2. 重啟zabbix-agent

    # systemctl restart zabbix-agent

三.運行tomcat容器

在Server-A運行tomcat容器

# docker run --name tomcat -p 8080:8080 -dit tomcat:jdk8-adoptopenjdk-hotspot

將zabbix賬號添加到docker組。參考

# sudo gpasswd  -a zabbix docker

外部訪問測試一下

四.創建自定義Docker模板

我們可以定義一個比較通用的Docker模板,有服務需要被監控時,直接鏈接該模板即可。

  1. 創建群組

    點擊【配置】-【主機群組】-【創建主機群組】

    定義一個組名 Docker Group

    配置項
    * 組名 Docker Group
  2. 創建模板

    創建一個自定義模板,模板名稱Docker Template,選擇上步驟創建的Docker Group群組

    配置項
    * 模板名稱 Docker Template
    * 群組 Docker Group

五.編寫腳本與自定義監控參數

我們需要編寫一個腳本,用於發現當前正在運行的docker容器(這裏使用容器名稱)。

  1. 在Server-A編寫發現運行容器的python腳本

    創建腳本

    # cd /data/zabbix
    # touch find_container.py
    # chmod a+x find_container.py
    # vim find_container.py

    腳本內容:

    #!/usr/bin/env python
    import os
    import json
    
    # 查看當前運行的docker容器
    t=os.popen(""" docker ps  |grep -v 'CONTAINER ID'|awk {'print $NF'} """)
    container_name = []
    for container in  t.readlines():
            r = os.path.basename(container.strip())
            container_name += [{'{#CONTAINERNAME}':r}]
    # 轉換成json數據
    print json.dumps({'data':container_name},sort_keys=True,indent=4,separators=(',',':'))

    運行腳本,查看一下json數據格式:

    {
        "data":[
            {
                "{#CONTAINERNAME}":"tomcat"
            }
        ]
    }
  2. 在Server-A自定義容器發現參數

    我們需要自定義一個鍵值對的配置類型,以便Zabbix可以通過鍵讀取到值。

    增加自定義參數

    # cd /etc/zabbix/zabbix_agentd.d
    # vim userparameter_find_container.conf
    docker.container /data/zabbix/find_container.py (腳本的運行結果)
    UserParameter=docker.container,/data/zabbix/find_container.py
  3. 在Server-A創建查看容器JVM GC情況的腳本

    我們可以使用jstat -gcutil 命令查看GC情況

    創建python腳本

    # cd /data/zabbix
    # touch monitor_gc.py
    # chmod a+x monitor_gc.py
    # vim monitor_gc.py

    腳本內容

    #!/usr/bin/python
    import sys
    import os
    
    def monitor_gc(container_name, keyword):
            cmd = ''' docker exec %s bash -c "jstat -gcutil 1" | grep -v S0 | awk '{print $%s}' ''' %(container_name, keyword)
            value = os.popen(cmd).read().replace("\n","")
            print value
    
    if __name__ == '__main__':
            # 參數1:容器的名稱
            # 參數2:查看第幾列(例如 Eden區在第3列傳入3,Full GC次數在第9列傳入9)
            container_name, keyword = sys.argv[1], sys.argv[2]
            monitor_gc(container_name, keyword)

    測試腳本,查看當前tomcat容器Full GC次數

    # /data/zabbix/monitor_gc.py 'tomcat' '9'

  4. 在Server-A自定義Zabbix JVM GC參數

    同樣,增加一個conf文件,表示自定義參數

    # cd /etc/zabbix/zabbix_agentd.d
    # touch userparameter_gc_status.conf
    # vim userparameter_gc_status.conf
    jvm.gc.status[*] /data/zabbix/monitor_gc.py $1 $2
    UserParameter=jvm.gc.status[*], /data/zabbix/monitor_gc.py $1 $2

    jvm.gc.status[*] 表示可以使用參數。其中$1表示參數1,即容器名稱;$2表示參數2,需要查看哪項GC信息,$1 $2都是通過Zabbix配置時傳遞的。

  5. 在Zabbix server上測試自定義參數

    為zabbix sever安裝zabbix-get

    # yum install -y zabbix-get

    測試自定義參數,如果有權限問題,可以參考

    # zabbix_get -s 192.168.152.142 -p 10050 -k docker.container
    # zabbix_get -s 192.168.152.142 -p 10050 -k "jvm.gc.status['tomcat', 9]"

六.Zabbix模板增加自動發現規則

上述配置中,已經可以通過腳本獲取到已運行的容器信息,此步驟將通過Zabbix配置界面,在模板中添加自動發現規則,以發現被監控主機中正在運行的docker容器,並利用這些獲取的數據進一步監控容器中jvm數據。

  1. 創建自動發現規則

    點擊【配置】-【模板】-【Docker Template】

    點擊【自動發現規則】-【創建發現規則】

    先配置【自動發現規則】

    配置項
    * 名稱 發現正在運行的Docker容器規則
    類型 Zabbix 客戶端
    * 鍵值 docker.container (這是我們上述步驟中自定義的鍵值)
    其他配置 根據需要配置

    鍵值配置項是之前

    再配置【過濾器】

    則配置自定義腳本返回json數據中的

    配置項
    {#CONTAINERNAME}
  2. 添加監控項原型

    點擊新建的自動發現規則的【監控項原型】-【創建監控項原型】

    輸入參數

    配置項
    * 名稱 Tomcat Full GC次數監控項
    類型 Zabbix 客戶端
    * 鍵值 jvm.gc.status[{#CONTAINERNAME} , 9]
    其他配置項 根據需要填寫

    鍵值是定義的參數,{#CONTAINERNAME} 是jvm.gc.status的參數1,使用了自動發現規則,發現到的docker容器名稱(本文中即是 tomcat);參數2 9 則是表示需要查看FullGC次數,FGC列(第9列)

    除此之外,還可以添加Old老年代(對應第4列),Full GC時間(對應第10列)等監控項,這裏就不一一添加了,和上述過程基本一致,只需修改參數2即可(也可以利用剛新建的監控項原型進行【克隆】)。

七.鏈接模板

將上述鏈接到Server-A主機

八.DashBoard添加可視化圖形

回到Zabbix首頁可以為新增的自定義監控項,增加圖形(添加圖形步驟可以參考)

九.其他

部署問題

  • zabbix在執行腳本時,是使用的zabbix賬戶,因此可能要注意要給zabbix賬號賦予權限。

    例如,zabbix賬戶無法使用docker命令,將zabbix添加到docker組

    # sudo gpasswd -a zabbix docker
  • zabbix server無法執行agent自定義參數中的腳本

    為agent主機設置

    # setenforce 0

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

【其他文章推薦】

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

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

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

小三通海運與一般國際貿易有何不同?

小三通快遞通關作業有哪些?

021.掌握Pod-Pod調度策略

一 Pod生命周期管理

1.1 Pod生命周期

Pod在整個生命周期過程中被系統定義了如下各種狀態。

狀態值 描述
Pending API Server已經創建該Pod,且Pod內還有一個或多個容器的鏡像沒有創建,包括正在下載鏡像的過程。
Running Pod內所有容器均已創建,且至少有一個容器處於運行狀態、正在啟動狀態或正在重啟狀態。
Succeeded Pod內所有容器均成功執行退出,且不會重啟。
Failed Pod內所有容器均已退出,但至少有一個容器退出為失敗狀態。
Unknown 由於某種原因無法獲取該Pod狀態,可能由於網絡通信不暢導致。

1.2 Pod重啟策略

Pod重啟策略(RestartPolicy)應用於Pod內的所有容器,並且僅在Pod所處的Node上由kubelet進行判斷和重啟操作。當某個容器異常退出或者健康檢查失敗時,kubelet將根據RestartPolicy的設置來進行相應操作。
Pod的重啟策略包括Always、OnFailure和Never,默認值為Always。

  • Always:當容器失效時,由kubelet自動重啟該容器;
  • OnFailure:當容器終止運行且退出碼不為0時,由kubelet自動重啟該容器;
  • Never:不論容器運行狀態如何,kubelet都不會重啟該容器。

       kubelet重啟失效容器的時間間隔以sync-frequency乘以2n來計算,例如1/2/4/8倍等,最長延時5min,並且在成功重啟后的10min后重置該時間。

Pod的重啟策略與控制方式關聯,當前可用於管理Pod的控制器包括ReplicationController、Job、DaemonSet及直接管理kubelet管理(靜態Pod)。
不同控制器的重啟策略限制如下:

  • RC和DaemonSet:必須設置為Always,需要保證該容器持續運行;
  • Job:OnFailure或Never,確保容器執行完成后不再重啟;
  • kubelet:在Pod失效時重啟,不論將RestartPolicy設置為何值,也不會對Pod進行健康檢查。








Pod包含的容器數 Pod當前的狀態 發生事件 Pod的結果狀態
RestartPolicy=Always RestartPolicy=OnFailure RestartPolicy=Never
包含1個容器 Running 容器成功退出 Running Succeeded Succeeded
包含1個容器 Running 容器失敗退出 Running Running Failed
包括兩個容器 Running 1個容器失敗退出 Running Running Running
包括兩個容器 Running 容器被OOM殺掉 Running Running Failed

1.3 Pod健康檢查

對Pod的健康檢查可以通過兩類探針來檢查:LivenessProbe和ReadinessProbe。
LivenessProbe探針:用於判斷容器是否存活(running狀態),如果LivenessProbe探針探測到容器不健康,則kubelet將殺掉該容器,並根據容器的重啟策略做相應處理。若一個容器不包含LivenessProbe探針,kubelet認為該容器的LivenessProbe探針返回值用於是“Success”。
ReadineeProbe探針:用於判斷容器是否啟動完成(ready狀態)。如果ReadinessProbe探針探測到失敗,則Pod的狀態將被修改。Endpoint Controller將從Service的Endpoint中刪除包含該容器所在Pod的Eenpoint。
kubelet定期執行LivenessProbe探針來診斷容器的健康狀態,通常有以下三種方式:

  • ExecAction:在容器內執行一個命令,若返回碼為0,則表明容器健康。

示例:通過執行”cat /tmp/health”命令判斷一個容器運行是否正常。容器初始化並創建該文件,10s后刪除該文件,15s秒通過命令判斷,由於該文件已被刪除,因此判斷該容器Fail,導致kubelet殺掉該容器並重啟。

  1 [root@uk8s-m-01 study]# vi dapi-liveness.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: dapi-liveness-pod
  6   labels:
  7     test: liveness-exec
  8 spec:
  9   containers:
 10     - name: dapi-liveness
 11       image: busybox
 12       args:
 13       - /bin/sh
 14       - -c
 15       - echo ok > /tmp/health; sleep 10; rm -rf /tmp/health; sleep 600
 16       livenessProbe:
 17         exec:
 18           command:
 19           - cat
 20           - /tmp/health
 21 
 22 [root@uk8s-m-01 study]# kubectl describe pod dapi-liveness-pod

  • TCPSocketAction:通過容器的IP地址和端口號執行TCP檢查,若能建立TCP連接,則表明容器健康。

示例:

  1 [root@uk8s-m-01 study]# vi dapi-tcpsocket.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: dapi-healthcheck-tcp
  6 spec:
  7   containers:
  8     - name: nginx
  9       image: nginx
 10       ports:
 11       - containerPort: 80
 12       livenessProbe:
 13         tcpSocket:
 14           port: 80
 15         initialDelaySeconds: 30
 16         timeoutSeconds: 1
 17 
 18 [root@uk8s-m-01 study]# kubectl create -f dapi-tcpsocket.yaml


提示:對於每種探測方式,都需要設置如下兩個參數,其包含的含義如下:

initialDelaySeconds:啟動容器後進行首次健康檢查的等待時間,單位為s;

timeoutSeconds:健康檢查發送請求后等待響應的超時時間,單位為s,當超時發生時,kubelet會認為容器已經無法提供服務,將會重啟該容器。

二 Pod調度

Kubernetes中,Pod通常是容器的載體,一般需要通過Deployment、DaemonSet、RC、Job等對象來完成一組Pod的調度與自動控制功能。

2.1 Depolyment/RC自動調度

Deployment或RC的主要功能之一就是自動部署一個容器應用的多份副本,以及持續監控副本的數量,在集群內始終維持用戶指定的副本數量。
示例:

  1 [root@uk8s-m-01 study]# vi nginx-deployment.yaml
  2 apiVersion: apps/v1beta1
  3 kind: Deployment
  4 metadata:
  5   name: nginx-deployment-01
  6 spec:
  7   replicas: 3
  8   template:
  9     metadata:
 10       labels:
 11         app: nginx
 12     spec:
 13       containers:
 14       - name: nginx
 15         image: nginx:1.7.9
 16         ports:
 17         - containerPort: 80
 18 
 19 [root@uk8s-m-01 study]# kubectl get deployments
 20 NAME                  READY   UP-TO-DATE   AVAILABLE   AGE
 21 nginx-deployment-01   3/3     3            3           30s
 22 [root@uk8s-m-01 study]# kubectl get rs
 23 NAME                             DESIRED   CURRENT   READY   AGE
 24 nginx-deployment-01-5754944d6c   3         3         3       75s
 25 [root@uk8s-m-01 study]# kubectl get pod | grep nginx
 26 nginx-deployment-01-5754944d6c-hmcpg   1/1     Running     0          84s
 27 nginx-deployment-01-5754944d6c-mcj8q   1/1     Running     0          84s
 28 nginx-deployment-01-5754944d6c-p42mh   1/1     Running     0          84s

2.2 NodeSelector定向調度

當需要手動指定將Pod調度到特定Node上,可以通過Node的標籤(Label)和Pod的nodeSelector屬性相匹配。
# kubectl label nodes <node-name> <label-key>=<label-value>
node節點創建對應的label后,可通過在定義Pod的時候加上nodeSelector的設置實現指定的調度。
示例:

  1 [root@uk8s-m-01 study]# kubectl label nodes 172.24.9.14 speed=io
  2 node/172.24.9.14 labeled
  3 [root@uk8s-m-01 study]# vi nginx-master-controller.yaml
  4 kind: ReplicationController
  5 metadata:
  6   name: nginx-master
  7   labels:
  8     name: nginx-master
  9 spec:
 10   replicas: 1
 11   selector:
 12     name: nginx-master
 13   template:
 14     metadata:
 15       labels:
 16         name: nginx-master
 17     spec:
 18       containers:
 19       - name: master
 20         image: nginx:1.7.9
 21         ports:
 22         - containerPort: 80
 23       nodeSelector:
 24         speed: io
 25 
 26 [root@uk8s-m-01 study]# kubectl create -f nginx-master-controller.yaml
 27 [root@uk8s-m-01 study]# kubectl get pods -o wide
 28 NAME                READY   STATUS    RESTARTS    AGE    IP            NODE
 29 nginx-master-7fjgj  1/1     Running   0           82s    172.24.9.71   172.24.9.14


提示:可以將集群中具有不同特點的Node貼上不同的標籤,實現在部署時就可以根據應用的需求設置NodeSelector來進行指定Node範圍的調度。

注意:若在定義Pod中指定了NodeSelector條件,但集群中不存在符合該標籤的Node,即使集群有其他可供使用的Node,Pod也無法被成功調度。

2.3 NodeAffinity親和性調度

親和性調度機制極大的擴展了Pod的調度能力,主要增強功能如下:

  1. 更具表達力,即更精細的力度控制;
  2. 可以使用軟限制、優先採用等限制方式,即調度器在無法滿足優先需求的情況下,會使用其他次條件進行滿足;
  3. 可以依據節點上正在運行的其他Pod的標籤來進行限制,而非節點本身的標籤,從而實現Pod之間的親和或互斥關係。

目前有兩種節點親和力表達:
requiredDuringSchedulingIgnoredDuringExecution:硬規則,必須滿足指定的規則,調度器才可以調度Pod至Node上(類似nodeSelector,語法不同)。
preferredDuringSchedulingIgnoredDuringExecution:軟規則,優先調度至滿足的Node的節點,但不強求,多個優先級規則還可以設置權重值。
IgnoredDuringExecution指:如果一個Pod所在的節點在Pod運行期間標籤發生了變化,不再符合該Pod的節點親和性需求,則系統將忽略Node上Label的變化,該Pod能繼續在該節點運行。
示例:
條件1:只運行在amd64的節點上;盡量運行在ssd節點上。

  1 [root@uk8s-m-01 study]# vi nodeaffinity-pod.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: with-node-affinity
  6 spec:
  7   affinity:
  8     nodeAffinity:
  9       requiredDuringSchedulingIgnoredDuringExecution:
 10         nodeSelectorTerms:
 11         - matchExpressions:
 12           - key: kubernetes.io/arch
 13             operator: In
 14             values:
 15             - amd64
 16       preferredDuringSchedulingIgnoredDuringExecution:
 17       - weight: 1
 18         preference:
 19           matchExpressions:
 20           - key: disk-type
 21             operator: In
 22             values:
 23             - ssd
 24   containers:
 25   - name: with-node-affinity
 26     image: gcr.azk8s.cn/google_containers/pause:2.0


NodeAffinity操作語法;In、NotIn、Exists、DoesNotExist、Gt、Lt。NotIn和DoesNotExist可以實現互斥功能。
NodeAffinity規則設置注意事項:

  • 若同時定義nodeSelector和nodeAffinity,則必須兩個條件都滿足,Pod才能最終運行指定在Node上;;
  • 若nodeAffinity指定多個nodeSelectorTerms,則只需要其中一個能夠匹配成功即可;
  • 若nodeSelectorTerms中有多個matchExpressions,則一個節點必須滿足所有matchExpressions才能運行該Pod。

2.4 PodAffinity親和性調度

PodAffinity根據節點上正在運行的Pod標籤而不是Node標籤來判斷和調度,要求對節點和Pod兩個條件進行匹配。
規則描述為:若在具有標籤X的Node上運行了一個或多個符合條件Y的Pod,則Pod應該(或者不應該)運行在這個Node上。
X通常為Node節點的機架、區域等概念,Pod是屬於某個命名空間,所以條件Y表達的是一個或全部命名空間中的一個Label Selector。
Pod親和性定義與PodSpec的affinity字段下的podAffinity字段里,互斥性定義於同一層次的podAntiAffinity子字段中。
舉例:

  1 [root@uk8s-m-01 study]# vi nginx-flag.yaml	#創建名為pod-flag,帶有兩個標籤的Pod
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: pod-affinity
  6 spec:
  7   affinity:
  8     podAffinity:
  9       requiredDuringSchedulingIgnoredDuringExecution:
 10       - labelSelector:
 11           matchExpressions:
 12           - key: security
 13             operator: In
 14             values:
 15             - S1
 16         topologyKey: kubernetes.io/hostname
 17   containers:
 18   - name: with-pod-affinity
 19     image: gcr.azk8s.cn/google_containers/pause:2.0

  1 [root@uk8s-m-01 study]# vi nginx-affinity-in.yaml	#創建定義標籤security=S1,對應如上Pod “Pod-flag”。
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: pod-affinity
  6 spec:
  7   affinity:
  8     podAffinity:
  9       requiredDuringSchedulingIgnoredDuringExecution:
 10       - labelSelector:
 11           matchExpressions:
 12           - key: security
 13             operator: In
 14             values:
 15             - S1
 16         topologyKey: kubernetes.io/hostname
 17   containers:
 18   - name: with-pod-affinity
 19     image: gcr.azk8s.cn/google_containers/pause:2.0
 20 
 21 [root@uk8s-m-01 study]# kubectl create -f nginx-affinity-in.yaml
 22 [root@uk8s-m-01 study]# kubectl get pods -o wide


提示:由上Pod親和力可知,兩個Pod處於同一個Node上。

  1 [root@uk8s-m-01 study]# vi nginx-affinity-out.yaml	#創建不能與參照目標Pod運行在同一個Node上的調度策略
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: anti-affinity
  6 spec:
  7   affinity:
  8     podAffinity:
  9       requiredDuringSchedulingIgnoredDuringExecution:
 10       - labelSelector:
 11           matchExpressions:
 12           - key: security
 13             operator: In
 14             values:
 15             - S1
 16         topologyKey: failure-domain.beta.kubernetes.io/zone
 17     podAntiAffinity:
 18       requiredDuringSchedulingIgnoredDuringExecution:
 19       - labelSelector:
 20           matchExpressions:
 21           - key: security
 22             operator: In
 23             values:
 24             - nginx
 25         topologyKey: kubernetes.io/hostname
 26   containers:
 27   - name: anti-affinity
 28     image: gcr.azk8s.cn/google_containers/pause:2.0
 29 
 30 [root@uk8s-m-01 study]# kubectl get pods -o wide	#驗證

2.5 Taints和Tolerations(污點和容忍)

Taint:使Node拒絕特定Pod運行;
Toleration:為Pod的屬性,表示Pod能容忍(運行)標註了Taint的Node。
Taint語法:$ kubectl taint node node1 key=value:NoSchedule
解釋:為node1加上一個Taint,該Taint的鍵為key,值為value,Taint的效果為NoSchedule。即除非特定聲明可以容忍此Taint,否則不會調度至node1上。
Toleration示例:

  1 tolerations:
  2 - key: "key"
  3   operator: "Equal"
  4   value: "value"
  5   effect: "NoSchedule"

  1 tolerations:
  2 - key: "key"
  3   operator: "Exists"
  4   effect: "NoSchedule"

注意:Pod的Toleration聲明中的key和effect需要與Taint的設置保持一致,並且滿足以下條件:

  • operator的值是Exists(無須指定value);
  • operator的值是Equal並且value相等;
  • 空的key配合Exists操作符能夠匹配所有的鍵和值;
  • 空的effect匹配所有的effect。


若不指定operator,則默認值為Equal。
taint說明:系統允許在同一個Node上設置多個taint,也可以在Pod上設置多個toleration。Kubernetes調度器處理多個taint和toleration的邏輯順序:首先列出節點中所有的taint,然後忽略pod的toleration能夠匹配的部分,剩下的沒有忽略掉的taint就是對pod的效果。以下是幾種特殊情況:
若剩餘的taint中存在effect=NoSchedule,則調度器不會把該Pod調度到這一節點上;
若剩餘的taint中沒有NoSchedule效果,但有PreferNoSchedule效果,則調度器會嘗試不把這個Pod指派到此節點;
若剩餘taint的效果有NoSchedule,並且這個Pod已經在該節點上運行,則會被驅逐,若沒有在該節點上運行,也不會再被調度到該節點上。
示例:

  1 $ kubectl taint node node1 key=value1:NoSchedule
  2 $ kubectl taint node node1 key=value1:NoExecute
  3 $ kubectl taint node node1 key=value2:NoSchedule
  4 tolerations:
  5 - key: "key1"
  6   operator: "Equal"
  7   value: "value"
  8   effect: "NoSchedule"
  9 tolerations:
 10 - key: "key1"
 11   operator: "Equal"
 12   value: "value1"
 13   effect: "NoExecute"


釋義:此Pod聲明了兩個容忍,且能匹配Node1的taint,但是由於沒有能匹配第三個taint的toleration,因此此Pod依舊不能調度至此Node。若該Pod已經在node1上運行了,那麼在運行時設置了第3個taint,它還能繼續在node1上運行,這是因為Pod可以容忍前兩個taint。
通常,若node加上effect=NoExecute的taint,那麼該Node上正在運行的所有無對應toleration的Pod都會被立刻驅逐,而具有相應toleration的Pod則永遠不會被驅逐。同時,系統可以給具有NoExecute效果的toleration加入一個可選的tolerationSeconds字段,表明Pod可以在taint添加到Node之後還能在此Node運行多久。

  1 tolerations:
  2 - key: "key1"
  3   operator: "Equal"
  4   value: "value"
  5   effect: "NoSchedule"
  6   tolerationSeconds: 3600

釋義:若Pod正在運行,所在節點被加入一個匹配的taint,則這個pod會持續在該節點運行3600s后被驅逐。若在此期限內,taint被移除,則不會觸發驅逐事件。
Taints和Tolerations常用場景:

  • 獨佔節點:

給特定的節點運行特定應用。
$ kubectl taint nodes 【nodename】 dedicated=groupName:NoSchedule
同時在Pod中設置對應的toleration配合,帶有合適toleration的Pod允許同時使用其他節點一樣使用有taint的節點。

  • 具有特殊硬件設備的節點

集群中部分特殊硬件(如安裝了GPU),則可以把不需要佔用GPU的Pod禁止在此Node上調度。

  1 $ kubectl taint nodes 【nodename】 special=true:NoSchedule
  2 $ kubectl taint nodes 【nodename】 special=true:PreferNoSchedule

  • 定義Pod驅逐行為

NoExecute的taint對節點上正在運行的Pod有以下影響:

    1. 沒有設置toleration的pod會被立刻驅逐;
    2. 配置了對應toleration的pod,若沒有為tolerationSeconds賦值,則會一直保留在此節點中;
    3. 配置了對應toleration的pod,且為tolerationSeconds賦值,則在指定時間后驅逐。

2.6 DaemonSet

DaemonSet是在每個Node上調度一個Pod的資源對象,用於管理集群中每個Node僅運行一份Pod的副本實例。
常見場景:
在每個Node上運行一個GlusterFS存儲的Daemon進程;
在每個Node上運行一個日誌採集程序,例如Fluentd;
在每個Node上運行一個性能監控程序,採集該Node的運行性能數據,例如Prometheus。
示例:

  1 [root@uk8s-m-01 study]# vi fluentd-ds.yaml
  2 apiVersion: extensions/v1beta1
  3 kind: DaemonSet
  4 metadata:
  5   name: fluentd-cloud-logging
  6   namespace: kube-system
  7   labels:
  8     k8s-app: fluentd-cloud-logging
  9 spec:
 10   template:
 11     metadata:
 12       namespace: kube-system
 13       labels:
 14         k8s-app: fluentd-cloud-logging
 15     spec:
 16       containers:
 17       - name: fluentd-cloud-logging
 18         image: gcr.azk8s.cn/google_containers/fluentd-elasticsearch:1.17
 19         resources:
 20           limits:
 21             cpu: 100m
 22             memory: 200Mi
 23         env:
 24         - name: FLUENTD_ARGS
 25           value: -q
 26         volumeMounts:
 27         - name: varlog
 28           mountPath: /var/log
 29           readOnly: false
 30         - name: containers
 31           mountPath: /var/lib/docker/containers
 32           readOnly: false
 33       volumes:
 34       - name: containers
 35         hostPath:
 36           path: /var/lib/docker/containers
 37       - name: varlog
 38         hostPath:
 39           path: /var/log

2.7 Job批處理調度

通過Kubernetes Job資源對象可以定義並啟動一個批處理任務,批處理任務通過并行(或者串行)啟動多個計算進程去處理一批工作項。根據批處理方式不同,批處理任務可以分為如下幾種模式:
Job Template Expansion模式:一個Job對象對應一個待處理的Work item,有幾個work item就產生幾個獨立的Job。通常適合Work item數量少、每個Work item要處理的數據量比較大的場景。
Queue with Pod Per Work Item模式:採用一個任務隊列存放Work item,一個Job對象作為消費者去完成這些Work item。此模式下,Job會啟動N個Pod,每個Pod都對應一個Work item。
Queue with Variable Pod Count模式:採用一個任務隊列存放Work item,一個Job對象作為消費者去完成這些Work item,但此模式下Job啟動的數量是可變的。
Kubernetes將Job氛圍以下三類:

  • Non-parallel Jobs

通常一個Job只啟動一個Pod,除非Pod異常,才會重啟該Pod,一旦此Pod正常結束,Job將結束。

  • Parallel Jobs with a fixed completion count

并行Job會啟動多個Pod,此時需要設定Job的.spec.completions參數為一個正數,當正常結束的Pod數量達至此參數設定的值后,Job結束。同時.spec.parallelism參數用來控制并行度,即同時啟動幾個Job來處理Work Item。

  • Parallel Jobs with a work queue

任務隊列方式的并行Job需要一個獨立的Queue,Work Item都在一個Queue中存放,不能設置Job的.spec.completions參數,此時Job具有以下特性:

    1. 每個Pod都能獨立判斷和決定是否還有任務項需要處理;
    2. 如果某個Pod正常結束,則Job不會再啟動新的Pod;
    3. 如果一個Pod成功結束,則此時應該不存在其他Pod還在工作的情況。它們應該都處於即將結束、退出的狀態;
    4. 如果所有Pod都結束了,且至少有一個Pod成功結束,則整個Jod成功結束。

2.8 Cronjob定時任務

表達式:Minutes Hours DayofMonth Month DayofWeek Year
Minutes:可出現”,”、”_”、”*”、”/”,有效範圍為0~59的整數;
Hours:出現”,”、”_”、”*”、”/”,有效範圍為0~23的整數;
DayofMonth:出現”,”、”_”、”*”、”/”、”L”、”W”、”C”,有效範圍為0~31的整數;
Month:可出現”,”、”_”、”*”、”/”,有效範圍為1~12的整數或JAN~DEC;
DayofWeek:出現”,”、”_”、”*”、”/”、”L”、”W”、”C”、”#”,有效範圍為1~7的整數或SUN~SAT;
*: 表示匹配該域的任意值, 假如在Minutes域使用“*”, 則表示每分鐘都會觸發事件。
/: 表示從起始時間開始觸發, 然後每隔固定時間觸發一次,例如在Minutes域設置為5/20, 則意味着第1次觸發在第5min時, 接下來每20min觸發一次, 將在第25min、 第45min等時刻分別觸發。
示例:*/1 * * * * #每隔1min執行一次任務

  1 [root@uk8s-m-01 study]# vi cron.yaml
  2 apiVersion: batch/v2alpha1
  3 kind: CronJob
  4 metadata:
  5   name: hello
  6 spec:
  7   schedule: "*/1 * * * *"
  8   jobTemplate:
  9     spec:
 10       template:
 11         spec:
 12           containers:
 13           - name: hello
 14             image: busybox
 15             args:
 16             - /bin/sh
 17             - -c
 18             - date; echo Hello from the Kubernetes cluster
 19           restartPolicy: OnFailure

  1 [root@master study]# kubectl create -f cron.yaml
  2 [root@master study]# kubectl get cronjob hello
  3 NAME    SCHEDULE      SUSPEND   ACTIVE   LAST SCHEDULE   AGE
  4 hello   */1 * * * *   False     0        <none>          29s
  5 [root@master study]# kubectl get pods
  6 NAME                     READY   STATUS      RESTARTS   AGE
  7 hello-1573378080-zvvm5   0/1     Completed   0          68s
  8 hello-1573378140-9pmwz   0/1     Completed   0          8s
  9 [root@node1 ~]# docker logs c7					#node節點查看日誌
 10 Sun Nov 10 09:31:13 UTC 2019
 11 Hello from the Kubernetes cluster
 12 [root@master study]# kubectl get jobs				#查看任務
 13 NAME               COMPLETIONS   DURATION   AGE
 14 hello-1573378500   1/1           8s         3m7s
 15 hello-1573378560   1/1           4s         2m7s
 16 hello-1573378620   1/1           6s         67s
 17 hello-1573378680   1/1           4s         7s
 18 [root@master study]# kubectl get pods -o wide | grep hello-1573378680	#以job任務查看對應的pod
 19 [root@master study]# kubectl delete cj hello			#刪除cronjob

2.9 初始化容器

在很多應用場景中, 應用在啟動之前都需要進行如下初始化操作。

  • 等待其他關聯組件正確運行( 例如數據庫或某個後台服務) 。
  • 基於環境變量或配置模板生成配置文件。
  • 從遠程數據庫獲取本地所需配置, 或者將自身註冊到某个中央數據庫中。
  • 下載相關依賴包, 或者對系統進行一些預配置操作。

示例:以Nginx應用為例, 在啟動Nginx之前, 通過初始化容器busybox為Nginx創建一個index.html主頁文件。同時init container和Nginx設置了一個共享的Volume, 以供Nginx訪問init container設置的index.html文件。

  1 [root@uk8s-m-01 study]# vi nginx-init-containers.yaml
  2 apiVersion: v1
  3 kind: Pod
  4 metadata:
  5   name: nginx
  6   annotations:
  7 spec:
  8   initContainers:
  9   - name: install
 10     image: busybox
 11     command:
 12     - wget
 13     - "-O"
 14     - "/work-dir/index.html"
 15     - http://kubernetes.io
 16     volumeMounts:
 17     - name: workdir
 18       mountPath: "/work-dir"
 19   containers:
 20   - name: nginx
 21     image: nginx:1.7.9
 22     ports:
 23     - containerPort: 80
 24     volumeMounts:
 25     - name: workdir
 26       mountPath: /usr/share/nginx/html
 27   dnsPolicy: Default
 28   volumes:
 29   - name: workdir
 30     emptyDir: {}

  1 [root@uk8s-m-01 study]# kubectl get pods
  2 NAME    READY   STATUS     RESTARTS   AGE
  3 nginx   0/1     Init:0/1   0          2s
  4 [root@uk8s-m-01 study]# kubectl get pods
  5 NAME    READY   STATUS    RESTARTS   AGE
  6 nginx   1/1     Running   0          13s
  7 [root@uk8s-m-01 study]# kubectl describe pod nginx		#查看事件可知會先創建init容器,名為install


init容器與應用容器的區別如下。
(1) init container的運行方式與應用容器不同, 它們必須先於應用容器執行完成, 當設置了多個init container時, 將按順序逐個運行, 並且只有前一個init container運行成功后才能運行后一個init container。 當所有init container都成功運行后, Kubernetes才會初始化Pod的各種信息, 並開始創建和運行應用容器。
(2) 在init container的定義中也可以設置資源限制、 Volume的使用和安全策略, 等等。 但資源限制的設置與應用容器略有不同。

  • 如果多個init container都定義了資源請求/資源限制, 則取最大的值作為所有init container的資源請求值/資源限制值。
  • Pod的有效(effective) 資源請求值/資源限制值取以下二者中的較大值。
    • 所有應用容器的資源請求值/資源限制值之和。
    • init container的有效資源請求值/資源限制值。
  • 調度算法將基於Pod的有效資源請求值/資源限制值進行計算,即init container可以為初始化操作預留系統資源, 即使後續應用容器無須使用這些資源。
  • Pod的有效QoS等級適用於init container和應用容器。
  • 資源配額和限制將根據Pod的有效資源請求值/資源限制值計算生效。
  • Pod級別的cgroup將基於Pod的有效資源請求/限制, 與調度機制

一致。
(3) init container不能設置readinessProbe探針, 因為必須在它們成功運行后才能繼續運行在Pod中定義的普通容器。在Pod重新啟動時, init container將會重新運行, 常見的Pod重啟場景如下。

  • init container的鏡像被更新時, init container將會重新運行, 導致Pod重啟。 僅更新應用容器的鏡像只會使得應用容器被重啟。
  • Pod的infrastructure容器更新時, Pod將會重啟。
  • 若Pod中的所有應用容器都終止了, 並且RestartPolicy=Always, 則Pod會重啟。

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

【其他文章推薦】

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

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

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

小三通海運與一般國際貿易有何不同?

小三通快遞通關作業有哪些?

從最近面試聊聊我所感受的職業天花板

## 特別特別嚴肅的申明 (正經的)

未免引起誤解,標題已修改。

我一開始寫這篇文章,也純粹是有感而發。實在沒想到會引起如此多的關注。甚至還被社區大佬翻牌。說實話,誠惶誠恐。

再次申明一遍,我寫的也僅僅只是我個人的感受,我就是萬萬千千的普通碼農中的一個,所寫文章也僅僅是從自我角度出發。

不具備任何普適性參考性。請大家看清標題,只是個人感受,不適用.net整個行業。尤其請大家不要拿我的經歷作為語言選擇的參考。

。net也好,java也好,每個語言都有自己的生態,選擇語言之前請先確認自己的能力,請不要抱怨語言害了你,請先審視自己是否一直在努力進步,還是只是在隨大流。

博主就是一個普通碼農,太高端的層面很難觸及,找工作的途徑也只有朋友內推和招聘網站。

那麼我在這兩種途徑下所接觸到的信息,就是我目前的薪資已經到了這兩種渠道下的頂薪。所以,我自我感覺是我現在觸到我的天花板。

那麼我向上突破的方法,要麼向社區大佬學習,努力突破到那個圈子,但是對於這個自己實在沒底,大佬的圈子太高,感覺自己能力不夠,對自己實在沒信心。

第二種就是多語言發展,修鍊技術內功,管理內功,我覺着在努努力踮踮腳還是能夠到30K、40K月薪的小尾巴。

最後奉勸各位園友,語言從來都不是決定自己前途的關鍵因素,更多的還是要修鍊自己的內功不斷向上突破。

博主寫這篇文章的初衷只是想表述自己目前所面臨的職業困境,實在沒想到會有如此多的關注,用詞上的不當,給大家造成困擾,萬分抱歉。

 

#0 前言

入職新公司沒多久,閑來無事在博客園閑逛,看到園友分享的面試經歷,正好自己這段時間面試找工作,也挺多感想的,乾脆趁這個機會總結整理一下。

博主13年開始實習,14年畢業。到現在也工作五六年了。今年面試最大的感受就是觸及了.net的天花板。坐標,杭州。

 #1 背景

今年九月份從一家創業公司離職,原因么自然是公司創業失敗倒閉。

當初以技術合伙人的身份進入,雄心勃勃,然後挨了一頓社會毒打,從此老實做人,面朝黃土背朝天,老老實實去搬磚。

九月份出來,已經是中旬,開始刷新簡歷,準備穩坐釣魚台,等着電話信息轟炸。然後,等了两天,等了一首涼涼。直到這個時候博主才意識到,形式不對。

我的思維還停留在兩三年前,工 作遍地,只要更新下簡歷就會有無數的面試邀請。同志門,情況變了呀,行業寒冬真不只是說說而已。

沒辦法,只好花錢,刷新下簡歷,瀏覽崗位,主動出擊。中間接到了好幾個獵頭電話, 但特么都是java。好想吐槽一下,簡歷上.net辣么大的字,你們真的不識字么,21世紀了啊喂。

#2 某建築類軟件公司

主營業務:建築軟件,公司已上市。

技術框架:.net平台,具體的不是特別了解

招聘崗位:.net高級開發工程師

面試:一共四輪面試。

第一輪:就是HR了,簡單聊了下情況,為什麼離職,之前薪資多少,期望薪資多少。

第二輪:他們某業務線的部門經理和技術主管共同面試。

基本面試情況就是我在說他們在聽,我主要講解了項目的設計方案,使用的技術,遇到的困難,最終的解決方案。

技術面試官就問了兩個問題,一是從.net升級到netcore中間碰到過哪些問題。

第二個基於rabbitmq的分佈式事務是怎麼做的。

然後他們部門經理問了些團隊管理的問題。如何做團隊成員的任務分配,有團隊成員向你提出離職或者漲薪你怎麼處理,團隊的代碼質量如果管控

第三輪:他們的CTO,然後開始又是自我介紹。

只好把之前的又重複一遍,巴拉巴拉。最後就問了一個分佈式事務的解決方案有那些,平時是怎麼使用的。

最後聊了一下我的定位,就是進去是負責他們的平台架構,包括一些公用業務的架構封裝,老架構的netcore升級

第四輪:最後是他們的公司董事長,上來又是先自我介紹。然後問了下職業規劃。

接着就是拿着我的簡歷說這個工作跳動比較頻繁,尤其是從上一家比較大的公司跳槽到一個創業公司是基於一個什麼樣的考慮呢,感覺個人穩定性和職業性規劃都不夠。

博主當時內心的os是黑人問號臉??????我能是基於什麼樣的考慮,我為了世界和平好不好。

然後被大佬教育了一頓,灌輸了一些個人和公司共同體,什麼共贏發展什麼共同成長的理念。

結果:通過,HR小姐姐來談薪資。

只能給到20K,然後還是18k基本工資+2K的級別補貼,說是我進去之後定的級別是T3,

然後每年三四月份和九十月份可以申請調薪調級,強制要求995?????? 我特么跳槽不漲薪就算了你還給我降薪,還995,PASS。

#3 某醫美集團下轄子公司

主營業務:醫美行業的sass軟件

技術框架:GRPC

面試:一輪,技術主管。

招聘崗位:.net架構師

主要問題:依賴注入的生命周期,在框架設計中的應用場景有那些。

在技術選型時主要考慮的因素。

在框架設計時會應用到那些設計模式,主要應用場景是什麼。

對於netcore中間件的理解。

應對系統高併發的解決方案。

聊一聊對微服務的理解,基於netcore的微服務架構是怎麼設計的。

面試結果:通過。但薪資只有20K,哎呦喂,你都對不起你招聘崗位的名字呀。

#4 某物業管理軟件公司

主營業務:做小區物業管理軟件,公司兩百多人。

技術框架:.net mvc 三層

招聘崗位:.net副總監

面試:一輪。總監面試,但是木有問任何技術問題,也木有問任何團隊管理問題。逮者我之前的離職原因各種問。

面試結果:未通過。一臉懵逼的出來,都不知道為啥沒通過。老子也是信了你的邪。

#5 某電商初創企業。

主營業務:拍賣類的電商平台。公司是初創,技術團隊都沒組建完整。

面試:兩輪。

第一輪是他們的一個技術負責人,只是看看了簡歷,然後問了一個讓我哭笑不得問題,就是如果你進入公司,發現周圍人技術都比較菜的時候,你是不是會看不起別人。 笑哭!!!

第二輪是老闆,老闆就是主要負責畫大餅,聊前景,聊機遇。

結果:通過。工資待遇給到稅前24K。

但是我了解到老闆之前做互金,然後平台清盤。具體情況不清楚,大佬,惹不起,躲了躲了。

在這裏一定奉勸各位園友,互金平台或者老闆有互金背景的千萬小心。

我身邊已經不少朋友,被坑到,即使現在沒事,也說不定什麼時候就會被警察找上門。

就有朋友,剛入職公司沒多久,而且公司業務也不是做互金的,結果沒幾天,警察上門,老闆帶走就因為老闆之前做互金,還是出事兒了。

#6 某社交類公司

主營業務:付費社交app,主打東南亞市場

技術框架:.net 三層

招聘崗位:.net高級開發工程師

面試:三輪。

第一輪:部門的CTO面試,互相聊得挺愉快。

主要問了之前的項目微服務怎麼做的,服務拆分的粒度怎麼規劃,整個服務的架構怎麼規劃用到哪些技術。

然後問了數據庫方面的分庫分表怎麼做的,用的什麼中間件,分庫分表後主鍵id如何生成。

應對高併發架構上是怎麼處理的。如何保證redis的高併發高可用。面對緩存穿透、雪崩、擊穿怎麼解決的。

消息隊列的高可用、消息的冪等性,面對消息積壓如何處理。

接着就是聊團隊管理,還是人員管理,任務分配,質量保證這些問題。

接手一個新團隊后如何摸清各成員能力,不同能力的人工作上應該怎麼安排。

還有一個,就是你作為團隊主管你的工作時間是碎片化的,但同時你作為技術leader又要把控技術方案,而做技術是需要時間的連續性,你如何協調這兩者之間的衝突。

挺有意思,只有技術管理一肩挑的團隊才會遇到這種問題了。

最後介紹了一下團隊目前的組織架構,技術方向。嗯,要做.net升級,要做微服務。嗯,最後要轉java。誒,是不是有什麼奇怪的東西,.netcore它不香么。

第二輪:人事面試。嗯,就是問問離職原因,然後介紹了下公司業務發展,前景規劃,入職后的主要工作職能,然後談了下期望薪資。

第三輪:boos面。老闆,沒問什麼問題,就是聊了聊職業規劃,然後么他介紹公司發展方向,前景規劃,我作為一個負責任的捧哏, 當然舔着嘍。

面試結果:通過。薪資談到稅前24K。但五險一金都是最低標準繳納。年終獎說是0到12個月,看績效。

#7 某汽車製造公司的外包崗

面試:外包公司有個技術經理做了一個簡單電話面試。然後就約着到甲方的公司進行面試。面試兩輪,是甲方的兩個平台架構師。問題都大同小異,不贅述了。

面試結果:通過。但博主內心相當糾結,因為對於外包,網上實在是沒有好的評價,但是和兩個面試官聊得蠻愉快。

當初去面試了,也純粹是因為好奇,反正當時面試邀請也少,閑着也是閑着么。

薪資談到23k,對方說還是走了一個特別申請,甲方那邊兒再高給不了。五險一金都是最低標準。

但是HR說這個崗位是甲方為了儲備人才招聘的,我當天面試過後,甲方就把這個崗位招聘關了,只招我一個,等到明年三四月份內部編製出來,我是妥妥轉到甲方。

而且進去之後的工作也是和面試我的那個架構師一起工作,負責他們平台架構規劃。

一開始去面試之前我都說了工資要求和最低標準,滿口說沒問題,結果面試完了就又不行了。你個糟老頭子,壞的很,我信你個鬼。

#8 寫在最後

  中間也還有面試有其他幾家公司,套路問題都差不多,就不在寫出來了。找工作一共花費兩周時間,面試了也有八九家,但真正能給到期望工資的就那麼兩三家。這之間自己在網上主動投遞過,但基本都沒有回信。兩周過去,在回過頭來看,卻發現網上再找不到其他合適的崗位了,不是已經面試過,就是投遞了沒反應。到最後發現,我能選擇的就只有那麼幾家公司。而且,最嚴重的一個感受就是,我翻遍了所有的招聘網站,我目前所要的工資,已經是.net行業的天花板,往上沒有空間了。.net高級開發也好、.net架構師也好、技術經理也罷,能給到工資25K就已經是到頂了,而且崗位特別少。然後做cs方向的,價格開的比bs方向的還能高一些,頂薪能到三萬。做服務的.net被java搶佔了太多市場,即便有很多公司,初期是用.net做的,即便現在netcore已經跨平台,但公司做微服務還是要轉java,我真的好想問一句netcore它不香么,vs它不香么,都咋想的。

#9 尾篇

  最後的最後。整理一下博主在做netcore微服務所用到的相關技術,做個整體的總結。後續會一點一點具體介紹,希望能形成一個系列,希望最後能堅持寫完。

  服務註冊/發現:consul或zookeeper,各有優劣,個人傾向consul

  分佈式通訊:restful api形式或rpc。

  分佈式事件總線:推薦使用cap。cap同時支持 RabbitMQ,Kafka,Azure Service Bus 等進行底層之間的消息發送,同時內置了TCC實現。

       網關、熔斷、降級、限流:ocelot網關,應該是當下netcore平台下最火熱的網關開源項目了。同時集成了polly來滿足熔斷、降級、限流的功能要求。

  配置中心:攜程的開源項目Apollo。博主之前是為了業務需求自己寫的,不具通用性。

  微服務監控:分佈式調用鏈跟蹤zipkin和skywalking,同時還可監控服務性能。推薦使用skywalking,對代碼無侵入。

        日誌監控ELK,這個不需要多介紹了,文章太多了。

  持續集成自動部署:GitLab+Jenkins+k8s

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

【其他文章推薦】

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

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

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

小三通海運與一般國際貿易有何不同?

小三通快遞通關作業有哪些?

Java開發者學習技術體系

01基礎技術體系

我認為知識技能體系化是判斷技術是否過關的第一步。知識體系化包含兩層含義:

1、 能夠知道技術知識圖譜(高清版圖譜掃文末二維碼)的內容

比如分佈式系統中常用的RPC技術,其背後就涉及到網絡IO(Netty)、網絡協議、服務發現(Zookeeper配置中心)、RPC服務治理(限流、熔斷、降級)、負載均衡等。

2、 能夠理清各類技術概念之間的區別和聯繫

在分佈式系統領域中,有很多相似的概念,但又分佈在不同的產品或層級中。比如負載均衡這個詞,DNS、LVS、Ngnix、F5等產品都能實現,而且在大型分佈式系統中他們會同時存在,那麼就要搞清楚他們各自的位於什麼層級,解決了什麼問題。

再比如緩存這項技術,有分佈式緩存、本地緩存、數據庫緩存,在往下還有硬件層級的緩存。同樣都是緩存,他們之間的區別又是什麼?

如果你仔細去觀察,大廠的後端開發工程師總是能對整個技術體系了如指掌,從而在系統設計與技術選型階段就能夠做出較為合理的架構。 

02實踐經驗的積累

       能否快速解決實戰中的業務問題是判斷技術是否過關的第二步。
       大家在面試的過程中,都會有一種體會:我的知識體系已經建立了,但在回答面試官問題的時候,總感覺像在背答案,而且也沒有辦法針對性的回答面試官問題。比如在面試官問到這些問題時:

  1. 我們知道消息隊列可應用於耦系統,應對異步消費等場景,那如何在網絡不可靠的場景下保證業務數據處理的正確性?
  2. 我們都知道在分佈式系統會用到緩存,那該如何設置緩存失效機制才能避免系統出現緩存雪崩?
  3. 我們都或多或少的知道系統發布上線的流程,但在大流量場景下採用何種發布機制才能盡可能的做到平滑?

能完善的解決這些問題是區分一個程序員是否有經驗的重要標誌,知識的體系化是可以從書本不斷的凝練來獲得,但經驗的積累需要通過實戰的不斷總結

對很多人來說很為難的一點是,平時寫着的業務代碼,很少有機會接觸到大廠的優秀實踐,那麼這時候更需要從如下兩個角度逼問:

1、當流量規模再提高几個量級,那麼我的系統會出現什麼問題?

2、假如其中一個環節出現了問題,那麼該怎麼保證系統的穩定性?

03技術的原理

上面的提到都是將技術用於業務實踐,以及高效的解決業務中出現的問題。但這是否就意味着自己的技術已經過關了呢?我認為還不能。

判斷技術是否過關的第三步是能否洞察技術背後的設計思想和原理。

如果你參加過一些大廠面試,還會問到一些開放性的問題:

1、 寫一段程序,讓其運行時的表現為觸發了5次Young GC、3次Full GC、然後3次Young GC;

2、 如果一個Java進程突然消失了,你會怎麼去排查這種問題?

3、 給了一段Spring加載Bean的代碼片段,闡述一下具體的執行流程?

       是不是看上去很難,是不是和自己準備的“題庫”中的問題不一樣?不知道從何處下手?如果你有這種感覺,那麼說明你的技術還需要繼續修鍊。

       你要明白的是這種開放性的問題,提問的角度千變萬化,但最終落腳點卻都是基本原理。如果你不了解GC的觸發條件,你就肯定無法答出第一題;同樣,如果你對Spring啟動機制了解的很清楚,那麼無論他給出的是什麼樣的代碼,你都能回答出代碼經歷的過程。如果你能以不變應萬變,那麼恭喜你,你的技術過關了。

       上面提到了很多技術問題,這裏我不做詳細的解釋,都能在下面的技術圖譜中找到答案:

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

【其他文章推薦】

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

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

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

新能源積分交易制度 或替代財政補貼

新能源汽車業發展日益成熟之時,單純的財政補貼帶來的車企騙補等諸多負面效應逐漸顯現。對此,有專家提出建議採用積分交易機制替代財政補貼政策,促進車企在傳統燃油車節能和新能源車技術進步兩方面共同發展。  
  中國汽車技術研究中心新能源汽車積分政策負責人時間表示,相比真金白銀的財政補貼,新能源積分交易制度靈活性更高,是當下國家推動新能源產業發展的一種可行方式。   新能源積分交易制度,即政府將企業年度“零排放”車型的銷售情況記錄成積分,以積分為依據來考核企業在節能減排方面是否達標。若企業積分不達標,可以購買同行業其餘公司的積分,或者向政府繳納高額罰款。   中國若要實施積分交易制度 政府部門職責需合理分配。   為此,積分政策負責人時間給出幾點建議:作為政策的制定者政府,需要部門間進行合理的職責分配,物質保障方面,中國新能源基礎設施建設尚需進一步完善。意識形態方面,當前,消費者對新能源車的認識還不充足,對制度實施造成一定阻礙。企業方面,不同規模的企業須在政策上區別對待,為中小企業的發展提供空間;當企業規模有所變更時,應當提供相應的扶持政策。   文章來源:人民網

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

【其他文章推薦】

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

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

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

星際爭霸2 AI開發(持續更新)

準備

我的環境是python3.6,sc2包0.11.1
機器學習包下載鏈接:
地圖下載鏈接
pysc2是DeepMind開發的星際爭霸Ⅱ學習環境。 它是封裝星際爭霸Ⅱ機器學習API,同時也提供Python增強學習環境。
以神族為例編寫代碼,神族建築科技圖如下:

採礦

# -*- encoding: utf-8 -*-
'''
@File    :   __init__.py.py    
@Modify Time      @Author       @Desciption
------------      -------       -----------
2019/11/3 12:32   Jonas           None
'''

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer


class SentdeBot(sc2.BotAI):
    async def on_step(self, iteration: int):
        await self.distribute_workers()


run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)
],realtime = True)

注意
game_data.py的assert self.id != 0註釋掉
pixel_map.py的assert self.bits_per_pixel % 8 == 0, "Unsupported pixel density"註釋掉
否則會報錯

運行結果如下,农民開始採礦

可以正常採礦

建造农民和水晶塔

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *


class SentdeBot(sc2.BotAI):
    async def on_step(self, iteration: int):
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()

    # 建造农民
    async def build_workers(self):
        # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
        for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
            # 是否有50晶體礦
            if self.can_afford(UnitTypeId.PROBE):
                await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且水晶不是正在建造
        if self.supply_left<5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON,near=nexuses.first)

## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)
],realtime = True)

運行結果如下,基地造农民,农民造水晶

收集氣體和開礦

代碼如下

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *


class SentdeBot(sc2.BotAI):
    async def on_step(self, iteration: int):
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()

    # 建造农民
    async def build_workers(self):
        # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
        for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
            # 是否有50晶體礦
            if self.can_afford(UnitTypeId.PROBE):
                await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))

    ## 開礦
    async def expand(self):
        if self.units(UnitTypeId.NEXUS).amount<3 and self.can_afford(UnitTypeId.NEXUS):
            await self.expand_now()

## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)
], realtime=False)

run_game的realtime設置成False,可以在加速模式下運行遊戲。
運行效果如下:

可以建造吸收廠和開礦

建造軍隊

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *


class SentdeBot(sc2.BotAI):
    async def on_step(self, iteration: int):
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()

    # 建造农民
    async def build_workers(self):
        # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
        for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
            # 是否有50晶體礦
            if self.can_afford(UnitTypeId.PROBE):
                await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))

    ## 開礦
    async def expand(self):
        if self.units(UnitTypeId.NEXUS).amount<2 and self.can_afford(UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            if self.units(UnitTypeId.PYLON).ready.exists:
                # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
                if self.units(UnitTypeId.GATEWAY).ready.exists:
                    if not self.units(UnitTypeId.CYBERNETICSCORE):
                        if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                            await self.build(UnitTypeId.CYBERNETICSCORE,near = pylon)
                # 否則建造折躍門
                else:
                    if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                        await self.build(UnitTypeId.GATEWAY,near=pylon)

    # 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
            if self.can_afford(UnitTypeId.STALKER) and self.supply_left>0:
                await self.do(gw.train(UnitTypeId.STALKER))



## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Easy)
], realtime=False)

運行結果如下:

可以看到,我們建造了折躍門和控制核心並訓練了追獵者

控制部隊進攻

代碼如下


import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *
import random

class SentdeBot(sc2.BotAI):
    async def on_step(self, iteration: int):
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.attack()

    # 建造农民
    async def build_workers(self):
        # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
        for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
            # 是否有50晶體礦
            if self.can_afford(UnitTypeId.PROBE):
                await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))

    ## 開礦
    async def expand(self):
        if self.units(UnitTypeId.NEXUS).amount<3 and self.can_afford(UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
            if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):
                if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                    await self.build(UnitTypeId.CYBERNETICSCORE,near = pylon)
            # 否則建造折躍門
            elif len(self.units(UnitTypeId.GATEWAY))<=3:
                if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                    await self.build(UnitTypeId.GATEWAY,near=pylon)

    ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
            if self.can_afford(UnitTypeId.STALKER) and self.supply_left>0:
                await self.do(gw.train(UnitTypeId.STALKER))

    ## 尋找目標
    def find_target(self,state):
        if len(self.known_enemy_units)>0:
            # 隨機選取敵方單位
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_units)>0:
            # 隨機選取敵方建築
            return random.choice(self.known_enemy_structures)
        else:
            # 返回敵方出生點位
            return self.enemy_start_locations[0]

    ## 進攻
    async def attack(self):
        # 追獵者數量超過15個開始進攻
        if self.units(UnitTypeId.STALKER).amount>15:
            for s in self.units(UnitTypeId.STALKER).idle:
                await self.do(s.attack(self.find_target(self.state)))

        # 防衛模式:視野範圍內存在敵人,開始攻擊
        if self.units(UnitTypeId.STALKER).amount>5:
            if len(self.known_enemy_units)>0:
                for s in self.units(UnitTypeId.STALKER).idle:
                    await self.do(s.attack(random.choice(self.known_enemy_units)))

## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Medium)
], realtime=False)

運行結果如下

可以看到,4個折躍門訓練追獵者並發動進攻。

擊敗困難電腦

我們目前的代碼只能擊敗中等和簡單電腦,那麼如何擊敗困難電腦呢?
代碼如下


import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *
import random


class SentdeBot(sc2.BotAI):
    def __init__(self):
        # 經過計算,每分鐘大約165迭代次數
        self.ITERATIONS_PER_MINUTE = 165
        # 最大农民數量
        self.MAX_WORKERS = 65

    async def on_step(self, iteration: int):
        self.iteration = iteration
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.attack()

    # 建造农民
    async def build_workers(self):
        # 星靈樞鈕*16(一個基地配備16個农民)大於農民數量並且現有农民數量小於MAX_WORKERS
        if len(self.units(UnitTypeId.NEXUS))*16>len(self.units(UnitTypeId.PROBE)) and len(self.units(UnitTypeId.PROBE))<self.MAX_WORKERS:
                # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
                for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
                    # 是否有50晶體礦建造农民
                    if self.can_afford(UnitTypeId.PROBE):
                        await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0,nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0,vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR,vaspene))

    ## 開礦
    async def expand(self):
        # (self.iteration / self.ITERATIONS_PER_MINUTE)是一個緩慢遞增的值,動態開礦
        if self.units(UnitTypeId.NEXUS).amount<self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
            if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):
                if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                    await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)
            # 否則建造折躍門
            # (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一個緩慢遞增的值
            elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
                if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                    await self.build(UnitTypeId.GATEWAY, near=pylon)
            # 控制核心存在的情況下建造星門
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
                    if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):
                        await self.build(UnitTypeId.STARGATE, near=pylon)

    ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
            if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:

                if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:
                    await self.do(gw.train(UnitTypeId.STALKER))

        for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:
            if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(UnitTypeId.VOIDRAY))

    ## 尋找目標
    def find_target(self,state):
        if len(self.known_enemy_units)>0:
            # 隨機選取敵方單位
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_units)>0:
            # 隨機選取敵方建築
            return random.choice(self.known_enemy_structures)
        else:
            # 返回敵方出生點位
            return self.enemy_start_locations[0]

    ## 進攻
    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {UnitTypeId.STALKER: [15, 5],
                            UnitTypeId.VOIDRAY: [8, 3]}

        for UNIT in aggressive_units:
            # 攻擊模式
            if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][
                1]:
                for s in self.units(UNIT).idle:
                    await self.do(s.attack(self.find_target(self.state)))
            # 防衛模式
            elif self.units(UNIT).amount > aggressive_units[UNIT][1]:
                if len(self.known_enemy_units) > 0:
                    for s in self.units(UNIT).idle:
                        await self.do(s.attack(random.choice(self.known_enemy_units)))
## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)
], realtime=False)

運行結果如下

可以看到,擊敗了困難人族電腦,但是電腦選擇了rush戰術,我們寫得AI腳本會輸掉遊戲。顯然,這不是最佳方案。
“只有AI才能拯救我的勝率”,請看下文。

採集地圖數據

這次我們只造一個折躍門,全力通過星門造虛空光輝艦
修改offensive_force_buildings(self)方法的判斷

elif len(self.units(GATEWAY)) < 1:
                if self.can_afford(GATEWAY) and not self.already_pending(GATEWAY):
                    await self.build(GATEWAY, near=pylon)

註釋或者刪除build_offensive_force(self)的建造追獵者的代碼

        ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        # for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
        #     if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:
        #
        #         if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:
        #             await self.do(gw.train(UnitTypeId.STALKER))

        for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:
            if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(UnitTypeId.VOIDRAY))

attack(self)中的aggressive_units註釋掉Stalker
導入numpy和cv2庫

game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)

建立以地圖Heigt為行,Width為列的三維矩陣

for nexus in self.units(NEXUS):
            nex_pos = nexus.position
            print(nex_pos)
            cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)  # BGR

遍歷星靈樞紐,獲取下一個位置,畫圓,circle(承載圓的img, 圓心, 半徑, 顏色, thickness=-1表示填充)
接下來我們要垂直翻轉三維矩陣,因為我們建立的矩陣左上角是原點(0,0),縱坐標向下延申,橫坐標向右延申。翻轉之後就成了正常的坐標系。

flipped = cv2.flip(game_data, 0)

圖像縮放,達到可視化最佳。

        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)
        cv2.imshow('Intel', resized)
        cv2.waitKey(1)

至此,完整代碼如下

import sc2
from sc2 import run_game, maps, Race, Difficulty
from sc2.player import Bot, Computer
from sc2.constants import *
import random
import numpy as np
import cv2


class SentdeBot(sc2.BotAI):
    def __init__(self):
        # 經過計算,每分鐘大約165迭代次數
        self.ITERATIONS_PER_MINUTE = 165
        # 最大农民數量
        self.MAX_WORKERS = 65

    async def on_step(self, iteration: int):
        self.iteration = iteration
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.intel()
        await self.attack()

    async def intel(self):
        # 根據地圖建立的三維矩陣
        game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)
        for nexus in self.units(UnitTypeId.NEXUS):
            nex_pos = nexus.position
            # circle(承載圓的img, 圓心, 半徑, 顏色, thickness=-1表示填充)
            # 記錄星靈樞紐的位置
            cv2.circle(game_data, (int(nex_pos[0]), int(nex_pos[1])), 10, (0, 255, 0), -1)
        # 圖像翻轉垂直鏡像
        flipped = cv2.flip(game_data, 0)
        # 圖像縮放
        # cv2.resize(原圖像,輸出圖像的大小,width方向的縮放比例,height方向縮放的比例)
        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)
        cv2.imshow('Intel', resized)

        # cv2.waitKey(每Xms刷新圖像)
        cv2.waitKey(1)

    # 建造农民
    async def build_workers(self):
        # 星靈樞鈕*16(一個基地配備16個农民)大於農民數量並且現有农民數量小於MAX_WORKERS
        if len(self.units(UnitTypeId.NEXUS)) * 16 > len(self.units(UnitTypeId.PROBE)) and len(
                self.units(UnitTypeId.PROBE)) < self.MAX_WORKERS:
            # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
            for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
                # 是否有50晶體礦建造农民
                if self.can_afford(UnitTypeId.PROBE):
                    await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0, vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR, vaspene))

    ## 開礦
    async def expand(self):
        # (self.iteration / self.ITERATIONS_PER_MINUTE)是一個緩慢遞增的值,動態開礦
        if self.units(UnitTypeId.NEXUS).amount < self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(
                UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
            if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):
                if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                    await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)
            # 否則建造折躍門
            # (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一個緩慢遞增的值
            # elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
            elif len(self.units(UnitTypeId.GATEWAY)) < 1:
                if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                    await self.build(UnitTypeId.GATEWAY, near=pylon)
            # 控制核心存在的情況下建造星門
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
                    if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):
                        await self.build(UnitTypeId.STARGATE, near=pylon)

    ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:
            if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(UnitTypeId.VOIDRAY))

    ## 尋找目標
    def find_target(self, state):
        if len(self.known_enemy_units) > 0:
            # 隨機選取敵方單位
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_units) > 0:
            # 隨機選取敵方建築
            return random.choice(self.known_enemy_structures)
        else:
            # 返回敵方出生點位
            return self.enemy_start_locations[0]

    ## 進攻
    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {UnitTypeId.VOIDRAY: [8, 3]}

        for UNIT in aggressive_units:
            # 攻擊模式
            if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][1]:
                for s in self.units(UNIT).idle:
                    await self.do(s.attack(self.find_target(self.state)))
            # 防衛模式
            elif self.units(UNIT).amount > aggressive_units[UNIT][1]:
                if len(self.known_enemy_units) > 0:
                    for s in self.units(UNIT).idle:
                        await self.do(s.attack(random.choice(self.known_enemy_units)))


## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)
], realtime=False)

運行結果如下

採集到了地圖位置。

偵察

在intel(self)里創建一個字典draw_dict,UnitTypeId作為key,半徑和顏色是value


        draw_dict = {
            UnitTypeId.NEXUS: [15, (0, 255, 0)],
            UnitTypeId.PYLON: [3, (20, 235, 0)],
            UnitTypeId.PROBE: [1, (55, 200, 0)],
            UnitTypeId.ASSIMILATOR: [2, (55, 200, 0)],
            UnitTypeId.GATEWAY: [3, (200, 100, 0)],
            UnitTypeId.CYBERNETICSCORE: [3, (150, 150, 0)],
            UnitTypeId.STARGATE: [5, (255, 0, 0)],
            UnitTypeId.ROBOTICSFACILITY: [5, (215, 155, 0)],

            UnitTypeId.VOIDRAY: [3, (255, 100, 0)]
        }

迭代同上

for unit_type in draw_dict:
            for unit in self.units(unit_type).ready:
                pos = unit.position
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1)

存儲三族的主基地名稱(星靈樞紐,指揮中心,孵化場),刻畫敵方建築。

# 主基地名稱
        main_base_names = ["nexus", "supplydepot", "hatchery"]
        # 記錄敵方基地位置
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            if enemy_building.name.lower() not in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 5, (200, 50, 212), -1)
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            if enemy_building.name.lower() in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 15, (0, 0, 255), -1)

刻畫敵方單位,如果是农民畫得小些,其他單位則畫大些。

        for enemy_unit in self.known_enemy_units:

            if not enemy_unit.is_structure:
                worker_names = ["probe", "scv", "drone"]
                # if that unit is a PROBE, SCV, or DRONE... it's a worker
                pos = enemy_unit.position
                if enemy_unit.name.lower() in worker_names:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (55, 0, 155), -1)
                else:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (50, 0, 215), -1)

在offensive_force_buildings(self)方法中添加建造机械台

            if self.units(CYBERNETICSCORE).ready.exists:
                if len(self.units(ROBOTICSFACILITY)) < 1:
                    if self.can_afford(ROBOTICSFACILITY) and not self.already_pending(ROBOTICSFACILITY):
                        await self.build(ROBOTICSFACILITY, near=pylon)

創建scout(),訓練Observer

async def scout(self):
        if len(self.units(OBSERVER)) > 0:
            scout = self.units(OBSERVER)[0]
            if scout.is_idle:
                enemy_location = self.enemy_start_locations[0]
                move_to = self.random_location_variance(enemy_location)
                print(move_to)
                await self.do(scout.move(move_to))

        else:
            for rf in self.units(ROBOTICSFACILITY).ready.noqueue:
                if self.can_afford(OBSERVER) and self.supply_left > 0:
                    await self.do(rf.train(OBSERVER))

生成隨機位置,很簡單。意思是橫坐標累計遞增-0.2和0.2倍的橫坐標,限制條件為如果x超過橫坐標,那麼就是橫坐標最大值。
縱坐標同理。

    def random_location_variance(self, enemy_start_location):
        x = enemy_start_location[0]
        y = enemy_start_location[1]

        x += ((random.randrange(-20, 20))/100) * enemy_start_location[0]
        y += ((random.randrange(-20, 20))/100) * enemy_start_location[1]

        if x < 0:
            x = 0
        if y < 0:
            y = 0
        if x > self.game_info.map_size[0]:
            x = self.game_info.map_size[0]
        if y > self.game_info.map_size[1]:
            y = self.game_info.map_size[1]

        go_to = position.Point2(position.Pointlike((x,y)))
        return go_to

完整代碼如下

# -*- encoding: utf-8 -*-
'''
@File    :   demo.py
@Modify Time      @Author       @Desciption
------------      -------       -----------
2019/11/3 12:32   Jonas           None
'''

import sc2
from sc2 import run_game, maps, Race, Difficulty, position
from sc2.player import Bot, Computer
from sc2.constants import *
import random
import numpy as np
import cv2


class SentdeBot(sc2.BotAI):
    def __init__(self):
        # 經過計算,每分鐘大約165迭代次數
        self.ITERATIONS_PER_MINUTE = 165
        # 最大农民數量
        self.MAX_WORKERS = 50

    async def on_step(self, iteration: int):
        self.iteration = iteration
        await self.scout()
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.intel()
        await self.attack()

    ## 偵察
    async def scout(self):
        if len(self.units(UnitTypeId.OBSERVER)) > 0:
            scout = self.units(UnitTypeId.OBSERVER)[0]
            if scout.is_idle:
                enemy_location = self.enemy_start_locations[0]
                move_to = self.random_location_variance(enemy_location)
                print(move_to)
                await self.do(scout.move(move_to))

        else:
            for rf in self.units(UnitTypeId.ROBOTICSFACILITY).ready.noqueue:
                if self.can_afford(UnitTypeId.OBSERVER) and self.supply_left > 0:
                    await self.do(rf.train(UnitTypeId.OBSERVER))

    async def intel(self):
        game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)

        # UnitTypeId作為key,半徑和顏色是value
        draw_dict = {
            UnitTypeId.NEXUS: [15, (0, 255, 0)],
            UnitTypeId.PYLON: [3, (20, 235, 0)],
            UnitTypeId.PROBE: [1, (55, 200, 0)],
            UnitTypeId.ASSIMILATOR: [2, (55, 200, 0)],
            UnitTypeId.GATEWAY: [3, (200, 100, 0)],
            UnitTypeId.CYBERNETICSCORE: [3, (150, 150, 0)],
            UnitTypeId.STARGATE: [5, (255, 0, 0)],
            UnitTypeId.ROBOTICSFACILITY: [5, (215, 155, 0)],

            UnitTypeId.VOIDRAY: [3, (255, 100, 0)],
            # OBSERVER: [3, (255, 255, 255)],
        }

        for unit_type in draw_dict:
            for unit in self.units(unit_type).ready:
                pos = unit.position
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1)

        # 主基地名稱
        main_base_names = ["nexus", "supplydepot", "hatchery"]
        # 記錄敵方基地位置
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            # 不是主基地建築,畫小一些
            if enemy_building.name.lower() not in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 5, (200, 50, 212), -1)
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            if enemy_building.name.lower() in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 15, (0, 0, 255), -1)

        for enemy_unit in self.known_enemy_units:

            if not enemy_unit.is_structure:
                worker_names = ["probe", "scv", "drone"]
                # if that unit is a PROBE, SCV, or DRONE... it's a worker
                pos = enemy_unit.position
                if enemy_unit.name.lower() in worker_names:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (55, 0, 155), -1)
                else:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (50, 0, 215), -1)

        for obs in self.units(UnitTypeId.OBSERVER).ready:
            pos = obs.position
            cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (255, 255, 255), -1)

        # flip horizontally to make our final fix in visual representation:
        flipped = cv2.flip(game_data, 0)
        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)

        cv2.imshow('Intel', resized)
        cv2.waitKey(1)

    def random_location_variance(self, enemy_start_location):
        x = enemy_start_location[0]
        y = enemy_start_location[1]

        x += ((random.randrange(-20, 20)) / 100) * enemy_start_location[0]
        y += ((random.randrange(-20, 20)) / 100) * enemy_start_location[1]

        if x < 0:
            x = 0
        if y < 0:
            y = 0
        if x > self.game_info.map_size[0]:
            x = self.game_info.map_size[0]
        if y > self.game_info.map_size[1]:
            y = self.game_info.map_size[1]

        go_to = position.Point2(position.Pointlike((x, y)))
        return go_to

    # 建造农民
    async def build_workers(self):
        # 星靈樞鈕*16(一個基地配備16個农民)大於農民數量並且現有农民數量小於MAX_WORKERS
        if len(self.units(UnitTypeId.NEXUS)) * 16 > len(self.units(UnitTypeId.PROBE)) and len(
                self.units(UnitTypeId.PROBE)) < self.MAX_WORKERS:
            # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
            for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
                # 是否有50晶體礦建造农民
                if self.can_afford(UnitTypeId.PROBE):
                    await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0, vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR, vaspene))

    ## 開礦
    async def expand(self):
        # (self.iteration / self.ITERATIONS_PER_MINUTE)是一個緩慢遞增的值,動態開礦
        if self.units(UnitTypeId.NEXUS).amount < self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(
                UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
            if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):
                if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                    await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)
            # 否則建造折躍門
            # (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一個緩慢遞增的值
            # elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
            elif len(self.units(UnitTypeId.GATEWAY)) < 1:
                if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                    await self.build(UnitTypeId.GATEWAY, near=pylon)
            # 控制核心存在的情況下建造机械台
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.ROBOTICSFACILITY)) < 1:
                    if self.can_afford(UnitTypeId.ROBOTICSFACILITY) and not self.already_pending(
                            UnitTypeId.ROBOTICSFACILITY):
                        await self.build(UnitTypeId.ROBOTICSFACILITY, near=pylon)

            # 控制核心存在的情況下建造星門
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
                    if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):
                        await self.build(UnitTypeId.STARGATE, near=pylon)

    ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        # for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
        #     if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:
        #
        #         if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:
        #             await self.do(gw.train(UnitTypeId.STALKER))

        for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:
            if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(UnitTypeId.VOIDRAY))

    ## 尋找目標
    def find_target(self, state):
        if len(self.known_enemy_units) > 0:
            # 隨機選取敵方單位
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_units) > 0:
            # 隨機選取敵方建築
            return random.choice(self.known_enemy_structures)
        else:
            # 返回敵方出生點位
            return self.enemy_start_locations[0]

    ## 進攻
    async def attack(self):
        # {UNIT: [n to fight, n to defend]}
        aggressive_units = {UnitTypeId.VOIDRAY: [8, 3]}

        for UNIT in aggressive_units:
            # 攻擊模式
            if self.units(UNIT).amount > aggressive_units[UNIT][0] and self.units(UNIT).amount > aggressive_units[UNIT][
                1]:
                for s in self.units(UNIT).idle:
                    await self.do(s.attack(self.find_target(self.state)))
            # 防衛模式
            elif self.units(UNIT).amount > aggressive_units[UNIT][1]:
                if len(self.known_enemy_units) > 0:
                    for s in self.units(UNIT).idle:
                        await self.do(s.attack(random.choice(self.known_enemy_units)))


## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Hard)
], realtime=False)

運行結果如下,紅色和粉紅色是敵方單位。

創建訓練數據

統計資源、人口和軍隊人口比,在intel方法添加如下代碼

        # 追蹤資源、人口和軍隊人口比
        line_max = 50
        mineral_ratio = self.minerals / 1500
        if mineral_ratio > 1.0:
            mineral_ratio = 1.0

        vespene_ratio = self.vespene / 1500
        if vespene_ratio > 1.0:
            vespene_ratio = 1.0

        population_ratio = self.supply_left / self.supply_cap
        if population_ratio > 1.0:
            population_ratio = 1.0

        plausible_supply = self.supply_cap / 200.0

        military_weight = len(self.units(UnitTypeId.VOIDRAY)) / (self.supply_cap - self.supply_left)
        if military_weight > 1.0:
            military_weight = 1.0

        # 农民/人口      worker/supply ratio
        cv2.line(game_data, (0, 19), (int(line_max * military_weight), 19), (250, 250, 200), 3)
        # 人口/200    plausible supply (supply/200.0)
        cv2.line(game_data, (0, 15), (int(line_max * plausible_supply), 15), (220, 200, 200), 3)
        # (人口-現有人口)/人口  population ratio (supply_left/supply)
        cv2.line(game_data, (0, 11), (int(line_max * population_ratio), 11), (150, 150, 150), 3)
        # 氣體/1500   gas/1500
        cv2.line(game_data, (0, 7), (int(line_max * vespene_ratio), 7), (210, 200, 0), 3)
        # 晶體礦/1500  minerals minerals/1500
        cv2.line(game_data, (0, 3), (int(line_max * mineral_ratio), 3), (0, 255, 25), 3)

運行結果如下,左下角自上而下依次是“农民/人口”,“人口/200”,“(人口-現有人口)/人口”,“氣體/1500”,“晶體礦/1500”

採集進攻行為數據,在attack方法中加入如下代碼

        if len(self.units(UnitTypeId.VOIDRAY).idle) > 0:
            choice = random.randrange(0, 4)
            target = False
            if self.iteration > self.do_something_after:
                if choice == 0:
                    # 什麼都不做
                    wait = random.randrange(20, 165)
                    self.do_something_after = self.iteration + wait

                elif choice == 1:
                    # 攻擊離星靈樞紐最近的單位
                    if len(self.known_enemy_units) > 0:
                        target = self.known_enemy_units.closest_to(random.choice(self.units(UnitTypeId.NEXUS)))

                elif choice == 2:
                    # 攻擊敵方建築
                    if len(self.known_enemy_structures) > 0:
                        target = random.choice(self.known_enemy_structures)

                elif choice == 3:
                    # 攻擊敵方出生位置
                    target = self.enemy_start_locations[0]

                if target:
                    for vr in self.units(UnitTypeId.VOIDRAY).idle:
                        await self.do(vr.attack(target))
                y = np.zeros(4)
                y[choice] = 1
                print(y)
                self.train_data.append([y, self.flipped])

輸出如下結果

···
[1. 0. 0. 0.]
[0. 0. 1. 0.]
[0. 0. 0. 1.]
[0. 0. 1. 0.]
[1. 0. 0. 0.]
···

為了使用self.flipped = cv2.flip(game_data, 0),修改

        flipped = cv2.flip(game_data, 0)
        resized = cv2.resize(flipped, dsize=None, fx=2, fy=2)

        self.flipped = cv2.flip(game_data, 0)
        resized = cv2.resize(self.flipped, dsize=None, fx=2, fy=2)

init 方法添加do_something_after和train_data

    def __init__(self):
        self.ITERATIONS_PER_MINUTE = 165
        self.MAX_WORKERS = 50
        self.do_something_after = 0
        self.train_data = []

採集攻擊數據的時候不需要畫圖,我們在類前加HEADLESS = False,intel方法代碼修改如下

        self.flipped = cv2.flip(game_data, 0)

        if not HEADLESS:
            resized = cv2.resize(self.flipped, dsize=None, fx=2, fy=2)
            cv2.imshow('Intel', resized)
            cv2.waitKey(1)

加入on_end方法,只存儲勝利的數據,在和代碼同級目錄新建train_data文件夾

    def on_end(self, game_result):
        print('--- on_end called ---')
        print(game_result)

        if game_result == Result.Victory:
            np.save("train_data/{}.npy".format(str(int(time.time()))), np.array(self.train_data))

完整代碼如下

import os
import time

import sc2
from sc2 import run_game, maps, Race, Difficulty, position, Result
from sc2.player import Bot, Computer
from sc2.constants import *
import random
import numpy as np
import cv2

HEADLESS = True
# os.environ["SC2PATH"] = 'F:\StarCraft II'

class SentdeBot(sc2.BotAI):
    def __init__(self):
        # 經過計算,每分鐘大約165迭代次數
        self.ITERATIONS_PER_MINUTE = 165
        # 最大农民數量
        self.MAX_WORKERS = 50
        self.do_something_after = 0
        self.train_data = []

    def on_end(self, game_result):
        print('--- on_end called ---')
        print(game_result)

        if game_result == Result.Victory:
            np.save("train_data/{}.npy".format(str(int(time.time()))), np.array(self.train_data))

    async def on_step(self, iteration: int):
        self.iteration = iteration
        await self.scout()
        await self.distribute_workers()
        await self.build_workers()
        await self.build_pylons()
        await self.build_assimilators()
        await self.expand()
        await self.offensive_force_buildings()
        await self.build_offensive_force()
        await self.intel()
        await self.attack()

    ## 偵察
    async def scout(self):
        if len(self.units(UnitTypeId.OBSERVER)) > 0:
            scout = self.units(UnitTypeId.OBSERVER)[0]
            if scout.is_idle:
                enemy_location = self.enemy_start_locations[0]
                move_to = self.random_location_variance(enemy_location)
                print(move_to)
                await self.do(scout.move(move_to))

        else:
            for rf in self.units(UnitTypeId.ROBOTICSFACILITY).ready.noqueue:
                if self.can_afford(UnitTypeId.OBSERVER) and self.supply_left > 0:
                    await self.do(rf.train(UnitTypeId.OBSERVER))

    async def intel(self):
        game_data = np.zeros((self.game_info.map_size[1], self.game_info.map_size[0], 3), np.uint8)

        # UnitTypeId作為key,半徑和顏色是value
        draw_dict = {
            UnitTypeId.NEXUS: [15, (0, 255, 0)],
            UnitTypeId.PYLON: [3, (20, 235, 0)],
            UnitTypeId.PROBE: [1, (55, 200, 0)],
            UnitTypeId.ASSIMILATOR: [2, (55, 200, 0)],
            UnitTypeId.GATEWAY: [3, (200, 100, 0)],
            UnitTypeId.CYBERNETICSCORE: [3, (150, 150, 0)],
            UnitTypeId.STARGATE: [5, (255, 0, 0)],
            UnitTypeId.ROBOTICSFACILITY: [5, (215, 155, 0)],

            UnitTypeId.VOIDRAY: [3, (255, 100, 0)],
            # OBSERVER: [3, (255, 255, 255)],
        }

        for unit_type in draw_dict:
            for unit in self.units(unit_type).ready:
                pos = unit.position
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), draw_dict[unit_type][0], draw_dict[unit_type][1], -1)

        # 主基地名稱
        main_base_names = ["nexus", "supplydepot", "hatchery"]
        # 記錄敵方基地位置
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            # 不是主基地建築,畫小一些
            if enemy_building.name.lower() not in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 5, (200, 50, 212), -1)
        for enemy_building in self.known_enemy_structures:
            pos = enemy_building.position
            if enemy_building.name.lower() in main_base_names:
                cv2.circle(game_data, (int(pos[0]), int(pos[1])), 15, (0, 0, 255), -1)

        for enemy_unit in self.known_enemy_units:

            if not enemy_unit.is_structure:
                worker_names = ["probe", "scv", "drone"]
                # if that unit is a PROBE, SCV, or DRONE... it's a worker
                pos = enemy_unit.position
                if enemy_unit.name.lower() in worker_names:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (55, 0, 155), -1)
                else:
                    cv2.circle(game_data, (int(pos[0]), int(pos[1])), 3, (50, 0, 215), -1)

        for obs in self.units(UnitTypeId.OBSERVER).ready:
            pos = obs.position
            cv2.circle(game_data, (int(pos[0]), int(pos[1])), 1, (255, 255, 255), -1)


        # 追蹤資源、人口和軍隊人口比
        line_max = 50
        mineral_ratio = self.minerals / 1500
        if mineral_ratio > 1.0:
            mineral_ratio = 1.0

        vespene_ratio = self.vespene / 1500
        if vespene_ratio > 1.0:
            vespene_ratio = 1.0

        population_ratio = self.supply_left / self.supply_cap
        if population_ratio > 1.0:
            population_ratio = 1.0

        plausible_supply = self.supply_cap / 200.0

        military_weight = len(self.units(UnitTypeId.VOIDRAY)) / (self.supply_cap - self.supply_left)
        if military_weight > 1.0:
            military_weight = 1.0

        # 农民/人口      worker/supply ratio
        cv2.line(game_data, (0, 19), (int(line_max * military_weight), 19), (250, 250, 200), 3)
        # 人口/200    plausible supply (supply/200.0)
        cv2.line(game_data, (0, 15), (int(line_max * plausible_supply), 15), (220, 200, 200), 3)
        # (人口-現有人口)/人口  population ratio (supply_left/supply)
        cv2.line(game_data, (0, 11), (int(line_max * population_ratio), 11), (150, 150, 150), 3)
        # 氣體/1500   gas/1500
        cv2.line(game_data, (0, 7), (int(line_max * vespene_ratio), 7), (210, 200, 0), 3)
        # 晶體礦/1500  minerals minerals/1500
        cv2.line(game_data, (0, 3), (int(line_max * mineral_ratio), 3), (0, 255, 25), 3)




        # flip horizontally to make our final fix in visual representation:
        self.flipped = cv2.flip(game_data, 0)

        if HEADLESS:
            resized = cv2.resize(self.flipped, dsize=None, fx=2, fy=2)

            cv2.imshow('Intel', resized)
            cv2.waitKey(1)

    def random_location_variance(self, enemy_start_location):
        x = enemy_start_location[0]
        y = enemy_start_location[1]

        x += ((random.randrange(-20, 20)) / 100) * enemy_start_location[0]
        y += ((random.randrange(-20, 20)) / 100) * enemy_start_location[1]

        if x < 0:
            x = 0
        if y < 0:
            y = 0
        if x > self.game_info.map_size[0]:
            x = self.game_info.map_size[0]
        if y > self.game_info.map_size[1]:
            y = self.game_info.map_size[1]

        go_to = position.Point2(position.Pointlike((x, y)))
        return go_to

    # 建造农民
    async def build_workers(self):
        # 星靈樞鈕*16(一個基地配備16個农民)大於農民數量並且現有农民數量小於MAX_WORKERS
        if len(self.units(UnitTypeId.NEXUS)) * 16 > len(self.units(UnitTypeId.PROBE)) and len(
                self.units(UnitTypeId.PROBE)) < self.MAX_WORKERS:
            # 星靈樞紐(NEXUS)無隊列建造,可以提高晶體礦的利用率,不至於佔用資源
            for nexus in self.units(UnitTypeId.NEXUS).ready.noqueue:
                # 是否有50晶體礦建造农民
                if self.can_afford(UnitTypeId.PROBE):
                    await self.do(nexus.train(UnitTypeId.PROBE))

    ## 建造水晶
    async def build_pylons(self):
        ## 供應人口和現有人口之差小於5且建築不是正在建造
        if self.supply_left < 5 and not self.already_pending(UnitTypeId.PYLON):
            nexuses = self.units(UnitTypeId.NEXUS).ready
            if nexuses.exists:
                if self.can_afford(UnitTypeId.PYLON):
                    await self.build(UnitTypeId.PYLON, near=nexuses.first)

    ## 建造吸收廠
    async def build_assimilators(self):
        for nexus in self.units(UnitTypeId.NEXUS).ready:
            # 在瓦斯泉上建造吸收廠
            vaspenes = self.state.vespene_geyser.closer_than(15.0, nexus)
            for vaspene in vaspenes:
                if not self.can_afford(UnitTypeId.ASSIMILATOR):
                    break
                worker = self.select_build_worker(vaspene.position)
                if worker is None:
                    break
                if not self.units(UnitTypeId.ASSIMILATOR).closer_than(1.0, vaspene).exists:
                    await self.do(worker.build(UnitTypeId.ASSIMILATOR, vaspene))

    ## 開礦
    async def expand(self):
        # (self.iteration / self.ITERATIONS_PER_MINUTE)是一個緩慢遞增的值,動態開礦
        if self.units(UnitTypeId.NEXUS).amount < self.iteration / self.ITERATIONS_PER_MINUTE and self.can_afford(
                UnitTypeId.NEXUS):
            await self.expand_now()

    ## 建造進攻性建築
    async def offensive_force_buildings(self):
        # print(self.iteration / self.ITERATIONS_PER_MINUTE)
        if self.units(UnitTypeId.PYLON).ready.exists:
            pylon = self.units(UnitTypeId.PYLON).ready.random
            # 根據神族建築科技圖,折躍門建造過後才可以建造控制核心
            if self.units(UnitTypeId.GATEWAY).ready.exists and not self.units(UnitTypeId.CYBERNETICSCORE):
                if self.can_afford(UnitTypeId.CYBERNETICSCORE) and not self.already_pending(UnitTypeId.CYBERNETICSCORE):
                    await self.build(UnitTypeId.CYBERNETICSCORE, near=pylon)
            # 否則建造折躍門
            # (self.iteration / self.ITERATIONS_PER_MINUTE)/2 是一個緩慢遞增的值
            # elif len(self.units(UnitTypeId.GATEWAY)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
            elif len(self.units(UnitTypeId.GATEWAY)) < 1:
                if self.can_afford(UnitTypeId.GATEWAY) and not self.already_pending(UnitTypeId.GATEWAY):
                    await self.build(UnitTypeId.GATEWAY, near=pylon)
            # 控制核心存在的情況下建造机械台
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.ROBOTICSFACILITY)) < 1:
                    if self.can_afford(UnitTypeId.ROBOTICSFACILITY) and not self.already_pending(
                            UnitTypeId.ROBOTICSFACILITY):
                        await self.build(UnitTypeId.ROBOTICSFACILITY, near=pylon)

            # 控制核心存在的情況下建造星門
            if self.units(UnitTypeId.CYBERNETICSCORE).ready.exists:
                if len(self.units(UnitTypeId.STARGATE)) < ((self.iteration / self.ITERATIONS_PER_MINUTE) / 2):
                    if self.can_afford(UnitTypeId.STARGATE) and not self.already_pending(UnitTypeId.STARGATE):
                        await self.build(UnitTypeId.STARGATE, near=pylon)

    ## 造兵
    async def build_offensive_force(self):
        # 無隊列化建造
        # for gw in self.units(UnitTypeId.GATEWAY).ready.noqueue:
        #     if not self.units(UnitTypeId.STALKER).amount > self.units(UnitTypeId.VOIDRAY).amount:
        #
        #         if self.can_afford(UnitTypeId.STALKER) and self.supply_left > 0:
        #             await self.do(gw.train(UnitTypeId.STALKER))

        for sg in self.units(UnitTypeId.STARGATE).ready.noqueue:
            if self.can_afford(UnitTypeId.VOIDRAY) and self.supply_left > 0:
                await self.do(sg.train(UnitTypeId.VOIDRAY))

    ## 尋找目標
    def find_target(self, state):
        if len(self.known_enemy_units) > 0:
            # 隨機選取敵方單位
            return random.choice(self.known_enemy_units)
        elif len(self.known_enemy_units) > 0:
            # 隨機選取敵方建築
            return random.choice(self.known_enemy_structures)
        else:
            # 返回敵方出生點位
            return self.enemy_start_locations[0]

    ## 進攻
    async def attack(self):
        if len(self.units(UnitTypeId.VOIDRAY).idle) > 0:
            choice = random.randrange(0, 4)
            target = False
            if self.iteration > self.do_something_after:
                if choice == 0:
                    # 什麼都不做
                    wait = random.randrange(20, 165)
                    self.do_something_after = self.iteration + wait

                elif choice == 1:
                    # 攻擊離星靈樞紐最近的單位
                    if len(self.known_enemy_units) > 0:
                        target = self.known_enemy_units.closest_to(random.choice(self.units(UnitTypeId.NEXUS)))

                elif choice == 2:
                    # 攻擊敵方建築
                    if len(self.known_enemy_structures) > 0:
                        target = random.choice(self.known_enemy_structures)

                elif choice == 3:
                    # 攻擊敵方出生位置
                    target = self.enemy_start_locations[0]

                if target:
                    for vr in self.units(UnitTypeId.VOIDRAY).idle:
                        await self.do(vr.attack(target))
                y = np.zeros(4)
                y[choice] = 1
                print(y)
                self.train_data.append([y, self.flipped])


## 啟動遊戲
run_game(maps.get("AcidPlantLE"), [
    Bot(Race.Protoss, SentdeBot()), Computer(Race.Terran, Difficulty.Medium)
], realtime=False)

可以看到train_data文件夾下存儲了勝利數據

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

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

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

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

【python測試開發棧】帶你徹底搞明白python3編碼原理

在之前的文章中,我們介紹過編碼格式的發展史:[文章傳送門-todo]。今天我們通過幾個例子,來徹底搞清楚python3中的編碼格式原理,這樣你之後寫python腳本時碰到編碼問題,才能有章可循。

我們先搞清楚幾個概念:

  • 系統默認編碼:指python解釋器默認的編碼格式,在python文件頭部沒有聲明其他編碼格式時,python3默認的編碼格式是utf-8。
  • 本地默認編碼:操作系統默認的編碼,常見的Windows的默認編碼是gbk,Linux的默認編碼是UTF-8。
  • python文件頭部聲明編碼格式:修改的是文件的默認編碼格式,只是會影響python解釋器讀取python文件時的編碼格式,並不會改變系統默認編碼和本地默認編碼。

通過python自帶的庫,可以查看系統默認編碼和本地默認編碼

Python 3.7.4 (tags/v3.7.4:e09359112e, Jul  8 2019, 20:34:20) [MSC v.1916 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import sys
>>> sys.getdefaultencoding()
'utf-8'
>>> import locale
>>> locale.getdefaultlocale()
('zh_CN', 'cp936')
>>>

注意,因為我在windows系統的電腦上 進行測試,所以系統默認編碼返回“cp936”, 這是代碼頁(是字符編碼集的別名),而936對應的就是gbk。如果你在linux或者mac上執行上面的代碼,應該會返回utf-8編碼。

其實總結來看,容易出現亂碼的場景,基本都與讀寫程序有關,比如:讀取/寫入某個文件,或者從網絡流中讀取數據等,因為這個過程中涉及到了編碼解碼的過程,只要編碼和解碼的編碼格式對應不上,就容易出現亂碼。下面我們舉兩個具體的例子,來驗證下python的編碼原理,幫助你理解這個過程。注意:下面的例子都是在pycharm中寫的。

01默認的編碼格式

我們新建一個encode_demo.py的文件,其文件默認的編碼格式是UTF-8(可以從pycharm右下角看到編碼格式),代碼如下:

"""
    @author: asus
    @time: 2019/11/21
    @function: 驗證編碼格式
"""
import sys, locale


def write_str_default_encode():
    s = "我是一個str"
    print(s)
    print(type(s))
    print(sys.getdefaultencoding())
    print(locale.getdefaultlocale())

    with open("utf_file", "w", encoding="utf-8") as f:
        f.write(s)
    with open("gbk_file", "w", encoding="gbk") as f:
        f.write(s)
    with open("jis_file", "w", encoding="shift-jis") as f:
        f.write(s)


if __name__ == '__main__':
    write_str_default_encode()

我們先來猜測下結果,因為我們沒有聲明編碼格式,所以python解釋器默認用UTF-8去解碼文件,因為文件默認編碼格式就是UTF-8,所以字符串s可以正常打印。同時以UTF-8編碼格式寫文件不會出現亂碼,而以gbk和shift-jis(日文編碼)寫文件會出現亂碼(這裏說明一點,我是用pycharm直接打開生成的文件查看的,編輯器默認編碼是UTF-8,如果在windows上用記事本打開則其默認編碼跟隨系統是GBK,gbk_file和utf_file均不會出現亂碼,只有jis_file是亂碼),我們運行看下結果:

# 運行結果
我是一個str
<class 'str'>
utf-8
('zh_CN', 'cp936')

# 寫文件utf_file、gbk_file、jis_file文件內容分別是:
我是一個str
����һ��str
�䐥�꘢str

和我們猜測的結果一致,下面我們做個改變,在文件頭部聲明個編碼格式,再來看看效果。

02 python頭文件聲明編碼格式

因為上面文件encode_demo.py的格式是UTF-8,那麼我們就將其變為gbk編碼。同樣的我們先來推測下結果,在pycharm中,在python文件頭部聲明編碼為gbk后(頭部加上 # coding=gbk ),文件的編碼格式變成gbk,同時python解釋器會用gbk去解碼encode_demo.py文件,所以運行結果應該和用UTF-8編碼時一樣。運行結果如下:

# 運行結果
我是一個str
<class 'str'>
utf-8
('zh_CN', 'cp936')

# 寫文件utf_file、gbk_file、jis_file文件內容分別是:
我是一個str
����һ��str
�䐥�꘢str

結果確實是一樣的,證明我們推論是正確的。接下來我們再做個嘗試,假如我們將(# coding=gbk)去掉(需要注意,在pycharm中將 # coding=gbk去掉,並不會改變文件的編碼格式,也就是說encode_demo.py還是gbk編碼),我們再運行一次看結果:

  File "D:/codespace/python/pythonObject/pythonSample/basic/encodeDemo/encode_demo.py", line 4
SyntaxError: Non-UTF-8 code starting with '\xd1' in file D:/codespace/python/pythonObject/pythonSample/basic/encodeDemo/encode_demo.py on line 5, but no encoding declared; see http://python.org/dev/peps/pep-0263/ for details

運行直接報錯了,我們加個斷點,看看具體的異常信息:

看錯誤提示是UnicodeDecodeError,python解釋器在對encode_demo.py文件解碼時,使用默認的UTF-8編碼,但是文件本身是gbk編碼,所以當碰到有中文沒辦法識別時,就拋出DecodeError。

03 敲黑板,划重點

python3中的str和bytes

python3的重要特性之一就是對字符串和二進制流做了嚴格的區分,我們聲明的字符串都是str類型,不過Str和bytes是可以相互轉換的:

def str_transfor_bytes():
    s = '我是一個測試Str'
    print(type(s))
    # str 轉bytes
    b = s.encode()
    print(b)
    print(type(b))
    # bytes轉str
    c = b.decode('utf-8')
    print(c)
    print(type(c))


if __name__ == '__main__':
    str_transfor_bytes()

需要注意一點:在調用encode()和decode()方法時,如果不傳參數,則會使用python解釋器默認的編碼格式UTF-8(如果不在python頭文件聲明編碼格式)。但是如果傳參的話,encode和decode使用的編碼格式要能對應上。

python3默認編碼是UTF-8?還是Unicode?

經常在很多文章里看到,python3的默認編碼格式是Unicode,但是我在本文中卻一直在說python3的默認編碼格式是UTF-8,那麼哪種說法是正確的呢?其實兩種說法都對,主要得搞清楚Unicode和UTF-8的區別(之前文章有提到):

  • Unicode是一個字符集,說白了就是把各種編碼的映射關係全都整合起來,不過它是不可變長的,全部都以兩個字節或四個字節來表示,佔用的內存空間比較大。
  • UTF-8是Unicode的一種實現方式,主要對 Unicode 碼的數據進行轉換,方便存儲和網絡傳輸 。它是可變長編碼,比如對於英文字母,它使用一個字節就可以表示。

在python3內存中使用的字符串全都是Unicode碼,當python解釋器解析python文件時,默認使用UTF-8編碼。

open()方法默認使用本地編碼

在上面的例子中,我們往磁盤寫入文件時,都指定了編碼格式。如果不指定編碼格式,那麼默認將使用操作系統本地默認的編碼格式,比如:Linux默認是UTF-8,windows默認是GBK。其實這也好理解,因為和磁盤交互,肯定要考慮操作系統的編碼格式。這有區別於encode()和decode()使用的是python解釋器的默認編碼格式,千萬別搞混淆了。

總結

不知道你看完上面的例子后,是否已經徹底理解了python3的編碼原理。不過所有的編碼問題,都逃不過“編碼”和“解碼”兩個過程,當你碰到編碼問題時,先確定源文件使用的編碼,再確定目標文件需要的編碼格式,只要能匹配,一般就可以解決編碼的問題。

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

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

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

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

fastjason常用方法

什麼是fastjson?

Fastjson是一個Java語言編寫的高性能功能完善的JSON庫。它採用一種“假定有序快速匹配”的算法,把JSON Parse的性能提升到極致,是目前Java語言中最快的JSON庫。Fastjson接口簡單易用,已經被廣泛使用在緩存序列化、協議交互、Web輸出、Android客戶端等多種應用場景。

主要特點:

  • 快速FAST (比其它任何基於Java的解析器和生成器更快,包括jackson)
  • 強大(支持普通JDK類包括任意Java Bean Class、Collection、Map、Date或enum)
  • 零依賴(沒有依賴其它任何類庫除了JDK)

背景

最近關於fastjson的消息,引起了很多人的關注!

fastjson爆出重大漏洞,攻擊者可使整個業務癱瘓

漏洞描述

常用JSON組件FastJson存在遠程代碼執行漏洞,攻擊者可通過精心構建的json報文對目標服務器執行任意命令,從而獲得服務器權限。此次爆發的漏洞為以往漏洞中autoType的繞過。

影響範圍

FastJson < 1.2.48

很多開發者才猛然發現,fastjson已經深入到我們開發工作的方方面面。那麼除了趕快升級你的json外,我們來挖挖fastjson最常用的用法。

fastjson常用方式

1.maven依賴(記得升級到1.2.48以上版本哦)

        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
         <groupId>com.alibaba</groupId>
         <artifactId>fastjson</artifactId>
         <version>1.2.62</version>
        </dependency>    

2.FastJson對於json格式字符串的解析主要用到了一下三個類:

(1)JSON:fastJson的解析器,用於JSON格式字符串與JSON對象及javaBean之間的轉換。

(2)JSONObject:fastJson提供的json對象。

(3)JSONArray:fastJson提供json數組對象。

3.常用方式

3.1 string和java對象

 

實例1:對象轉json字符串

        Map<String,String> map=new HashMap<String,String>();
        map.put("code","0");
        map.put("message","ok");
        String json=JSON.toJSONString(map);
        System.out.println(json);

輸出結果為:

{"code":"0","message":"ok"}

實例2:字符串轉對象

        Map<String,String> map=new HashMap<String,String>();
        map.put("code","0");
        map.put("message","ok");
        String json=JSON.toJSONString(map);
        System.out.println(json);
        
        Map obj=(Map)JSON.parse(json);
        System.out.println("code="+obj.get("code")+",message="+obj.get("message"));

輸出結果

{"code":"0","message":"ok"}
code=0,message=ok

3.2 工具類JSONObject

    public static void main(String[] args) {
        Map<String,String> map=new HashMap<String,String>();
        map.put("code","0");
        map.put("message","ok");
        String json=JSON.toJSONString(map);
        System.out.println(json);
        
        Map obj=(Map)JSON.parse(json);
        System.out.println("code="+obj.get("code")+",message="+obj.get("message"));        
        
        String code=JSON.parseObject(json).getString("code");
        String message=JSON.parseObject(json).getString("message");
        System.out.println("code="+code+",message="+message);
    }

輸出結果

{"code":"0","message":"ok"}
code=0,message=ok
code=0,message=ok

3.3 數組對象

List<user> list=new ArrayList<user>(JSONArray.parseArray(jsonString,user.class)); 

Fastjson 與各種JSON庫的性能比較:

 

json庫 序列化性能 反序列化性能 jar大小
fastjson 1201 1216 fastjson-1.1.26.jar(356k)
fastjson-1.1.25-android.jar(226k)
jackson 1408 1915 jackson-annotations-2.1.1.jar(34k)
jackson-core-2.1.1.jar(206k)
jackson-databind-2.1.1.jar(922k)
總共1162k
gson 7421 5065 gson-2.2.2.jar(189k)
json-lib 27555 87292 json-lib-2.4-jdk15.jar(159k)


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

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

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

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

德國頒佈最新電動汽車補貼計畫 共投入12億歐元

據報導,自7月1日起,德國頒佈最新的電動車補貼計畫,截止目前已經有近2000位申請者,其中寶馬車主占多數。  
  為了促進電動車等環保車型的普及,德國為每位購買電動車的消費者提供4000歐元的補貼,插電式混合動力車的補貼為3000歐元。在計畫實施後,有1791位插電式混合動力車的車主申請了補貼,其中有581位購買了寶馬的車型。同時還有444位申請者購買了雷諾車型,大眾汽車買主為154位。   據統計,目前德國人汽車擁有量為4500萬輛,而其中僅有5萬輛是純電動或者是混合動力車輛。為改善這一情況,德國此次計畫共投入12億歐元,由政府和汽車製造商平攤,希望能夠在2019年6月底,即計畫截止期前售出40萬輛電動車。   文章來源:環球網

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

【其他文章推薦】

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

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

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