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

import requests
class OrderService:    def __init__(self):        self.session = requests.Session()
    def create_order(self, order_data):        try:            # Step 1: Reserve stock            res = self.session.post('http://inventory-service/reserve-stock', json=order_data)            res.raise_for_status()
            # Step 2: Charge payment            res = self.session.post('http://payment-service/charge-payment', json=order_data)            res.raise_for_status()
            # Step 3: Confirm order            res = self.session.post('http://order-service/confirm-order', json=order_data)            res.raise_for_status()        except requests.exceptions.RequestException as e:            # Step 4: Cancel order            self.session.post('http://order-service/cancel-order', json=order_data)            self.session.post('http://payment-service/refund-payment', json=order_data)            self.session.post('http://inventory-service/release-stock', json=order_data)            raise e

import psycopg2from psycopg2 import sqlfrom psycopg2.extras import DictCursorimport jsonimport threadingimport time
# Define the message schemamessage_schema = {    "id": "",    "data": {},    "status": ""}
# Connect to the PostgreSQL databaseconn = psycopg2.connect(    dbname="my_database",    user="my_username",    password="my_password",    host="localhost",    port="5432")
# Create the messages table if it doesn't existwith conn.cursor() as cur:    cur.execute("""        CREATE TABLE IF NOT EXISTS messages (            id UUID PRIMARY KEY,            data JSONB NOT NULL,            status TEXT NOT NULL        )    """)    conn.commit()
# Define the message producerdef send_message(data):    with conn.cursor() as cur:        # Create the message using the schema        message = message_schema.copy()        message["id"] = str(uuid.uuid4())        message["data"] = data        message["status"] = "new"                # Insert the message into the messages table        cur.execute("""            INSERT INTO messages (id, data, status)            VALUES (%s, %s, %s)        """, (message["id"], json.dumps(message["data"]), message["status"]))        conn.commit()
# Define the message consumerdef message_consumer():    while True:        with conn.cursor(cursor_factory=DictCursor) as cur:            # Select the oldest message from the messages table that has a status of "new"            cur.execute("""                SELECT *                FROM messages                WHERE status = 'new'                ORDER BY id                LIMIT 1                FOR UPDATE SKIP LOCKED            """)            row = cur.fetchone()            if row:                message = dict(row)                # Update the message status to "processing"                cur.execute("""                    UPDATE messages                    SET status = 'processing'                    WHERE id = %s                """, (message["id"],))                conn.commit()                # Process the message                process_message(message)                # Update the message status to "processed"                cur.execute("""                    UPDATE messages                    SET status = 'processed'                    WHERE id = %s                """, (message["id"],))                conn.commit()        time.sleep(1)
# Start the message consumer in a separate threadconsumer_thread = threading.Thread(target=message_consumer)consumer_thread.start()

import jsonimport pikafrom sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, DateTime
# Create SQLAlchemy engine and metadataengine = create_engine('postgresql://user:password@localhost:5432/db')metadata = MetaData()
# Define the Outbox table schemaoutbox_table = Table('outbox', metadata,                     Column('id', Integer, primary_key=True),                     Column('event_type', String),                     Column('payload', String),                     Column('created_at', DateTime)                    )
# Define RabbitMQ connection parametersrabbitmq_params = pika.ConnectionParameters(host='localhost', port=5672)
# Define RabbitMQ queue namequeue_name = 'outbox'
# Connect to RabbitMQconnection = pika.BlockingConnection(rabbitmq_params)channel = connection.channel()
# Declare the queuechannel.queue_declare(queue=queue_name, durable=True)
# Define function to publish messages to RabbitMQdef publish_message(message):    channel.basic_publish(exchange='',                          routing_key=queue_name,                          body=message,                          properties=pika.BasicProperties(                              delivery_mode=2,  # Make messages persistent                          ))
# Define function to read messages from the Outbox table and publish them to RabbitMQdef process_outbox():    with engine.connect() as conn:        result = conn.execute(outbox_table.select().order_by(outbox_table.c.id))        for row in result:            # Create message payload as a dictionary            payload = {                'event_type': row['event_type'],                'data': json.loads(row['payload'])            }            # Convert payload to JSON string            message = json.dumps(payload)            # Publish message to RabbitMQ            publish_message(message)            # Delete message from Outbox table            conn.execute(outbox_table.delete().where(outbox_table.c.id == row['id']))
# Define function to write data to the Outbox tabledef write_to_outbox(event_type, data):    with engine.connect() as conn:        # Convert data to JSON string        payload = json.dumps(data)        # Insert data into Outbox table        conn.execute(outbox_table.insert().values(            event_type=event_type,            payload=payload,            created_at=datetime.now()        ))
# Example usage:write_to_outbox('user_created', {'user_id': 123, 'name': 'John Doe'})process_outbox()

import stomp
class MyListener(stomp.ConnectionListener):
    def __init__(self, conn):        self.conn = conn
    def on_message(self, headers, message):        # Process the message        print("Received message:", message)        # Acknowledge the message        self.conn.ack(id=headers['message-id'], subscription=headers['subscription'])
    def on_error(self, headers, message):        print('Received an error "{}": {}'.format(headers, message))        def on_disconnected(self):        print('Disconnected from the message broker.')
# Define the message broker connection settingsbroker_url = "tcp://localhost:61613"username = "myusername"password = "mypassword"queue_name = "/queue/myqueue"
# Establish a connection to the message brokerconn = stomp.Connection([(broker_url, 61613)])conn.set_listener('', MyListener(conn))conn.start()conn.connect(username, password)
try:    # Begin the transaction    conn.begin()    # Send the first message    conn.send(body="Message 1", destination=queue_name)    # Send the second message    conn.send(body="Message 2", destination=queue_name)    # Commit the transaction    conn.commit()except Exception as ex:    # Rollback the transaction in case of errors    conn.rollback()    print("Error sending messages:", str(ex))
# Disconnect from the message brokerconn.disconnect()