分布式事务的 21 种武器 - 3

import pikaimport json
# Define the RabbitMQ connection parameterscredentials = pika.PlainCredentials('guest', 'guest')parameters = pika.ConnectionParameters('localhost', 5672, '/', credentials)
# Define the messages to be sent between servicesstart_order_message = {'order_id': '12345', 'items': [{'id': '1', 'name': 'item1'}, {'id': '2', 'name': 'item2'}]}payment_message = {'order_id': '12345', 'amount': 100.0}shipping_message = {'order_id': '12345', 'items': [{'id': '1', 'name': 'item1'}, {'id': '2', 'name': 'item2'}]}
# Define the compensation messages to be sent in case of failurecancel_payment_message = {'order_id': '12345', 'amount': 100.0}cancel_shipping_message = {'order_id': '12345', 'items': [{'id': '1', 'name': 'item1'}, {'id': '2', 'name': 'item2'}]}
# Define the function to send messages to the RabbitMQ brokerdef send_message(queue_name, message):    connection = pika.BlockingConnection(parameters)    channel = connection.channel()    channel.queue_declare(queue=queue_name)    channel.basic_publish(exchange='', routing_key=queue_name, body=json.dumps(message))    connection.close()
# Define the function to handle the start of the orderdef start_order():    # Send the start order message to the Order service    send_message('start_order', start_order_message)
# Define the function to handle the payment of the orderdef payment():    try:        # Send the payment message to the Payment service        send_message('payment', payment_message)    except Exception as e:        # Send the cancel payment message to the Payment service in case of failure        send_message('cancel_payment', cancel_payment_message)        raise e
# Define the function to handle the shipping of the orderdef shipping():    try:        # Send the shipping message to the Shipping service        send_message('shipping', shipping_message)    except Exception as e:        # Send the cancel shipping message to the Shipping service in case of failure        send_message('cancel_shipping', cancel_shipping_message)        raise e
# Define the function to handle the cancellation of the orderdef cancel_order():    # Send the cancel payment message to the Payment service    send_message('cancel_payment', cancel_payment_message)
    # Send the cancel shipping message to the Shipping service    send_message('cancel_shipping', cancel_shipping_message)
# Define the main function to execute the Sagadef execute_saga():    try:        # Start the order        start_order()
        # Perform the payment        payment()
        # Perform the shipping        shipping()    except Exception as e:        # Cancel the order in case of failure        cancel_order()        raise e
# Call the main function to execute the Sagaexecute_saga()

import uuidimport jsonimport time
class BankAccount:    def __init__(self):        self.balance = 0        self.event_sourcing = EventSourcing()
    def deposit(self, amount):        event = Event('deposit', {'amount': amount})        self.event_sourcing.add_event(event)        self.balance += amount
    def withdraw(self, amount):        if self.balance < amount:            raise ValueError('Insufficient balance')        event = Event('withdraw', {'amount': amount})        self.event_sourcing.add_event(event)        self.balance -= amount
    def get_balance(self):        return self.balance
    def get_events(self):        return self.event_sourcing.get_events()
    def get_event_by_id(self, event_id):        return self.event_sourcing.get_event_by_id(event_id)
    def replay_events(self):        self.balance = 0        for event in self.event_sourcing.get_events():            if event.type == 'deposit':                self.balance += event.data['amount']            elif event.type == 'withdraw':                self.balance -= event.data['amount']
class Event:    def __init__(self, type, data):        self.id = uuid.uuid4()        self.timestamp = int(time.time())        self.type = type        self.data = data
class EventSourcing:    def __init__(self):        self.event_store = EventStore()
    def add_event(self, event):        self.event_store.store_event(event)
    def get_events(self):        return self.event_store.get_events()
    def get_event_by_id(self, event_id):        return self.event_store.get_event_by_id(event_id)
class EventStore:    def __init__(self):        self.events = []
    def store_event(self, event):        self.events.append(event)
    def get_events(self):        return self.events
    def get_event_by_id(self, event_id):        for event in self.events:            if event.id == event_id:                return event        raise ValueError('Event not found')
class Auditor:    def __init__(self, event_store):        self.event_store = event_store
    def log_events(self):        for event in self.event_store.get_events():            print(json.dumps({'id': str(event.id), 'type': event.type, 'data': event.data}))
account = BankAccount()auditor = Auditor(account.event_sourcing.event_store)
account.deposit(100)account.withdraw(50)account.deposit(75)
print('Current balance:', account.get_balance())
print('All events:')auditor.log_events()
event_id = account.get_events()[1].idevent = account.get_event_by_id(event_id)print('Event details:', json.dumps({'id': str(event.id), 'type': event.type, 'data': event.data}))

from abc import ABC, abstractmethodfrom typing import Any, Dict, Optional
class Command(ABC):    pass
class CreateProductCommand(Command):    def __init__(self, name: str, price: float):        self.name = name        self.price = price
class UpdateProductCommand(Command):    def __init__(self, product_id: str, name: Optional[str] = None, price: Optional[float] = None):        self.product_id = product_id        self.name = name        self.price = price
class DeleteProductCommand(Command):    def __init__(self, product_id: str):        self.product_id = product_id
class Query(ABC):    pass
class GetProductQuery(Query):    def __init__(self, product_id: str):        self.product_id = product_id
class GetAllProductsQuery(Query):    pass
class Product:    def __init__(self, id: str, name: str, price: float):        self.id = id        self.name = name        self.price = price
class ProductRepository:    def __init__(self):        self.products = []
    def create(self, name: str, price: float) -> Product:        product = Product(str(len(self.products) + 1), name, price)        self.products.append(product)        return product
    def get(self, id: str) -> Optional[Product]:        for product in self.products:            if product.id == id:                return product        return None
    def get_all(self) -> List[Product]:        return self.products
    def update(self, id: str, name: Optional[str] = None, price: Optional[float] = None) -> Optional[Product]:        for product in self.products:            if product.id == id:                if name is not None:                    product.name = name                if price is not None:                    product.price = price                return product        return None
    def delete(self, id: str) -> bool:        for product in self.products:            if product.id == id:                self.products.remove(product)                return True        return False
class ProductCommandHandler:    def __init__(self, repository: ProductRepository):        self.repository = repository
    def handle(self, command: Command) -> Optional[Product]:        if isinstance(command, CreateProductCommand):            return self.repository.create(command.name, command.price)        elif isinstance(command, UpdateProductCommand):            return self.repository.update(command.product_id, command.name, command.price)        elif isinstance(command, DeleteProductCommand):            success = self.repository.delete(command.product_id)            return success
class ProductQueryHandler:    def __init__(self, repository: ProductRepository):        self.repository = repository
    def handle(self, query: Query) -> Optional[Any]:        if isinstance(query, GetProductQuery):            return self.repository.get(query.product_id)        elif isinstance(query, GetAllProductsQuery):            return self.repository.get_all()
class ProductService:    def __init__(self, command_handler: ProductCommandHandler, query_handler: ProductQueryHandler):        self.command_handler = command_handler        self.query_handler = query_handler
    def create_product(self, name: str, price: float) -> Product:        command = CreateProductCommand(name, price)        return self.command_handler.handle(command)
    def get_product(self, id: str) -> Optional[Product]:        query = GetProductQuery(id)        return self.query_handler.handle(query)
    def get_all_products(self) -> List[Product]:        query = GetAllProductsQuery()        return self