作者：朱晗 中國電子雲
事務處理幾乎在每一個信息系統中都會涉及，它存在的意義是為了保證系統數據符合期望的，且相互關聯的數據之間不會產生矛盾，即數據狀態的一致性。
按照數據庫的經典理論，原子性、隔離性、持久性。原子性要求數據要麼修改要麼回滾，隔離性要求事務之間相互獨立不影響，持久性要求事務的執行能正確的持久化，不丟失數據。mysql 類的行式數據庫通過 mvcc 多版本視圖和 wal 預寫日志等技術的協作，實現了單個服務使用單個數據源或者單個服務使用多個數據源場景的多事務的原子性、隔離性和持久性。
鳳凰架構這本書中有描述，單個服務使用單個數據源稱之為本地事務，單個服務使用多個數據源稱之為全局事務，而分布式事務特指多個服務同時訪問多個數據源的事務處理機制。
分布式事務的實現有很多方式，如可靠性事務隊列，TCC事務，SAGA事務等。
可靠性事務隊列，也就是BASE，聽起來和強一致性的ACID，"酸鹼"格格不入，它作為最終一致性的概念起源，系統性地總結了一種針對分布式事務的技術手段。
TCC 較為煩瑣，如同名字所示，它分為以下三個階段。
SAGA 事務將事務進行了拆分，大事務拆分若干個小事務，將整個分布式事務 T 分解為 n 個子事務，同時為為每一個子事務設計對應的補償動作。盡管補償操作通常比凍結或撤銷容易實現，但保證正向、反向恢復過程的能嚴謹地進行也需要花費不少的工夫。
DBPack 的分布式事務致力於實現對用戶的業務無入侵，使用時下流行的sidecar架構，主要使用 ETCD Watch 機制來驅動分布式事務提交回滾，它對 HTTP 流量和 MYSQL 流量做了攔截代理，支持 AT 模式（自動補償 SQL）和 TCC 模式（自動補償 HTTP 請求）。
DBPack 的 AT 模式性能取決於全局鎖的釋放速度，哪個事務競爭到了全局鎖就能對業務數據做修改，在單位時間內，全局鎖的釋放速度越快，競爭到鎖的事務越多，性能越高。DBPack 創建全局事務、注冊分支事務只是在 ETCD 插入兩條 KV 數據，事務提交回滾時修改對應數據的狀態，通過 ETCD Watch 機制感知到數據的變化就能立即處理數據的提交回滾，從而在交互上減少了很多 RPC 請求。
DBPack 的 TCC 模式中，請求會先到達 sidecar 後再注冊 TCC 事務分支，確保 Prepare 先於 Cancel 執行。具體到操作的業務數據，建議使用 XID 和 BranchID 加鎖。
以上的前戲已鋪墊，後文以講解python 微服務代碼為主，不涉及 dbpack 源碼，感興趣的童鞋可去自行調試了解。
https://github.com/CECTC/dbpack-samples/blob/main/python
這裡會提到三個微服務，首先是是事務發起方，其次是訂單系統，最後是產品庫存系統。而每一個微服務，都使用dbpack代理。事務發起方請求成功後，當訂單正常commit後，產品庫存要發生正常扣除，一旦一個微服務未完成，另一個則要發生回滾，也就是說，兩個微服務系統要保持一致。
首先，模擬分布式事務發起方的服務，該服務會注冊兩個 handler，一個會發起正常的請求，走 dbpack 代理發起分布式事務，另一個會則會非正常返回。事務發起方會根據 http 的請求情況，決定是否要發起分布式事務回滾。
以下借用了 flask web 框架實現了事務發起方的兩個handler，通過兩個http請求我們可以模擬分布式事務發起或者回滾。
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
create_so_url = "http://order-svc:3001/createSo"
update_inventory_url = "http://product-svc:3002/allocateInventory"
@app.route('/v1/order/create', methods=['POST'])
def create_1():
return create_so(rollback=False)
@app.route('/v1/order/create2', methods=['POST'])
def create_2():
return create_so(rollback=True)
def create_so(rollback=True):
xid = request.headers.get("x-dbpack-xid")
so_items = [dict(
product_sysno=1,
product_name="apple iphone 13",
original_price=6799,
cost_price=6799,
deal_price=6799,
quantity=2,
)]
so_master = [dict(
buyer_user_sysno = 10001,
seller_company_code = "SC001",
receive_division_sysno = 110105,
receive_address = "beijing",
receive_zip = "000001",
receive_contact = "scott",
receive_contact_phone = "18728828296",
stock_sysno = 1,
payment_type = 1,
so_amt = 6999 * 2,
status = 10,
appid = "dk-order",
so_items = so_items,
)]
success = (jsonify(dict(success=True, message="success")), 200)
failed = (jsonify(dict(success=False, message="failed")), 400)
headers = {
"Content-Type": "application/json",
"xid": xid
}
so_req = dict(req=so_master)
resp1 = requests.post(create_so_url, headers=headers, json=so_req)
if resp1.status_code == 400:
return failed
ivt_req = dict(req=[dict(product_sysno= 1, qty=2)])
resp2 = requests.post(update_inventory_url, headers=headers, json=ivt_req)
if resp2.status_code == 400:
return failed
if rollback:
print("rollback")
return failed
return success
if __name__ == "__main__":
app.run(host="0.0.0.0", port=3000)
那麼如何使用 dbpack 代理該服務呢？
$./dist/dbpack start --config ../dbpack-samples/configs/config-aggregation.yaml
$ cat ../dbpack-samples/configs/config-aggregation.yaml
listeners:
- protocol_type: http
socket_address:
address: 0.0.0.0
port: 13000
config:
backend_host: aggregation-svc:3000
filters:
- httpDTFilter
filters:
- name: httpDTFilter
kind: HttpDistributedTransaction
conf:
appid: aggregationSvc
transaction_infos:
- request_path: "/v1/order/create"
timeout: 60000
- request_path: "/v1/order/create2"
timeout: 60000
distributed_transaction:
appid: aggregationSvc
retry_dead_threshold: 130000
rollback_retry_timeout_unlock_enable: true
etcd_config:
endpoints:
- etcd:2379
可想而知，以上的微服務兩個 handler 是通過 filters這部分的定義來配置攔截的。
接著是訂單系統。
from flask import Flask, jsonify, request
from datetime import datetime
import mysql.connector
import time
import random
app = Flask(__name__)
insert_so_master = "INSERT /*+ XID('{xid}') */ INTO order.so_master({keys}) VALUES ({placeholders})"
insert_so_item = "INSERT /*+ XID('{xid}') */ INTO order.so_item({keys}) VALUES ({placeholders})"
def conn():
retry = 0
while retry < 3:
time.sleep(5)
try:
c = mysql.connector.connect(
host="dbpack3",
port=13308,
user="dksl",
password="123456",
database="order",
autocommit=True,
)
if c.is_connected():
db_Info = c.get_server_info()
print("Connected to MySQL Server version ", db_Info)
return c
except Exception as e:
print(e.args)
retry += 1
connection = conn()
cursor = connection.cursor(prepared=True,)
@app.route('/createSo', methods=['POST'])
def create_so():
xid = request.headers.get('xid')
reqs = request.get_json()
if xid and "req" in reqs:
for res in reqs["req"]:
res["sysno"] = next_id()
res["so_id"] = res["sysno"]
res["order_date"] = datetime.now()
res_keys = [str(k) for k,v in res.items() if k != "so_items" and str(v) != ""]
so_master = insert_so_master.format(
xid=xid,
keys=", ".join(res_keys),
placeholders=", ".join(["%s"] * len(res_keys)),
)
try:
cursor.execute(so_master, tuple(res.get(k, "") for k in res_keys))
except Exception as e:
print(e.args)
so_items = res["so_items"]
for item in so_items:
item["sysno"] = next_id()
item["so_sysno"] = res["sysno"]
item_keys = [str(k) for k,v in item.items() if str(v) != "" ]
so_item = insert_so_item.format(
xid=xid,
keys=", ".join(item_keys),
placeholders=", ".join(["%s"] * len(item_keys)),
)
try:
cursor.execute(so_item, tuple(item.get(k, "") for k in item_keys))
except Exception as e:
print(e.args)
return jsonify(dict(success=True, message="success")), 200
return jsonify(dict(success=False, message="failed")), 400
def next_id():
return random.randrange(0, 9223372036854775807)
if __name__ == '__main__':
app.run(host="0.0.0.0", port=3001)
注意到 sql 中以注解的形式添加使用了 xid ，主要是方便配合 dbpack 識別後做出相應的分布式事務處理，也就是回滾還是commit。
這裡數據庫連接使用 autocommit 這種方式。同時，使用 python 中的 mysql.connector 這個 lib 來支持 sql 傳輸中的二段式加密傳輸協議，見代碼中聲明的prepared=true。
用以下命令，使用 dbpack 代理 order 微服務：
./dist/dbpack start --config ../dbpack-samples/configs/config-order.yaml
最後是產品庫存系統，詳細代碼如下：
from flask import Flask, jsonify, request
import time
import mysql.connector
app = Flask(__name__)
allocate_inventory_sql = "update /*+ XID('{xid}') */ product.inventory set available_qty = available_qty - %s, allocated_qty = allocated_qty + %s where product_sysno = %s and available_qty >= %s;"
def conn():
retry = 0
while retry < 3:
time.sleep(5)
try:
c = mysql.connector.connect(
host="dbpack2",
port=13307,
user="dksl",
password="123456",
database="product",
autocommit=True,
)
if c.is_connected():
db_Info = c.get_server_info()
print("Connected to MySQL Server version ", db_Info)
return c
except Exception as e:
print(e.args)
retry += 1
connection = conn()
cursor = connection.cursor(prepared=True,)
@app.route('/allocateInventory', methods=['POST'])
def create_so():
xid = request.headers.get('xid')
reqs = request.get_json()
if xid and "req" in reqs:
for res in reqs["req"]:
try:
cursor.execute(allocate_inventory_sql.format(xid=xid), (res["qty"], res["qty"], res["product_sysno"], res["qty"],))
except Exception as e:
print(e.args)
return jsonify(dict(success=True, message="success")), 200
return jsonify(dict(success=False, message="failed")), 400
if __name__ == '__main__':
app.run(host="0.0.0.0", port=3002)
同樣，用以下命令使用 dbpack 代理 product 微服務：
./dist/dbpack start --config ../dbpack-samples/configs/config-product.yaml
我們可以使用docker-compose一鍵拉起以上三個微服務：
docker-compose up
正常情況下，以下請求會觸發訂單系統和產品庫存系統的正常 commit：
curl -XPOST http://localhost:13000/v1/order/create
而以下命令雖然正常請求了訂單系統和產品庫存的 API，不管事務是否正常執行，由於事務發起方狀態碼不正常，要求"回滾"，所以會導致已經 commit 的微服務發生回滾，以此保證分布式系統的一致性：
curl -XPOST http://localhost:13000/v1/order/create2
大數據,微服務,分布式,Java,Python,Web前端,產品運營,交互 領取方式在篇尾!!! 基礎篇.互聯網架構,高級程序員必備視頻,Linux系統.JVM.大型分布式電商項目實戰視頻...... ...
介紹 使用python做web開發面臨的一個最大的問題就是性能,在解決C10K問題上顯的有點吃力.有些異步框架Tornado.Twisted.Gevent 等就是為了解決性能問題.這些框架在性能上有些 ...
一.簡單做一個背景說明1.為說明問題,本文簡單微服務架構示例如下 2.組件說明分布式架構,每個組件都是集群或者主備.具體說明如下:zuul service:網關,API調用都走zuul service ...
通過上一節<微服務之分布式跟蹤系統(springboot+zipkin)>我們簡單熟悉了zipkin的使用,但是收集的數據都保存在內存中重啟後數據丟失,不過zipkin的Storage除了 ...
python 微服務開發是一本講python 如果進行微服務開發的實戰類書籍,裡面包含了幾個很不錯的python 模塊,記錄下,方便後期回顧學習 處理並發的模塊 greenlet && ...
使用 Consul 作為 Python 微服務的配置中心 Consul 作為數據中心,提供了 k/v 存儲的功能,我們可以利用這個功能為 Python 微服務提供配置中心. Consul 提供了 HT ...
1. 前言 大家好,我是安果! 考慮到 Python 性能及效率性,Python Web 端一直不溫不火,JAVA 和 Golang 的微服務生態一直很繁榮,也被廣泛用於企業級應用開發當中 本篇文章 ...
這篇文章介紹一下在微服務(springboot開發)的項目中使用pintpoint監控的過程及效果展示. 背景 隨著項目微服務的進行,微服務數量逐漸增加,服務間的調用也越來越復雜,我們急切需要一個AP ...
A litmus test for whether an app has all config correctly factored out of the code is whether the co ...
在微服務架構下,微服務被拆分成多個微小的服務,每個微小的服務都部署在不同的服務器實例上,當我們定位問題,檢索日志的時候需要依次登錄每台服務器進行檢索. 這樣是不是感覺很繁瑣和效率低下.所以我們還需要一 ...
package com.example.databasetest; import android.content.Context;import android.database.sqlite.SQLi ...
在工作中, 主進程創建了子進程, 而子進程又創建了孫子進程, 然而子進程被莫名其妙的 kill 了, 結果主進程又啟動了一個子進程, 子進程又嘗試創建孫子進程, 但是這時候就有問題了, 因為孫子進程還 ...
原文:http://blog.sina.com.cn/s/blog_6254a8ca0100i0wr.html 原文也是轉的,哈哈,大家多轉轉,轉轉更健康.刪除了一些Xilinx的東西 前言 綜合是將 ...
一,出現亂碼的原因分析 1,保存文件時候,文件有自己的文件編碼,就是漢字,或者其他國語言,以什麼編碼來存儲 2,輸出的時候,要給內容指定編碼,如以網頁的形勢輸入時<meta http-equiv ...
我們可以通過二分查找法,在log(n)的時間內找到最小數的在數組中的位置,然後通過偏移來快速定位任意第K個數. 此處假設數組中沒有相同的數,原排列順序是遞增排列. 在輪轉後的有序數組中查找最小數的算法 ...
Problem Description Today is Gorwin’s birthday. So her mother want to realize her a wish. Gorwin say ...
什麼是soap? soap,簡單對象訪問協議,基於http傳輸xml數據,soap協議體是xml格式. SOAP 是一種網絡通信協議 SOAP即Simple Object Access Protoco ...
題目鏈接 題目要求: Given a binary tree, determine if it is a valid binary search tree (BST). Assume a BST is ...
量化投資實際上就是分析數據從而做出決策的過程python數據處理相關模塊NumPy:數組批量計算pandas:靈活的表計算Matplotlib:數據可視化 學習目標:用NumPy+pandas+Mat ...
說明:表示定義一組條件並確定指定對象是否符合這些條件的方法.此委托由 Array 和 List 類的幾種方法使用,用於在集合中搜索元素. var predicate = new Predicate&l ...
