Pythonとのダイナモにおける楽観的ロッキングの実装

データベース内の同じデータへの同時書き込みアクセスは、一貫性の問題につながる可能性があります.これはDynamoDBのデータについても同様です.これを解決するためのパターンがいくつかあります.このポストでは、Pythonで楽観的なロックを実行することを見ていきます.最初に我々は問題を調査して、解決策を議論して、その後、実装をチェックします.

問題

DynamoDBは、アプリケーションのデータについての唯一の真実のソースです.別の部分では、アプリケーションのさまざまな部分に書き込み、それからお読みになります.これは複数のクライアントが同時に複数のクライアントを行うという中央の解決策であるため、競合が発生する可能性があります.
例を見てみましょう.ここでは、我々は、顧客のアカウント上で現在の残高を追跡し、どのように多くのオーバードラフトが許可されている銀行のテーブルを持っている.私たちの口座123は100の現在のバランスを持って、更なる取引がブロックされるまで、- 500まで下がることができます.
会計士
バランス
オーバードラフト限界
123
100
- 500
...
...
...
今、このアカウントの残高を減少させる2つの並列トランザクションがあると仮定します.最初は400と2番目の300を取る.これらを順次処理すると、最初のトランザクションの後、100 - 400 = - 300のバランスで到着します.次に、第2のトランザクションを処理して他の300を引く.これはオーバードラフト限界(−300−300＝−600＜−500)の下で減少させる.その結果、2番目のトランザクションは失敗します.
すべてのトランザクションで動作する単一プロセスを持つことは、本質的に制限されています.したがって、トランザクションで並列に動作するように複数のプロセスを持つようにします.各ワーカープロセスは次の処理を行います.

テーブルから現在のレコードを読み込む

我々がオーバードラフト限界を越えないならば、トランザクションを記録に適用してください

更新されたレコードをデータベースに書き込む

ここでの問題点は、ステップ1、3の間に遅延があることである.この問題がどのように問題になるかを次の図に示します.

私たちの- 400と- 300トランザクションの個々の労働者プロセスによって処理されます.ワーカー1は小さなヘッドスタートを持ち、現在の残高を最初に読み込む.直後に労働者2も開始し、現在のバランスを読み取ります.現在、労働者の各々は、彼らが現在のバランスであると思うものに、取引を適用します.両方の個々のケースでは、支出限度は破られません.ワーカー1は少し前に始まったので、以前に終了し、更新された残高をデータベースに戻します.労働者2は密接に遅れていますが、これに気づかず、データベースに更新されたバランスを書きます.
これは2つの深刻な問題があります:バランスは間違っています、そして、取引の1つは可能であったべきではありません.これは、時々失われた更新問題として参照されます.
これらの問題があるシステムは壊れていると考えられることができます、そして、実際には、銀行はそのように作動しません.では、リソースへのアクセスの競合に対処する方法を見てみましょう.

解決策

リソースへの同時衝突アクセスの処理は、通常、ロックと呼ばれるプロセスを通して処理される.これに無数の異なる品種とニュアンスがありますが、タイトルとして我々は特定のフォームに焦点を当てる示唆している.楽観的で悲観的なロックの2つの基本的なアプローチがあります.
悲観的なロックは、あなたがそれに取り組む前にリソースにロックを置くことを意味します.このロックはリソース自体によって強制され、ロックを解放するまでリソースへの排他的なアクセスを許可します.他の人がリソースにアクセスしたいなら、彼らはロックアウトされるかもしれないか、実装に応じて読み込み専用のアクセスに制限されるかもしれません.悲観的なロックは、高価なときに操作を再試行するか、または多くの同時アクセスを期待して使用されます.
他の手で楽観的なロックは別のアプローチを好む.根底にある仮定は、ほとんどの時間がリソースへのアクセスへのアクセスがないということです、したがって、関連したパフォーマンスオーバーヘッドで排他的なロックがありません.代わりに、この手法は、発生したときに競合を検出し、それらを優雅に扱うことができます.
アイデアは、テーブル内の各項目は、各更新時にインクリメントされている番号のバージョン属性を取得/そのアイテムに置くことです.あなたが最初にアイテムを読むとき、あなたがそれを読んだとき、あなたは、それが持っていたバージョンのメモを取りますcurrent_version ). その後、ローカルに項目を処理し、それに変更を加えることができます.変更を継続する前に、項目のバージョン番号を増やす必要があります.
重要な部分は、あなたがDynamoDBにアイテムを書くとき、来ます.定期的に使えますPutItem or UpdateItem を呼び出しますが、ConditionExpression これは、バージョン属性の値がcurrent_version . これは、あなたが最初にそれを読んだとき、それが同じ状態にあるならば、あなたがアイテムを単に更新するのを確実にします.条件が満たされていない場合、UPDATE/PATはキャンセルされ、例外が送出されます.その後、再度項目を読み取ることができますし、処理をやり直し、それをテーブルに書き込むことを試みる.
これは、衝突がまれで再処理が安いシナリオを処理する効果的な方法です.例えば、銀行のユースケースからの労働者プロセスがどのように実施されるかについての例を見てみましょう.

実装

以下に注釈付きのPythonスクリプトを見つけます.元の読み取り、適用および書き込み手順をわずかに調整されている.読み取りステップでは、我々もcurrent_version アイテムの.それを書く前に、我々はバージョン番号を1で増やして、それから条件式を加えますput_item コール.状態チェックの例外処理は少し不明瞭に見えますが、残念ながらBOTO 3は“かなり”例外を公開しません.

import boto3
from boto3.dynamodb.conditions import Attr

# Step 0 Init

table = boto3.resource("dynamodb").Table("accounts")
transaction_value = -400 # This would come from somewhere else

# Step 1 Read the current item
item = table.get_item(Key={"AccountId": "123"})['Item']

# This is how item looks like:
# {
#   "AccountId": "123",
#   "Balance": 100,
#   "OverdraftLimit": -500,
#   "Version": 0,
# }

current_version = item["Version"]

# Step 2 Apply transaction
if item["Balance"] + transaction_value >= item["OverdraftLimit"]:
    item["Balance"] += transaction_value
else:
    raise ValueError("Overdraft limit breached!")

# Step 3 Write

# 3.1 Increase the version number so other workers know something changed
item["Version"] += 1

# 3.2 Try to write the item, but only if it hasn't been updated in the mean time
try:
    table.put_item(
        Item=item,
        ConditionExpression=Attr("Version").eq(current_version)
    )
except ClientError as err:
    if err.response["Error"]["Code"] == 'ConditionalCheckFailedException':
        # Somebody changed the item in the db while we were changing it!
        raise ValueError("Balance updated since read, retry!") from err
    else:
        raise err

概要

この記事では、複数のリソースから同時にDynamODBの同じ項目にアクセスするときに発生する問題を調べました.楽観的なロックは、これらの同時アクセスを検出して、扱う1つの技術です.また、概念の単純な実装がPythonでどのように見えるかを見ました.
このブログ記事を読んでくれてありがとう.ご質問、懸念やフィードバックがある場合は、躊躇しないでください.
-モリス