redis与mysql的数据一致性问题（ 网络分区）

# Python代码示例 - 一致性哈希算法
from hashlib import sha1
class ConsistentHashing:
    def __init__(self, nodes, replicas=3):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            for i in range(replicas):
                key = self.hash(f"{node}-{i}")
                self.ring[key] = node
                self.sorted_keys.append(key)
        self.sorted_keys.sort()
    def get_node(self, key):
        if not self.ring:
            return None
        h = self.hash(key)
        index = self._bisect_right(h)
        return self.ring[self.sorted_keys[index % len(self.sorted_keys)]]
    def _bisect_right(self, h):
        keys = self.sorted_keys
        high = len(keys)
        low = 0
        while low < high:
            mid = (low + high) // 2
            midval = keys[mid]
            if midval > h:
                high = mid
            else:
                low = mid + 1
        return low
    def hash(self, key):
        return int(sha1(key.encode()).hexdigest(), 16)
nodes = ['redis1', 'redis2', 'redis3']
consistent_hashing = ConsistentHashing(nodes)
node_for_key = consistent_hashing.get_node('product:123:stock')
print(f"Key 'product:123:stock' maps to node: {node_for_key}")

# Python代码示例 - 使用两阶段提交（2PC）
import redis
import MySQLdb
from kafka import KafkaProducer
def distributed_transaction(product_id):
    redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
    mysql_conn = MySQLdb.connect(host='localhost', user='user', password='password', db='ecommerce')
    producer = KafkaProducer(bootstrap_servers='localhost:9092')
    try:
        # 开始MySQL事务
        mysql_conn.begin()
        # 更新MySQL中商品信息
        cursor = mysql_conn.cursor()
        cursor.execute(f'UPDATE products SET stock = stock - 1 WHERE id = {product_id}')
        # 提交MySQL事务
        mysql_conn.commit()
        # 发送消息到Kafka，触发Redis的更新
        producer.send('product_stock_updates', {'product_id': product_id, 'action': 'decrement'})
        producer.flush()
        print(f"Product {product_id} stock decremented in MySQL and Kafka message sent.")
    except Exception as e:
        # 发生异常，回滚MySQL事务
        mysql_conn.rollback()
        print(f"Error: {e}")
    finally:
        mysql_conn.close()
# 调用分布式事务函数
distributed_transaction(123)