在 Table Storage azure 中使用 Etag 通过多线程更新属性

Posted 2023-03-28

【中文标题】在 Table Storage azure 中使用 Etag 通过多线程更新属性

【英文标题】：update a property by multi-thread with Etag in Table Storage azure

【发布时间】：2021-11-08 00:47:58

【问题描述】：

概述： 当我在容器/productID(folder)/blobName 下的 blob 存储中上传 blob 时，事件订阅会将此事件保存在存储队列中。之后，天蓝色函数轮询此事件并执行以下操作：

1- 从对应表中读取当前计数属性（如何 许多 blob 存储在 productID(folder)) 下

2-增加计数+1

3-写回对应的表中

4-返回

问题是一个竞争条件，我尝试将其放入 Lock() 中，如代码所示。这在我同时上传 1000 个文件时有效。但是如果我同时加载 10000 个文件并读取计数属性，它返回超过 10000 个是错误的，它必须只返回 10000 个。我还阻止了横向扩展，只创建了一个实例。 问题仍然是竞争条件（我不这么认为，但可能是）还是 Azure 运行时在函数上运行的事件不止一个？我不确定发生了什么。任何想法都会很好

class _tableStorage:
    def __init__(self, account, key):
        self.table_service = TableService(account, key)

    def create_table(self, table_name):
        self.table_service.create_table(table_name)

    def insert_entity_table(self, table_name, entity):
        self.table_service.insert_or_replace_entity(table_name, entity)

    def exist_table(self, table_name):
        return self.table_service.exists(table_name)

    def get_entity_table(self, table_name, entity):
        return self.table_service.get_entity(
            table_name, entity.PartitionKey, entity.RowKey)

    def get_all_entities_table(self, table_name):
        try:
            list = self.table_service.query_entities(table_name)
        except:
            logging.info('unknown error by listing entities')
        return list


def get_blob_meta(url):
    parsed_url = urlparse.urlparse(url)
    return 
        "storage": parsed_url.netloc.split('.')[0],
        "contianer": parsed_url.path.split('/')[1],
        "folder": parsed_url.path.split('/')[2]
    


threadLock = threading.Lock()


def main(msg: func.QueueMessage) -> None:
    url = json.loads(msg.get_body().decode(
        'utf-8').replace("'", "\""))['data']['url']
    logging.info(url)
    blob_meta = get_blob_meta(url)
    logging.info(blob_meta)
    table_service = _tableStorage(
        blob_meta['storage'],
        "xxxxxxxxxx")

    threadLock.acquire()
    entity = Entity()
    # should have same partition to be stored in one node.
    entity.PartitionKey = blob_meta['contianer']
    entity.RowKey = blob_meta['folder']
    if(not table_service.exist_table(blob_meta['contianer'])):
        table_service.create_table(blob_meta['contianer'])
        entity.count = 1
    else:
        entity.count = table_service.get_entity_table(
            blob_meta['contianer'], entity).count + 1

    table_service.insert_entity_table(blob_meta['contianer'], entity)
    threadLock.release()

【参考方案1】：

两种解决方案：

第一个多线程：

idea 是头部中的 ETag 标志，以确保原子处理。起初，我阅读了 count 属性和 ETag 标志。然后，我增加它。在我用表中增加的一个更新计数属性之前，if_match 会将我的 Etag 与表中存储的一个匹配，如果 ETag 匹配，那么计数将被更新，否则，它会抛出一个错误，我捕获这个错误并再次尝试读取和递增，直到更新成功 To understand more read the docu

 header_etag = "random-etag"
    response_etag = "random-response"
    while True:
        sleep(random.random())  # sleep between 0 and 1 second.
        header = table_service1.get_entity(
            client_table, client_table, client_product)
        header_etag = header['etag']
        new_count = header['Count'] + 1
        entity_product = create_product_entity(
            client_table, client_product, new_count, client_image_table)
        try:
            response_etag = table_service1.merge_entity(client_table, entity_product,
                                                        if_match=header_etag)
            break
        except:
            logging.info("race condition detected")

第二个

通过防止多线程解决：

本地调试在 local.settings.json

  "IsEncrypted": false,
  "Values": 

    "AzureFunctionsJobHost__extensions__queues__batchSize": 1,
    "AzureFunctionsJobHost__extensions__queues__newBatchThreshold": 0,

  

**在生产中** host.json

 "extensions": 
        "queues": 
            "batchSize": 1,
            "newBatchThreshold": 0
        
    

To understand pls see the documentation of microsoft