双11分布式事务推荐

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='transaction_queue')

def callback(ch, method, properties, body):
    # 处理事务逻辑
    print(f"Received {body}")
    # 执行补偿操作如果失败

channel.basic_consume(queue='transaction_queue', on_message_callback=callback, auto_ack=True)

print('Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def acquire_lock(lock_name, acquire_timeout=10):
    identifier = str(uuid.uuid4())
    end = time.time() + acquire_timeout
    while time.time() < end:
        if r.setnx(lock_name, identifier):
            return identifier
        time.sleep(0.001)
    return False

def release_lock(lock_name, identifier):
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(lock_name)
                if pipe.get(lock_name) == identifier:
                    pipe.multi()
                    pipe.delete(lock_name)
                    pipe.execute()
                    return True
                pipe.unwatch()
                break
            except redis.WatchError:
                pass
    return False