微服務(wù)遵循領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)（DDD），與開(kāi)發(fā)平臺(tái)無(wú)關(guān)。Python 微服務(wù)也不例外。Python3 的面向?qū)ο筇匦允沟冒凑?DDD 對(duì)服務(wù)進(jìn)行建模變得更加容易。

微服務(wù)架構(gòu)的強(qiáng)大之處在于它的多語(yǔ)言性。企業(yè)將其功能分解為一組微服務(wù)，每個(gè)團(tuán)隊(duì)自由選擇一個(gè)平臺(tái)。

我們的用戶管理系統(tǒng)已經(jīng)分解為四個(gè)微服務(wù)，分別是添加、查找、搜索和日志服務(wù)。添加服務(wù)在 Java 平臺(tái)上開(kāi)發(fā)并部署在 Kubernetes 集群上，以實(shí)現(xiàn)彈性和可擴(kuò)展性。這并不意味著其余的服務(wù)也要使用 Java 開(kāi)發(fā)，我們可以自由選擇適合個(gè)人服務(wù)的平臺(tái)。

讓我們選擇 Python 作為開(kāi)發(fā)查找服務(wù)的平臺(tái)。查找服務(wù)的模型已經(jīng)設(shè)計(jì)好了（參考 2022 年 3 月份的文章），我們只需要將這個(gè)模型轉(zhuǎn)換為代碼和配置。

Pythonic 方法

Python 是一種通用編程語(yǔ)言，已經(jīng)存在了大約 30 年。早期，它是自動(dòng)化腳本的首選。然而，隨著 Django 和 Flask 等框架的出現(xiàn)，它的受歡迎程度越來(lái)越高，現(xiàn)在各種領(lǐng)域中都在應(yīng)用它，如企業(yè)應(yīng)用程序開(kāi)發(fā)。數(shù)據(jù)科學(xué)和機(jī)器學(xué)習(xí)進(jìn)一步推動(dòng)了它的發(fā)展，Python 現(xiàn)在是三大編程語(yǔ)言之一。

許多人將 Python 的成功歸功于它容易編碼。這只是一部分原因。只要你的目標(biāo)是開(kāi)發(fā)小型腳本，Python 就像一個(gè)玩具，你會(huì)非常喜歡它。然而，當(dāng)你進(jìn)入嚴(yán)肅的大規(guī)模應(yīng)用程序開(kāi)發(fā)領(lǐng)域時(shí)，你將不得不處理大量的 ??if?? 和 ??else??，Python 變得與任何其他平臺(tái)一樣好或一樣壞。例如，采用一種面向?qū)ο蟮姆椒?！許多 Python 開(kāi)發(fā)人員甚至可能沒(méi)意識(shí)到 Python 支持類、繼承等功能。Python 確實(shí)支持成熟的面向?qū)ο箝_(kāi)發(fā)，但是有它自己的方式 -- Pythonic！讓我們探索一下！

領(lǐng)域模型

??AddService?? 通過(guò)將數(shù)據(jù)保存到一個(gè) MySQL 數(shù)據(jù)庫(kù)中來(lái)將用戶添加到系統(tǒng)中。??FindService?? 的目標(biāo)是提供一個(gè) REST API 按用戶名查找用戶。域模型如圖 1 所示。它主要由一些值對(duì)象組成，如 ??User?? 實(shí)體的??Name??、??PhoneNumber?? 以及 ??UserRepository??。

圖 1: 查找服務(wù)的域模型

讓我們從 ??Name?? 開(kāi)始。由于它是一個(gè)值對(duì)象，因此必須在創(chuàng)建時(shí)進(jìn)行驗(yàn)證，并且必須保持不可變。基本結(jié)構(gòu)如所示：

class Name:    value: str    def __post_init__(self):        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:            raise ValueError("Invalid Name")

如你所見(jiàn)，??Name?? 包含一個(gè)字符串類型的值。作為后期初始化的一部分，我們會(huì)驗(yàn)證它。

Python 3.7 提供了 ??@dataclass?? 裝飾器，它提供了許多開(kāi)箱即用的數(shù)據(jù)承載類的功能，如構(gòu)造函數(shù)、比較運(yùn)算符等。如下是裝飾后的 ??Name?? 類：

from dataclasses import dataclass@dataclassclass Name:    value: str    def __post_init__(self):        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:            raise ValueError("Invalid Name")

以下代碼可以創(chuàng)建一個(gè) ??Name?? 對(duì)象：

name = Name("Krishna")

??value?? 屬性可以按照如下方式讀取或?qū)懭耄?/p>

name.value = "Mohan"print(name.value)

可以很容易地與另一個(gè) ??Name?? 對(duì)象比較，如下所示：

other = Name("Mohan")if name == other:    print("same")

如你所見(jiàn)，對(duì)象比較的是值而不是引用。這一切都是開(kāi)箱即用的。我們還可以通過(guò)凍結(jié)對(duì)象使對(duì)象不可變。這是 ??Name?? 值對(duì)象的最終版本：

from dataclasses import dataclass@dataclass(frozen=True)class Name:    value: str    def __post_init__(self):        if self.value is None or len(self.value.strip()) < 8 or len(self.value.strip()) > 32:            raise ValueError("Invalid Name")

??PhoneNumber?? 也遵循類似的方法，因?yàn)樗彩且粋€(gè)值對(duì)象：

@dataclass(frozen=True)class PhoneNumber:    value: int    def __post_init__(self):        if self.value < 9000000000:            raise ValueError("Invalid Phone Number")

??User?? 類是一個(gè)實(shí)體，不是一個(gè)值對(duì)象。換句話說(shuō)，??User?? 是可變的。以下是結(jié)構(gòu)：

from dataclasses import dataclassimport datetime@dataclassclass User:    _name: Name    _phone: PhoneNumber    _since: datetime.datetime    def __post_init__(self):        if self._name is None or self._phone is None:            raise ValueError("Invalid user")        if self._since is None:            self.since = datetime.datetime.now()

你能觀察到 ??User?? 并沒(méi)有凍結(jié)，因?yàn)槲覀兿Ｍ强勺兊?。但是，我們不希望所有屬性都是可變的。?biāo)識(shí)字段如 ??_name?? 和 ??_since?? 是希望不會(huì)修改的。那么，這如何做到呢？

Python3 提供了所謂的描述符協(xié)議，它會(huì)幫助我們正確定義 getter 和 setter。讓我們使用 ??@property?? 裝飾器將 getter 添加到 ??User?? 的所有三個(gè)字段中。

@propertydef name(self) -> Name:    return self._name@propertydef phone(self) -> PhoneNumber:    return self._phone@propertydef since(self) -> datetime.datetime:    return self._since

??phone?? 字段的 setter 可以使用 ??@<字段>.setter?? 來(lái)裝飾：

@phone.setterdef phone(self, phone: PhoneNumber) -> None:    if phone is None:        raise ValueError("Invalid phone")    self._phone = phone

通過(guò)重寫(xiě) ??__str__()?? 函數(shù)，也可以為 ??User?? 提供一個(gè)簡(jiǎn)單的打印方法：

def __str__(self):    return self.name.value + " [" + str(self.phone.value) + "] since " + str(self.since)

這樣，域模型的實(shí)體和值對(duì)象就準(zhǔn)備好了。創(chuàng)建異常類如下所示：

class UserNotFoundException(Exception):    pass

域模型現(xiàn)在只剩下 ??UserRepository?? 了。Python 提供了一個(gè)名為 ??abc?? 的有用模塊來(lái)創(chuàng)建抽象方法和抽象類。因?yàn)?nbsp;??UserRepository?? 只是一個(gè)接口，所以我們可以使用 ??abc?? 模塊。

任何繼承自 ??abc.ABC?? 的類都將變?yōu)槌橄箢?，任何帶?nbsp;??@abc.abstractmethod?? 裝飾器的函數(shù)都會(huì)變?yōu)橐粋€(gè)抽象函數(shù)。下面是 ??UserRepository?? 的結(jié)構(gòu)：

from abc import ABC, abstractmethodclass UserRepository(ABC):    @abstractmethod    def fetch(self, name:Name) -> User:        pass

??UserRepository?? 遵循倉(cāng)儲(chǔ)模式。換句話說(shuō)，它在 ??User?? 實(shí)體上提供適當(dāng)?shù)?CRUD 操作，而不會(huì)暴露底層數(shù)據(jù)存儲(chǔ)語(yǔ)義。在本例中，我們只需要 ??fetch()?? 操作，因?yàn)?nbsp;??FindService?? 只查找用戶。

因?yàn)?nbsp;??UserRepository?? 是一個(gè)抽象類，我們不能從抽象類創(chuàng)建實(shí)例對(duì)象。創(chuàng)建對(duì)象必須依賴于一個(gè)具體類實(shí)現(xiàn)這個(gè)抽象類。數(shù)據(jù)層 ??UserRepositoryImpl?? 提供了 ??UserRepository?? 的具體實(shí)現(xiàn)：

class UserRepositoryImpl(UserRepository):    def fetch(self, name:Name) -> User:        pass

由于 ??AddService?? 將用戶數(shù)據(jù)存儲(chǔ)在一個(gè) MySQL 數(shù)據(jù)庫(kù)中，因此 ??UserRepositoryImpl?? 也必須連接到相同的數(shù)據(jù)庫(kù)去檢索數(shù)據(jù)。下面是連接到數(shù)據(jù)庫(kù)的代碼。注意，我們正在使用 MySQL 的連接庫(kù)。

from mysql.connector import connect, Errorclass UserRepositoryImpl(UserRepository):    def fetch(self, name:Name) -> User:        try:            with connect(                    host="mysqldb",                    user="root",                    password="admin",                    database="glarimy",                ) as connection:                with connection.cursor() as cursor:                    cursor.execute("SELECT * FROM ums_users where name=%s", (name.value,))                    row = cursor.fetchone()                    if cursor.rowcount == -1:                        raise UserNotFoundException()                    else:                        return User(Name(row[0]), PhoneNumber(row[1]), row[2])        except Error as e:            raise e

在上面的片段中，我們使用用戶 ??root?? / 密碼 ??admin?? 連接到一個(gè)名為 ??mysqldb?? 的數(shù)據(jù)庫(kù)服務(wù)器，使用名為 ??glarimy?? 的數(shù)據(jù)庫(kù)（模式）。在演示代碼中是可以包含這些信息的，但在生產(chǎn)中不建議這么做，因?yàn)檫@會(huì)暴露敏感信息。

??fetch()?? 操作的邏輯非常直觀，它對(duì) ??ums_users?? 表執(zhí)行 SELECT 查詢?；叵胍幌拢??AddService?? 正在將用戶數(shù)據(jù)寫(xiě)入同一個(gè)表中。如果 SELECT 查詢沒(méi)有返回記錄，??fetch()?? 函數(shù)將拋出 ??UserNotFoundException?? 異常。否則，它會(huì)從記錄中構(gòu)造 ??User?? 實(shí)體并將其返回給調(diào)用者。這沒(méi)有什么特殊的。

應(yīng)用層

最終，我們需要?jiǎng)?chuàng)建應(yīng)用層。此模型如圖 2 所示。它只包含兩個(gè)類：控制器和一個(gè) DTO。

圖 2: 添加服務(wù)的應(yīng)用層

眾所周知，一個(gè) DTO 只是一個(gè)沒(méi)有任何業(yè)務(wù)邏輯的數(shù)據(jù)容器。它主要用于在 ??FindService?? 和外部之間傳輸數(shù)據(jù)。我們只是提供了在 REST 層中將 ??UserRecord?? 轉(zhuǎn)換為字典以便用于 JSON 傳輸：

class UserRecord:    def toJSON(self):        return {            "name": self.name,            "phone": self.phone,            "since": self.since        }

控制器的工作是將 DTO 轉(zhuǎn)換為用于域服務(wù)的域?qū)ο?，反之亦然?？梢詮?nbsp;??find()?? 操作中觀察到這一點(diǎn)。

class UserController:    def __init__(self):        self._repo = UserRepositoryImpl()    def find(self, name: str):        try:            user: User = self._repo.fetch(Name(name))            record: UserRecord = UserRecord()            record.name = user.name.value            record.phone = user.phone.value            record.since = user.since            return record        except UserNotFoundException as e:            return None

??find()?? 操作接收一個(gè)字符串作為用戶名，然后將其轉(zhuǎn)換為 ??Name?? 對(duì)象，并調(diào)用 ??UserRepository?? 獲取相應(yīng)的 ??User?? 對(duì)象。如果找到了，則使用檢索到的 ??User`` 對(duì)象創(chuàng)建??UserRecord`?；叵胍幌?，將域?qū)ο筠D(zhuǎn)換為 DTO 是很有必要的，這樣可以對(duì)外部服務(wù)隱藏域模型。

??UserController?? 不需要有多個(gè)實(shí)例，它也可以是單例的。通過(guò)重寫(xiě) ??__new__??，可以將其建模為一個(gè)單例。

class UserController:    def __new__(self):        if not hasattr(self, ‘instance’):            self.instance = super().__new__(self)        return self.instance    def __init__(self):        self._repo = UserRepositoryImpl()    def find(self, name: str):        try:            user: User = self._repo.fetch(Name(name))            record: UserRecord = UserRecord()            record.name = user.name.getValue()            record.phone = user.phone.getValue()            record.since = user.since            return record        except UserNotFoundException as e:            return None

我們已經(jīng)完全實(shí)現(xiàn)了 ??FindService?? 的模型，剩下的唯一任務(wù)是將其作為 REST 服務(wù)公開(kāi)。

REST API

??FindService?? 只提供一個(gè) API，那就是通過(guò)用戶名查找用戶。顯然 URI 如下所示：

GET /user/{name}

此 API 希望根據(jù)提供的用戶名查找用戶，并以 JSON 格式返回用戶的電話號(hào)碼等詳細(xì)信息。如果沒(méi)有找到用戶，API 將返回一個(gè) 404 狀態(tài)碼。

我們可以使用 Flask 框架來(lái)構(gòu)建 REST API，它最初的目的是使用 Python 開(kāi)發(fā) Web 應(yīng)用程序。除了 HTML 視圖，它還進(jìn)一步擴(kuò)展到支持 REST 視圖。我們選擇這個(gè)框架是因?yàn)樗銐蚝?jiǎn)單。 創(chuàng)建一個(gè) Flask 應(yīng)用程序：

from flask import Flaskapp = Flask(__name__)

然后為 Flask 應(yīng)用程序定義路由，就像函數(shù)一樣簡(jiǎn)單：

@app.route('/user/<name>')def get(name):    pass

注意 ??@app.route?? 映射到 API ??/user/<name>??，與之對(duì)應(yīng)的函數(shù)的 ??get()??。

如你所見(jiàn)，每次用戶訪問(wèn) API 如 ??http://server:port/user/Krishna?? 時(shí)，都將調(diào)用這個(gè) ??get()?? 函數(shù)。Flask 足夠智能，可以從 URL 中提取 ??Krishna?? 作為用戶名，并將其傳遞給 ??get()?? 函數(shù)。

??get()?? 函數(shù)很簡(jiǎn)單。它要求控制器找到該用戶，并將其與通常的 HTTP 頭一起打包為 JSON 格式后返回。如果控制器返回 ??None??，則 ??get()?? 函數(shù)返回合適的 HTTP 狀態(tài)碼。

from flask import jsonify, abortcontroller = UserController()record = controller.find(name)if record is None:    abort(404)else:    resp = jsonify(record.toJSON())    resp.status_code = 200    return resp

最后，我們需要 Flask 應(yīng)用程序提供服務(wù)，可以使用 ??waitress?? 服務(wù)：

from waitress import serveserve(app, host="0.0.0.0", port=8080)

在上面的片段中，應(yīng)用程序在本地主機(jī)的 8080 端口上提供服務(wù)。最終代碼如下所示：

from flask import Flask, jsonify, abortfrom waitress import serveapp = Flask(__name__)@app.route('/user/<name>')def get(name):    controller = UserController()    record = controller.find(name)    if record is None:        abort(404)    else:        resp = jsonify(record.toJSON())        resp.status_code = 200        return respserve(app, host="0.0.0.0", port=8080)

部署

??FindService?? 的代碼已經(jīng)準(zhǔn)備完畢。除了 REST API 之外，它還有域模型、數(shù)據(jù)層和應(yīng)用程序?qū)?。下一步是?gòu)建此服務(wù)，將其容器化，然后部署到 Kubernetes 上。此過(guò)程與部署其他服務(wù)妹有任何區(qū)別，但有一些 Python 特有的步驟。

在繼續(xù)前進(jìn)之前，讓我們來(lái)看下文件夾和文件結(jié)構(gòu)：

+ ums-find-service+ ums- domain.py- data.py- app.py- Dockerfile- requirements.txt- kube-find-deployment.yml

如你所見(jiàn)，整個(gè)工作文件夾都位于 ??ums-find-service?? 下，它包含了 ??ums?? 文件夾中的代碼和一些配置文件，例如 ??Dockerfile??、??requirements.txt?? 和 ??kube-find-deployment.yml??。

??domain.py?? 包含域模型，??data.py?? 包含 ??UserRepositoryImpl??，??app.py?? 包含剩余代碼。我們已經(jīng)閱讀過(guò)代碼了，現(xiàn)在我們來(lái)看看配置文件。

第一個(gè)是 ??requirements.txt??，它聲明了 Python 系統(tǒng)需要下載和安裝的外部依賴項(xiàng)。我們需要用查找服務(wù)中用到的每個(gè)外部 Python 模塊來(lái)填充它。如你所見(jiàn)，我們使用了 MySQL 連接器、Flask 和 Waitress 模塊。因此，下面是 ??requirements.txt?? 的內(nèi)容。

Flask==2.1.1Flask_RESTfulmysql-connector-pythonwaitress

第二步是在 ??Dockerfile?? 中聲明 Docker 相關(guān)的清單，如下：

FROM python:3.8-slim-busterWORKDIR /umsADD ums /umsADD requirements.txt requirements.txtRUN pip3 install -r requirements.txtEXPOSE 8080ENTRYPOINT ["python"]CMD ["/ums/app.py"]

總的來(lái)說(shuō)，我們使用 Python 3.8 作為基線，除了移動(dòng) ??requirements.txt?? 之外，我們還將代碼從 ??ums?? 文件夾移動(dòng)到 Docker 容器中對(duì)應(yīng)的文件夾中。然后，我們指示容器運(yùn)行 ??pip3 install?? 命令安裝對(duì)應(yīng)模塊。最后，我們向外暴露 8080 端口（因?yàn)?waitress 運(yùn)行在此端口上）。

為了運(yùn)行此服務(wù)，我們指示容器使用使用以下命令：

python /ums/app.py

一旦 ??Dockerfile?? 準(zhǔn)備完成，在 ??ums-find-service?? 文件夾中運(yùn)行以下命令，創(chuàng)建 Docker 鏡像：

docker build -t glarimy/ums-find-service

它會(huì)創(chuàng)建 Docker 鏡像，可以使用以下命令查找鏡像：

docker images

嘗試將鏡像推送到 Docker Hub，你也可以登錄到 Docker。

docker logindocker push glarimy/ums-find-service

最后一步是為 Kubernetes 部署構(gòu)建清單。

在之前的文章中，我們已經(jīng)介紹了如何建立 Kubernetes 集群、部署和使用服務(wù)的方法。我假設(shè)仍然使用之前文章中的清單文件來(lái)部署添加服務(wù)、MySQL、Kafka 和 Zookeeper。我們只需要將以下內(nèi)容添加到 ??kube-find-deployment.yml?? 文件中：

apiVersion: apps/v1kind: Deploymentmetadata:name: ums-find-servicelabels:app: ums-find-servicespec:replicas: 3selector:matchLabels:app: ums-find-servicetemplate:metadata:labels:app: ums-find-servicespec:containers:- name: ums-find-serviceimage: glarimy/ums-find-serviceports:- containerPort: 8080---apiVersion: v1kind: Servicemetadata:name: ums-find-servicelabels:name: ums-find-servicespec:type: LoadBalancerports:- port: 8080selector:app: ums-find-service

上面清單文件的第一部分聲明了 ??glarimy/ums-find-service?? 鏡像的 ??FindService??，它包含三個(gè)副本。它還暴露 8080 端口。清單的后半部分聲明了一個(gè) Kubernetes 服務(wù)作為 ??FindService?? 部署的前端。請(qǐng)記住，在之前文章中，mysqldb 服務(wù)已經(jīng)是上述清單的一部分了。

運(yùn)行以下命令在 Kubernetes 集群上部署清單文件：

kubectl create -f kube-find-deployment.yml

部署完成后，可以使用以下命令驗(yàn)證容器組和服務(wù)：

kubectl get services

輸出如圖 3 所示：

圖 3: Kubernetes 服務(wù)

它會(huì)列出集群上運(yùn)行的所有服務(wù)。注意查找服務(wù)的外部 IP，使用 ??curl?? 調(diào)用此服務(wù)：

curl http://10.98.45.187:8080/user/KrishnaMohan

注意：10.98.45.187 對(duì)應(yīng)查找服務(wù)，如圖 3 所示。

如果我們使用 ??AddService?? 創(chuàng)建一個(gè)名為 ??KrishnaMohan?? 的用戶，那么上面的 ??curl?? 命令看起來(lái)如圖 4 所示：

圖 4: 查找服務(wù)

用戶管理系統(tǒng)（UMS）的體系結(jié)構(gòu)包含 ??AddService?? 和 ??FindService??，以及存儲(chǔ)和消息傳遞所需的后端服務(wù)，如圖 5 所示?？梢钥吹浇K端用戶使用 ??ums-add-service?? 的 IP 地址添加新用戶，使用 ??ums-find-service?? 的 IP 地址查找已有用戶。每個(gè) Kubernetes 服務(wù)都由三個(gè)對(duì)應(yīng)容器的節(jié)點(diǎn)支持。還要注意：同樣的 mysqldb 服務(wù)用于存儲(chǔ)和檢索用戶數(shù)據(jù)。

圖 5: UMS 的添加服務(wù)和查找服務(wù)

其他服務(wù)

UMS 系統(tǒng)還包含兩個(gè)服務(wù)：??SearchService?? 和 ??JournalService??。在本系列的下一部分中，我們將在 Node 平臺(tái)上設(shè)計(jì)這些服務(wù)，并將它們部署到同一個(gè) Kubernetes 集群，以演示多語(yǔ)言微服務(wù)架構(gòu)的真正魅力。最后，我們將觀察一些與微服務(wù)相關(guān)的設(shè)計(jì)模式。