Hey there! Let's break down your problem step by step. First off, you absolutely don't need RabbitMQ here—your use case is a single IoT device needing local message caching, so a lightweight, persistent local queue is way more appropriate. Spinning up RabbitMQ would just add unnecessary resource overhead to your device, which is overkill.

核心实现思路 #

We can solve this with two key components:

• Persistent local storage: Use a lightweight file-based database (like SQLite) to cache unsent messages, so they don't get lost if the device restarts.
• Asynchronous retry thread: Run a background thread that periodically checks the cache and tries to send messages until they succeed, using a 5-second interval as you suggested.
• Ordered delivery: Keep track of message timestamps to ensure we send them in the order they were generated.

Pythonic代码实现 #

Here's a modified version of your code that incorporates these ideas, with proper error handling and persistence:

import sqlite3
import json
import time
import threading
import boto3
from botocore.exceptions import ClientError

# Initialize SQS client (fill in your region and queue URL)
sqs = boto3.client('sqs', region_name='your-region')
QUEUE_URL = 'your-sqs-queue-url'

class LocalMessageCache:
    def __init__(self, db_path='message_cache.db'):
        self.db_path = db_path
        self._setup_database()

    def _setup_database(self):
        """Create the message cache table if it doesn't exist"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS messages (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    body TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    is_sent INTEGER DEFAULT 0
                )
            ''')
            conn.commit()

    def add_unsent_message(self, message_body):
        """Store a failed message in the cache"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('INSERT INTO messages (body) VALUES (?)', (message_body,))
            conn.commit()

    def get_unsent_messages(self):
        """Fetch all unsent messages, ordered by creation time"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('SELECT id, body FROM messages WHERE is_sent = 0 ORDER BY created_at ASC')
            return [dict(row) for row in cursor.fetchall()]

    def mark_message_sent(self, message_id):
        """Mark a message as successfully delivered"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('UPDATE messages SET is_sent = 1 WHERE id = ?', (message_id,))
            conn.commit()

# Initialize our local cache
message_cache = LocalMessageCache()

def _send_to_sqs(message_body):
    """Internal function to handle SQS delivery and error checking"""
    try:
        sqs.send_message(
            QueueUrl=QUEUE_URL,
            DelaySeconds=1,
            MessageAttributes={},
            MessageBody=message_body
        )
        return True
    except ClientError as e:
        error_code = e.response['Error']['Code']
        # Only cache messages if it's a network-related error
        if error_code in ['NetworkError', 'ConnectionError']:
            print(f"Network issue encountered: {e}")
            return False
        else:
            # For SQS-specific errors (like invalid queue), no need to retry
            print(f"SQS service error, not caching: {e}")
            return True
    except Exception as e:
        print(f"Unexpected error, caching message: {e}")
        return False

def send_message(function, userid):
    """Your public send message interface, now with caching"""
    msg_json = json.dumps({"function": function, "userid": userid})
    # Try sending immediately first
    if not _send_to_sqs(msg_json):
        message_cache.add_unsent_message(msg_json)

def retry_worker():
    """Background thread that retries unsent messages every 5 seconds"""
    while True:
        unsent_messages = message_cache.get_unsent_messages()
        if unsent_messages:
            print(f"Found {len(unsent_messages)} unsent messages, starting retry...")
            for msg in unsent_messages:
                if _send_to_sqs(msg['body']):
                    message_cache.mark_message_sent(msg['id'])
                    print(f"Successfully sent message ID {msg['id']}")
        time.sleep(5)

# Start the retry thread as a daemon (it will exit when the main program does)
threading.Thread(target=retry_worker, daemon=True).start()

Key Details & Optimizations #

• Persistence: SQLite ensures messages survive device restarts, which is critical for IoT devices that might power cycle unexpectedly.
• Smart Error Handling: We distinguish between network errors (which warrant caching) and SQS service errors (which don't, since retrying won't fix them).
• Ordered Delivery: Messages are sent in the order they were created, preserving the original sequence of events.
• Daemon Thread: The retry thread runs in the background and automatically exits when your main program stops, so you don't have to manage thread cleanup manually.

Extra Tips for Production #

• Exponential Backoff: Instead of a fixed 5-second interval, use exponential backoff (e.g., 5s → 10s → 20s → 30s) to avoid hammering the network when it's unstable.
• Message Expiry: Add an expiry timestamp to cached messages, so you don't retry messages that are too old (e.g., older than 7 days).
• Logging: Replace print statements with proper logging to a file, so you can debug issues later.
• Thread Safety: If multiple parts of your code send messages concurrently, add a lock around cache operations to prevent race conditions.

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