单AWS EC2负载均衡架构切换为Pub-Sub架构的技术咨询

Got it, let's break down how to shift your current EC2-based load balancing setup to a pub-sub architecture—here's a practical, step-by-step approach tailored to your AWS environment:

First, let's reimagine the flow: instead of your frontend EC2 directly picking a backend and making an HTTP call, it'll publish client requests to a message queue. Your backend worker servers will subscribe to this queue, pull requests as they come in, process them, then publish the results to a separate response queue. Finally, your frontend will listen for responses matching the original request and send them back to the client. This decouples your frontend from the backend entirely—no more hardcoding backend server addresses or managing load balancing logic manually.

Since you're already using AWS, stick to native services for seamless integration and minimal overhead:

• Amazon SQS (Simple Queue Service)：Perfect for this use case. It's a managed message queue that handles delivery guarantees, retries, and scaling out of the box. Use standard queues for high throughput, or FIFO queues if you need strict request ordering.
• Amazon SNS (Simple Notification Service)：If you ever need to broadcast requests to multiple backend groups in the future, SNS can publish to multiple SQS queues at once. But for your current "one request → one worker" flow, SQS alone is simpler.

3.1 Set Up Your Queues #

First, create two SQS queues in your AWS account:

• client-request-queue: Where your frontend will drop incoming client requests.
• backend-response-queue: Where workers will send back processed results.
• Optional: Add dead-letter queues (DLQs) to both. These catch messages that fail processing multiple times, so you can debug them without clogging your main queues.

3.2 Rewrite Your Frontend EC2 Logic #

Replace the "pick backend + make HTTP call" code with logic to publish requests and listen for responses. Here's a simplified Python example using boto3:

import boto3
import uuid
import json
from time import sleep

# Initialize SQS client
sqs = boto3.client('sqs')
REQUEST_QUEUE_URL = "your-request-queue-url"
RESPONSE_QUEUE_URL = "your-response-queue-url"

def handle_client_http_request(client_req):
    # Generate a unique ID to track this request
    request_id = str(uuid.uuid4())
    
    # Package request data + metadata (so workers know how to process it)
    message_payload = {
        "request_id": request_id,
        "http_method": client_req.method,
        "path": client_req.path,
        "body": client_req.get_json()
    }
    
    # Publish to request queue
    sqs.send_message(
        QueueUrl=REQUEST_QUEUE_URL,
        MessageBody=json.dumps(message_payload),
        MessageAttributes={
            "RequestID": {"StringValue": request_id, "DataType": "String"}
        }
    )
    
    # Poll response queue for matching result (use long polling to reduce empty checks)
    while True:
        response = sqs.receive_message(
            QueueUrl=RESPONSE_QUEUE_URL,
            MessageAttributeNames=["RequestID"],
            MaxNumberOfMessages=1,
            WaitTimeSeconds=20  # Long polling: wait up to 20s for a message
        )
        
        if "Messages" in response:
            for msg in response["Messages"]:
                # Check if this response matches our request ID
                if msg["MessageAttributes"]["RequestID"]["StringValue"] == request_id:
                    # Parse and return the result to the client
                    processed_result = json.loads(msg["Body"])
                    # Clean up: delete the response from the queue
                    sqs.delete_message(
                        QueueUrl=RESPONSE_QUEUE_URL,
                        ReceiptHandle=msg["ReceiptHandle"]
                    )
                    return processed_result
        sleep(1)

3.3 Update Backend Worker Servers #

Instead of waiting for HTTP requests, your workers will run a loop that pulls messages from the request queue, processes them, and sends results to the response queue. Example:

import boto3
import json

sqs = boto3.client('sqs')
REQUEST_QUEUE_URL = "your-request-queue-url"
RESPONSE_QUEUE_URL = "your-response-queue-url"

def process_client_request(request_data):
    # Your existing backend processing logic goes here
    # e.g., query a database, run computations, etc.
    return {"status": "success", "data": f"Processed request for {request_data['path']}"}

def worker_loop():
    while True:
        # Pull messages from the request queue (long polling)
        response = sqs.receive_message(
            QueueUrl=REQUEST_QUEUE_URL,
            MessageAttributeNames=["RequestID"],
            MaxNumberOfMessages=1,
            WaitTimeSeconds=20
        )
        
        if "Messages" in response:
            for msg in response["Messages"]:
                request_id = msg["MessageAttributes"]["RequestID"]["StringValue"]
                request_payload = json.loads(msg["Body"])
                
                # Process the request
                result = process_client_request(request_payload)
                
                # Send result to response queue
                sqs.send_message(
                    QueueUrl=RESPONSE_QUEUE_URL,
                    MessageBody=json.dumps(result),
                    MessageAttributes={
                        "RequestID": {"StringValue": request_id, "DataType": "String"}
                    }
                )
                
                # Delete the processed request from the queue (so it doesn't get retried)
                sqs.delete_message(
                    QueueUrl=REQUEST_QUEUE_URL,
                    ReceiptHandle=msg["ReceiptHandle"]
                )

3.4 Add Scaling & Reliability Guards #

• Auto-Scale Workers: Use EC2 Auto Scaling Groups tied to CloudWatch metrics (like ApproximateNumberOfMessagesVisible in your request queue). This lets AWS automatically spin up more workers when traffic spikes, and shut them down when it drops.
• Idempotent Processing: Make sure your backend logic can handle the same request multiple times (SQS will retry messages if they aren't deleted promptly). For example, use the request ID to skip duplicate processing.
• Timeout Handling: Add timeouts in your frontend code to avoid waiting indefinitely for a response. For workers, set a maximum processing time and fail requests that take too long (send them to the DLQ for later debugging).

• Incremental Rollout: Start by routing 10% of traffic to the pub-sub flow, keeping the rest on your old setup. Monitor performance and error rates before going full-in.
• Monitor Everything: Use CloudWatch to track queue lengths, processing times, and error rates. Use AWS X-Ray to trace requests from frontend to worker to response—this helps spot bottlenecks fast.

内容的提问来源于stack exchange，提问作者MayoMan