Welcome back! In this lesson, we’ll expand on the concept of event-driven architectures by introducing multi-threaded and distributed systems. These architectures are essential when workflows and events span across multiple services or machines. We’ll explore how to model distributed systems in the DSL, implement concurrency, and manage synchronization and fault tolerance in large-scale systems like ERP/CRM platforms or multi-sided marketplaces.
Table of Contents
-
Introduction to Multi-Threaded and Distributed Systems
-
Extending the DSL for Distributed Workflows
-
Implementing Multi-Threading in Flask and Python
-
Distributed Task Queues with Celery and RabbitMQ
-
Synchronizing Events Across Services
-
Practical Example: Distributed Order Processing System
-
Conclusion and Next Steps
1. Introduction to Multi-Threaded and Distributed Systems
What is a Multi-Threaded System?
In a multi-threaded system, multiple threads of execution run in parallel within a single application. This allows for tasks to be performed concurrently, improving performance and responsiveness, especially in CPU-bound tasks (e.g., heavy computations) or I/O-bound tasks (e.g., API requests or database operations).
What is a Distributed System?
A distributed system consists of multiple machines (nodes) working together to achieve a common goal. Each machine may handle a portion of the workload, with communication happening between nodes to coordinate tasks. Distributed systems are common in large-scale applications like e-commerce platforms, where different services (e.g., payment processing, order management, inventory) need to interact in a scalable way.
Key Concepts in Distributed Systems:
-
Concurrency: Multiple operations happening at the same time.
-
Task Queues: Distributing tasks across multiple workers (e.g., Celery with RabbitMQ).
-
Fault Tolerance: Ensuring the system can recover from failures (e.g., if a node crashes).
-
Eventual Consistency: Ensuring the system reaches a consistent state over time, even if updates happen asynchronously.
2. Extending the DSL for Distributed Workflows
Modeling Distributed Workflows in the DSL
When dealing with multi-threaded or distributed workflows, we may want to specify which tasks or processes are performed concurrently and which ones are distributed across multiple services. Let’s extend our DSL to support the definition of distributed tasks.
Syntax for Distributed Workflows in the DSL:
workflow OrderProcessing {
state Pending {
transition approve -> Approved by Manager;
transition reject -> Rejected by Manager;
}
state Approved {
transition fulfill -> Fulfilled by FulfillmentSystem if inventoryIsAvailable();
concurrent task notifyWarehouse, notifyCustomer -> Completed;
}
state Completed {
event shipmentConfirmed -> Shipped by LogisticsSystem;
}
state Shipped {
// End of workflow
}
}
Key Additions:
-
Concurrent Task: Defines tasks that run concurrently. For example, notifyWarehouse and notifyCustomer are executed in parallel.
-
Distributed Execution: The tasks can be handled by different services (e.g., one service for warehouse notifications, another for customer notifications).
-
Eventual State Transition: Once all concurrent tasks are complete, the state transitions to the next state (Completed).
3. Implementing Multi-Threading in Flask and Python
Python provides native support for multi-threading, which allows you to run multiple tasks in parallel. In Flask applications, threading can be useful for handling I/O-bound tasks, such as making multiple API requests at the same time.
Step 1: Using Python’s threading Module
Let’s implement a simple example where multiple tasks (like notifying the warehouse and customer) are handled concurrently using Python’s threading module.
Example: Multi-Threaded Task Execution
import threading
def notify_warehouse(order_id):
print(f”Notifying warehouse about order {order_id}”)
Simulate a delay (e.g., an API call)
time.sleep(2)
print(f”Warehouse notified for order {order_id}”)
def notify_customer(order_id):
print(f”Notifying customer about order {order_id}”)
Simulate a delay (e.g., sending an email)
time.sleep(1)
print(f”Customer notified for order {order_id}”)
def handle_concurrent_tasks(order_id):
Create threads for each task
warehouse_thread = threading.Thread(target=notify_warehouse, args=(order_id,))
customer_thread = threading.Thread(target=notify_customer, args=(order_id,))
Start both threads
warehouse_thread.start()
customer_thread.start()
Wait for both threads to complete
warehouse_thread.join()
customer_thread.join()
print(f”All tasks completed for order {order_id}”)
Run the tasks concurrently for order 123
handle_concurrent_tasks(123)
Explanation:
-
threading.Thread: Creates a separate thread for each task (notify_warehouse and notify_customer).
-
start(): Begins executing the task in the background.
-
join(): Waits for the thread to complete before moving to the next task.
Step 2: Integrating Multi-Threading with Flask
In a Flask application, you can use multi-threading to handle tasks like sending notifications concurrently. Here’s how to integrate this into the generated Flask workflow code.
Example: Concurrent Tasks in Flask Workflow:
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
workflow = OrderProcessingWorkflow()
try:
workflow.fulfill(‘FulfillmentSystem’)
Run concurrent tasks (notify warehouse and customer)
handle_concurrent_tasks(order_id)
Update the order status in the database
order = Order.query.get(order_id)
order.state = ‘Completed’
db.session.commit()
return jsonify({‘status’: ‘success’, ‘new_state’: ‘Completed’}), 200
except ValueError as e:
return jsonify({‘status’: ‘error’, ‘message’: str(e)}), 400
4. Distributed Task Queues with Celery and RabbitMQ
Why Use Distributed Task Queues?
In a distributed system, tasks may be too large or time-consuming to handle in a single thread or service. For example, notifying different parts of the system about an event (e.g., an order being fulfilled) might require coordination between multiple services. This is where distributed task queues, such as Celery, come in.
Step 1: Setting Up Celery with RabbitMQ
Celery, paired with RabbitMQ or Redis, can distribute tasks across multiple workers and services. Each worker can process a task independently, and Celery ensures that tasks are distributed evenly across available workers.
First, install RabbitMQ and configure Celery to use it as the message broker:
pip install celery
Next, configure Celery in your Flask application to use RabbitMQ as the broker:
from celery import Celery
def make_celery(app):
celery = Celery(
app.import_name,
backend=app.config
,
broker=app.config
)
celery.conf.update(app.config)
return celery
app = Flask(name)
app.config.update(
CELERY_BROKER_URL=‘amqp://localhost’,
CELERY_RESULT_BACKEND=‘rpc://’
)
celery = make_celery(app)
Step 2: Distributing Tasks Across Workers
For tasks that need to be processed concurrently (e.g., notifying the warehouse and customer), we can use Celery’s task queue to distribute them across different workers.
Example: Defining Distributed Tasks with Celery:
@celery.task
def notify_warehouse(order_id):
print(f”Notifying warehouse about order {order_id}”)
Simulate a delay (e.g., an API call)
time.sleep(2)
print(f”Warehouse notified for order {order_id}”)
@celery.task
def notify_customer(order_id):
print(f”Notifying customer about order {order_id}”)
Simulate a delay (e.g., sending an email)
time.sleep(1)
print(f”Customer notified for order {order_id}”)
Now, instead of running the tasks in separate threads, you can distribute them across multiple workers:
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
workflow = OrderProcessingWorkflow()
try:
workflow.fulfill(‘FulfillmentSystem’)
Run distributed tasks using Celery
notify_warehouse.delay(order_id)
notify_customer.delay(order_id)
Update the order status in the database
order = Order.query.get(order_id)
order.state = ‘Completed’
db.session.commit()
return jsonify({‘status’: ‘success’, ‘new_state’: ‘Completed’}), 200
except ValueError as e:
return jsonify({‘status’: ‘error’, ‘message’: str(e)}), 400
By using Celery, tasks are distributed across multiple workers, which can scale horizontally by adding more worker nodes.
5. Synchronizing Events Across Services
Why Synchronization is Important in Distributed Systems
In a distributed system, where tasks and events are processed by multiple services, synchronization becomes critical. You must ensure that all parts of the system are updated consistently and avoid race conditions where multiple parts of the system try
to update the same data simultaneously.
Techniques for Synchronizing Events:
-
Message Brokers: Systems like RabbitMQ or Kafka can help synchronize event delivery across services, ensuring that no message is lost and that all parts of the system are notified.
-
Database Locks: Use database locking mechanisms to prevent race conditions during concurrent updates.
-
Eventual Consistency: Accept that different parts of the system may be temporarily out of sync but will eventually reach a consistent state.
Implementing Synchronization with RabbitMQ:
We can use RabbitMQ as a message broker to ensure that events like shipmentConfirmed are processed in the correct order and handled by the right services.
@app.route(‘/order/int:order_id/shipment-confirmed’, methods=
)
def shipment_confirmed(order_id):
Publish shipmentConfirmed event to RabbitMQ
channel.basic_publish(
exchange=”,
routing_key=‘shipment_queue’,
body=json.dumps({‘order_id’: order_id})
)
return jsonify({‘status’: ‘success’, ‘new_state’: ‘Shipped’}), 200
6. Practical Example: Distributed Order Processing System
Let’s implement a complete distributed order processing system using the concepts we’ve discussed: distributed tasks with Celery, synchronization with RabbitMQ, and concurrent task execution.
DSL for Distributed Order Workflow:
workflow OrderProcessing {
state Pending {
transition approve -> Approved by Manager;
transition reject -> Rejected by Manager;
}
state Approved {
concurrent task notifyWarehouse, notifyCustomer -> Completed;
}
state Completed {
event shipmentConfirmed -> Shipped by LogisticsSystem;
}
state Shipped {
// End of workflow
}
}
Generated Workflow Code with Distributed Tasks:
from celery import Celery
class OrderProcessingWorkflow:
states =
def init(self):
self.state = ‘Pending’
def approve(self, actor):
if self.state == ‘Pending’ and actor == ‘Manager’:
self.state = ‘Approved’
else:
raise ValueError(“Invalid transition or actor”)
def fulfill(self, actor):
if self.state == ‘Approved’ and actor == ‘FulfillmentSystem’:
notify_warehouse.delay(order_id)
notify_customer.delay(order_id)
self.state = ‘Completed’
else:
raise ValueError(“Invalid transition or actor”)
def shipment_confirmed(self, order_id):
handle_shipment_confirmation.delay(order_id)
self.state = ‘Shipped’
7. Conclusion and Next Steps
In this lesson, we’ve:
-
Extended the DSL to support multi-threaded and distributed workflows.
-
Implemented multi-threading in Flask using Python’s threading module.
-
Distributed tasks across multiple workers using Celery and RabbitMQ.
-
Implemented synchronization techniques to ensure consistency across services.
This approach provides a scalable solution for handling distributed workflows and asynchronous tasks in large systems like ERP/CRM platforms or multi-sided marketplaces.
Practical Exercise
Task: Extend the DSL to support distributed event listeners, where multiple services can listen to the same event and act independently. Implement this in Flask with RabbitMQ or Kafka.
Hints:
-
Use RabbitMQ or Kafka to broadcast events to multiple consumers.
-
Update the DSL syntax to define event listeners for multiple services.
Example DSL:
workflow OrderProcessing {
state Pending {
transition approve -> Approved by Manager;
event notifyWarehouse -> Completed by WarehouseService;
event notifyCustomer -> Completed by CustomerService;
}
}
Looking Ahead
In the next lesson, we will focus on fault tolerance and recovery in distributed systems, ensuring that workflows can recover gracefully from failures and maintain data integrity across services.
Feel free to reach out if you have any questions. See you in the next lesson!