『Microservices & Nameko』Python 微服務實踐

時間 2019-11-30

標籤 microservices nameko python 微服實踐欄目 Python 简体版

原文原文鏈接

1. Microservices

微服務最近一二年很是熱門，談論也比較多，簡單的說，微服務將單一應用程序做爲由衆多小型服務構成之套件加以開發的方式，其中各項服務都擁有本身的進程並利用輕量化機制（一般爲HTTP源API）實現通訊。下面來一張示例圖:html

Microservices Architecture: python

上面2幅圖已經形象說明微服務是什麼東西了，同時軟件部署方式須要創建在容器上。微服務相關生態會在Java和Go語言中比較成熟，尤爲是Java。而Python做爲後端，這方面會比較弱一點，微服務框架目前能看到了也就Nameko，而且技術也沒那麼成熟，因目前業務Python場景比Go語言稍多，因此先來玩一下Python如何玩微服務吧。mysql

2. Service Mesh and Serverless

關於微服務，還有2個概念也比較熱，下面簡單說起一下。git

2.1 Service Mesh

Service Mesh 服務網格，這個概念剛開始晦澀難懂，網上也有人說事下一代微服務，簡單的說，當成千上萬的微服務部署在Kubernetes 上，總體來講也是至關複雜的，由於每一個微服務都須要健康檢查、處理錯誤、延時等，而Kubernetes雖然能夠提供健康檢查和自動恢復，可是還須要熔斷模式、服務發現、API管理、加密校驗等等。而這些就是 Service Mesh須要解決的問題。github

更加詳細的介紹： philcalcado.com/2017/08/03/…redis

做爲服務間通訊的基礎設施層，能夠將它比做是應用程序或者說微服務間的TCP/IP，負責服務之間的網絡調用、限流、熔斷和監控。對於編寫應用程序來講通常無須關心TCP/IP這一層（好比經過 HTTP 協議的 RESTful 應用），一樣使用Service Mesh也就無須關係服務之間的那些原來是經過應用程序或者其餘框架實現的事情，好比Spring Cloud，如今只要交給Service Mesh就能夠了。sql

Service Mesh有以下幾個特色：docker

應用程序間通信的中間層
輕量級網絡代理
應用程序無感知
解耦應用程序的重試/超時、監控、追蹤和服務發現

Service Mesh的架構以下圖所示： json

比較流行的開源軟件有Istio,有機會再去玩一下。flask

2.2 Serverless

無服務架構，第一次接觸是在AWS的技術峯會上，簡單的說就是不須要關心服務器，整個計算堆棧，包括運行功能代碼的操做系統進程，徹底由雲提供商管理。更增強化了 DevOps 的理念。

實際玩過AWS Lambda 無服務應用程序，確實很方便，簡化爲一個函數，經過 API Gateway + Lambda 則可實現Web服務。按請求量收費，這一點目前以爲很坑，尤爲是請求量大時，產生的費用遠遠比本身將應用部署在Docker上會貴不少。因此目前無服務架構的場景也是很是適合一些一次性任務，請求量調用很少的場景來講會很是方便，開發者成員就能夠本身開發本身部署，再也不須要關心服務器。能夠免除全部運維性操做，開發人員能夠更加專一於核心業務的開發，實現快速上線和迭代。

3. Python framework for building microservices

3.1 Nameko Introduce

Nameko是Python中的微服務框架，git地https://github.com/nameko/nameko，受歡迎度暫時還不高，官方文檔的介紹實現了：

It comes with built-in support for:

RPC over AMQP
Asynchronous events (pub-sub) over AMQP
Simple HTTP GET and POST
Websocket RPC and subscriptions (experimental)

簡單的說RPC創建在AMQP上，在AMQP上實現了發佈訂閱，實現了簡單的HTTP服務，還有Websocket RPC。

這簡直跟Java的生態徹底感受是小兒科。架構經過RabbitMQ做爲message broker，供給各個Nameko Service之間的通訊。

更多的細節請查看官方文檔。

3.2 Practice

接下來實踐一下，以某一個業務場景爲例。

場景: 假設社交場景中，評論別人的文章，服務器給文章做者推送一條消息告知有人評論，同時評論必須得先註冊。

涉及到2個微服務，註冊服務和推送服務，同時有一個評論接口。

3.2.1 環境搭建:

python3.5+
RabbitMQ
Redis 3.2.1
Nameko 2.11.0
Swagger
Flask 1.0.2

首先須要準備Python3環境，Redis簡單起見做爲用戶登陸註冊的存儲，Nameko用pip安裝，RabbitMQ最好用Docker安裝。

# RabbitMQ docker 安裝命令
docker search rabbitmq
docker pull rabbitmq:3.7-rc-management
docker run -d --hostname my-rabbit --name some-rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3.7-rc-management
# 須要默認運行在5672端口

Doc: https://github.com/docker-library/docs/tree/master/rabbitmq

# nameko 運行服務命令:
nameko run service --broker amqp://guest:guest@localhost

其中 guest:guest是RabbitMQ Docker鏡像的用戶名和密碼
複製代碼

同時爲了方便API測試，經過flasgger提供Swagger UI進行集成Flask。

準備好環境後，開始代碼部分演示。

3.2.2 代碼演示:

├── app
│   └── api.py
├── dependence
│   ├── __init__.py
│   └── services
│       ├── __init__.py
│       ├── config.py
│       └── redis_service.py
└── microservices
    ├── push.py
    └── register.py
複製代碼

代碼結構如上：

app中存儲的是api接口服務。
dependence能夠理解爲基礎模塊，可能不少的微服務都依賴的封裝好的服務，好比redis,mysql的接口，通常用Git倉庫的話，能夠submodule到具體服務的倉庫下，這裏測試就所有放一個倉庫了。
microservices 微服務代碼，這裏演示2個服務，註冊和推送服務。

須要實踐的是2個功能：

API代碼中如何調用微服務
微服務中如何調用其餘微服務

先介紹一下dependence中的代碼:

# content of redis_service
class RedisService(object):
    def __init__(self):
        self.redis_instance = RedisClient.get_redis(
            config.REDIS_NAME, config.REDIS_HOST, config.REDIS_PORT,
            config.REDIS_DB)
        self.users_key = "users"
        self.users_data_key = "users_data"

    def check_registered_and_get_info(self, u_id):
        """ Check if the user is registered and return user information if registered. """
        user_data = self.redis_instance.hget(self.users_data_key, u_id)
        if not user_data:
            return False, None
        return True, json.loads(user_data)

    def check_email_is_registered(self, email):
        u_id = self.redis_instance.hget(self.users_key, email)
        return u_id

    def register(self, u_id, email, data):
        self.redis_instance.hset(self.users_key, email, u_id)
        result = self.redis_instance.hset(self.users_data_key, u_id,json.dumps(data))
        return result
複製代碼

check_registered_and_get_info 校驗是否已經註冊，若是已經註冊則獲取用戶信息返回
check_email_is_registered 檢查郵箱是否重複註冊
register 註冊並存儲用戶信息

接下來看API代碼部分:

# content of api.py
import time
import random
from flask import Flask, request, jsonify
from flasgger import Swagger
from nameko.standalone.rpc import ClusterRpcProxy
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--port", help="app running port", type=int, default=5000)
parse_args = parser.parse_args()

app = Flask(__name__)
Swagger(app)

CONFIG = {'AMQP_URI': "amqp://guest:guest@localhost"}
複製代碼

代碼校長，分開展現，這部分是引用部分，須要nameko和Swagger，nameko是提供微服務的RPC服務代理，同時須要提供CONFIG，內容是Message Broker地址，其實就RabbitMQ。 Swagger是和Flask結合，方便網頁界面進行API測試。

# content of api.py
@app.route('/api/v1/comment', methods=['POST'])
def comment():
    """ Comment API Parameters Explain: timestamp 評論時間 u_id 用戶id content 評論內容 article_id 文章ID article_u_id 文章做者用戶id parent_comment_id 父評論id (optional) --- parameters: - name: body in: body required: true schema: id: comment properties: timestamp: type: integer u_id: type: string content: type: string article_id: type: integer article_u_id: type: integer parent_comment_id: type: integer responses: code: description: 0 Comment Success! message: description: Error Message! data: description: return comment_id """
    data = request.json
    article_u_id = data.get("article_u_id")
    u_id = data.get("u_id")
    code, message = 0, ""
    if not article_u_id or not u_id:
        code, message = 10003, "article_u_id or u_id is null."
        response = dict(code=code, message=message, data="")
        return jsonify(response)
    with ClusterRpcProxy(CONFIG) as rpc:
        user_data = rpc.register.check_registered(u_id)
        if not user_data:
            code, message = 10004, "You need to register to comment."
            response = dict(code=code, message=message, data="")
            return jsonify(response)

        # push message
        print("Push Message: article_u_id: {}".format(article_u_id))
        result, message = rpc.push.push(article_u_id, data.get("content"))
        print("push result: {}, message: {}".format(result, message))

    # save comment data
    print("Save Comment Data: article_id: {} content: {}".format(
        data.get("article_id"), data.get("content")))

    data = dict(comment_id=int(time.time()))
    response = dict(code=0, message="", data=data)
    return jsonify(response)
複製代碼

評論接口，描述部分是提供Swagger的API接口描述（其規範須要遵循Swagger規範，具體能夠查看官方文檔），提供評論者的用戶ID,文章ID,評論的內容，文章做者用戶ID（簡單起見，直接客戶端提供，正常場景是根據文章ID找到做者的用戶ID）。

實現的功能也很是簡單，先經過調用檢查註冊服務看評論者是否有註冊，沒有就直接返回須要註冊才能評論。若是已經註冊，則調用推送服務給做者進行推送通知。以後並保存評論信息，返回評論ID。

關鍵信息就是在註冊和推送微服務的實現，保存評論信息，我這裏直接print，沒有作實際的操做。

@app.route('/api/v1/register', methods=['POST'])
def register():
    """ Register API Parameters Explain: timestamp 註冊時間 email 註冊郵箱 name 名稱 language 語言 country 國家 --- parameters: - name: body in: body required: true schema: id: data properties: timestamp: type: integer email: type: string name: type: string language: type: string country: type: string responses: code: description: 0 register success. message: description: Error Message! data: description: return u_id """

    user_data = request.json
    email = user_data.get("email")
    code, message = 0, ""
    if not email:
        code, message = 10000, "email is null."
        response = dict(code=code, message=message, data="")
        return jsonify(response)
    u_id = None
    with ClusterRpcProxy(CONFIG) as rpc:
        u_id, message = rpc.register.register(email, user_data)
    if message:
        code = 10001
    data = dict(u_id=u_id)
    response = dict(code=code, message=message, data=data)
    return jsonify(response)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(parse_args.port), debug=True)
複製代碼

這是api.py最後一段，實現的是註冊接口，簡單的說就是調用註冊服務，若是已經註冊則直接返回，不然存儲用戶信息。

關鍵在於註冊服務的實現。

運行 python api.py 打開 http://localhost:5000/apidocs/能夠看到以下界面:

點開其中一個API，能夠看到以下界面:

很是方便進行API接口調試。

接下來重點來了，演示微服務代碼部分:

import random
from nameko.rpc import rpc
import sys
sys.path.append("..")
from dependence.services import RedisService

class RegisterService(object):
    name = "register"

    def __init__(self):
        self.redis_handle = RedisService()

    @rpc
    def check_registered(self, u_id):
        is_registered, user_data =   self.redis_handle.check_registered_and_get_info(u_id)
        if is_registered:
            return user_data
        return None

    @staticmethod
    def generate_u_id():
        """
        Test Function
        """
        return str(random.randint(7000000, 9999999))

    @rpc
    def register(self, email, user_data):
        u_id = self.redis_handle.check_email_is_registered(email)
        if u_id:
            return u_id, "already registered."
        u_id = self.generate_u_id()
        register_result = self.redis_handle.register(u_id, email, user_data)
        if register_result:
            return u_id, ""
        return None, "register failed.
複製代碼

關注register的實現，須要導入nameko.rpc,而且用rpc裝飾該函數。實現很是簡單，裏面代碼就是邏輯部分，生成u_id，而後存儲到redis中。

這樣就實踐了第一個功能，在API中調用微服務。

接下來看看 push 服務的實現:

import random
from nameko.rpc import rpc, RpcProxy
import sys
sys.path.append("..")
from dependence.services import RedisService

class PushService(object):
    name = "push"
    register_rpc = RpcProxy("register")

 @rpc
    def push(self, u_id, content):
        user_data = self.register_rpc.check_registered(u_id)
        if not user_data:
        print("User:{} not existed.".format(u_id))
            return False, "not registered."
        language, country = user_data["language"], user_data["country"]

        # get language push content
        print("Push Progress: u_id: {} language: {}, country: {}, content: {}".
              format(u_id, language, country, content))

        return True, "push success."
複製代碼

push服務中須要調用註冊服務，判斷文章做者是否註冊（其實可以發表文章確定是已經註冊，這裏是指演示），這樣就微服務中調用微服務，須要額外import RpcProxy，指定註冊服務 RpcProxy("register")，而後再服務中調用便可，而且拿到用戶的信息，判斷語言和國家，推送對應的語言內容。

總體來說，Nameko這個框架，代碼層實現很是簡單，輕量級，簡單實用。可是功能不全，Python 後端應用場景很少。

3.2.3 調試

開三個終端，分別運行:

cd microservices & nameko run push
cd microservices & nameko run register

cd app & python api.py
複製代碼

打開http://localhost:5000/apidocs/#/

準備註冊數據:

註冊信息1
{
  "country": "CN",
  "email": "nameko@nameko.com",
  "language": "ZH",
  "name": "xiaohua",
  "timestamp": 1553652949
}

註冊信息2
{
  "country": "CN",
  "email": "nameko2@nameko.com",
  "language": "ZH",
  "name": "xiaoming",
  "timestamp": 1553652950
}

複製代碼

操做示例:

返回信息:

返回信息1
{
  "code": 0,
  "data": {
    "u_id": "7434029"
  },
  "message": ""
}

返回信息2
{
  "code": 0,
  "data": {
    "u_id": "8240184"
  },
  "message": ""
}
複製代碼

調用push接口:

點擊執行execute。

返回信息:

python api.py 終端打印信息:

Push Message: article_u_id: 7434029
push result: True, message: push success.
Save Comment Data: article_id: 100 content: very good.
127.0.0.1 - - [27/Mar/2019 23:24:07] "POST /api/v1/comment HTTP/1.1" 200 -
複製代碼

Python 微服務就演示到這裏，完整代碼 github 地址github.com/CrystalSkyZ…

下一篇聊一下RPC。

更多精彩文章，請關注公衆號『天澄技術雜談』