如何实现命令模式的线程安全？多进程共享ArrayList存审计日志场景

Great question! Let's break down how to fix the thread-safety issue in your audit logging system using the Command Pattern, addressing the shared ArrayList problem and ensuring safe concurrent access across threads/processes.

Core Idea #

The Command Pattern helps us encapsulate each audit log entry as an immutable command object. Combined with thread-safe data structures and proper separation of concerns, we eliminate race conditions from shared mutable state.

Step 1: Define an Immutable Audit Command Interface #

First, create an abstract command that encapsulates the audit action and data. Make sure command objects are immutable (no setters, all fields final) so they're inherently thread-safe.

public interface AuditCommand {
    // Execute any pre-processing logic for the audit entry
    void execute();
    // Retrieve the finalized audit data for DB persistence
    AuditData getAuditData();
}

Step 2: Implement Concrete Audit Commands #

Create specific command implementations for different audit events. Since these objects are immutable, multiple threads can access them without risk of concurrent modification.

public class UserActionAuditCommand implements AuditCommand {
    private final AuditData auditData;

    // Initialize all audit data in the constructor (no changes after creation)
    public UserActionAuditCommand(String userId, String action, LocalDateTime timestamp) {
        this.auditData = new AuditData(userId, action, timestamp);
    }

    @Override
    public void execute() {
        // Optional: Validate data format, generate metadata, etc.
        // Since auditData is immutable, this logic is thread-safe
    }

    @Override
    public AuditData getAuditData() {
        return auditData;
    }
}

Step 3: Replace Shared ArrayList with a Thread-Safe Command Queue #

Instead of using a non-thread-safe ArrayList, use a concurrent queue like ConcurrentLinkedQueue (for unbounded queues) or ArrayBlockingQueue (for bounded queues). These collections handle thread-safe adds/removes without manual locking.

public class AuditCommandExecutor {
    // Thread-safe queue to hold audit commands
    private final Queue<AuditCommand> auditCommandQueue = new ConcurrentLinkedQueue<>();
    // Receiver responsible for DB persistence logic
    private final AuditDbPersistenceReceiver dbReceiver;

    public AuditCommandExecutor(AuditDbPersistenceReceiver dbReceiver) {
        this.dbReceiver = dbReceiver;
    }

    // Log an audit event by adding a command to the queue
    public void logAudit(AuditCommand command) {
        command.execute(); // Run pre-processing (safe, since command is immutable)
        auditCommandQueue.add(command); // Thread-safe add operation
    }

    // Called when a process/thread ends to persist all pending audit logs
    public void run() {
        // Batch fetch all pending commands to minimize DB roundtrips
        List<AuditCommand> pendingCommands = new ArrayList<>();
        AuditCommand cmd;
        while ((cmd = auditCommandQueue.poll()) != null) {
            pendingCommands.add(cmd);
        }

        if (!pendingCommands.isEmpty()) {
            // Convert commands to DB-ready data
            List<AuditData> auditDataList = pendingCommands.stream()
                    .map(AuditCommand::getAuditData)
                    .collect(Collectors.toList());
            // Delegate to receiver for batch DB save
            dbReceiver.batchSave(auditDataList);
        }
    }
}

Step 4: Implement the DB Persistence Receiver #

The receiver handles the actual database operations. Ensure batch saves are thread-safe (use connection pools, proper transaction isolation, etc.).

public class AuditDbPersistenceReceiver {
    public void batchSave(List<AuditData> auditDataList) {
        // Use a thread-safe DB connection pool for concurrent access
        try (Connection conn = DatabaseConnectionPool.getConnection()) {
            conn.setAutoCommit(false);
            // Execute bulk insert logic (e.g., JDBC batch updates)
            String sql = "INSERT INTO audit_logs (user_id, action, timestamp) VALUES (?, ?, ?)";
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                for (AuditData data : auditDataList) {
                    stmt.setString(1, data.getUserId());
                    stmt.setString(2, data.getAction());
                    stmt.setTimestamp(3, Timestamp.valueOf(data.getTimestamp()));
                    stmt.addBatch();
                }
                stmt.executeBatch();
            }
            conn.commit();
        } catch (SQLException e) {
            // Handle exceptions (rollback, logging, etc.)
            e.printStackTrace();
        }
    }
}

Key Thread-Safety Guarantees #

• Immutable Command Objects: No concurrent modification of audit data since commands can't be changed after creation.
• Thread-Safe Queue: ConcurrentLinkedQueue handles atomic adds/removes, eliminating race conditions from shared collection access.
• Batch Processing: Fetching all pending commands at once minimizes lock contention and reduces DB overhead.
• Isolated DB Operations: The receiver manages DB connections and transactions safely for concurrent persistence calls.

Note for Multi-Process Scenarios #

If you're working across separate processes (not just threads), shared in-memory queues won't work (processes don't share memory). Instead:

• Use a distributed message queue (e.g., Redis List, Kafka) to store audit commands.
• Each process sends commands to the queue, and a dedicated worker process runs the run() logic to persist to the DB.

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