如何设计数据生产者类与任意数量消费者类间的数据交互

Hey there! Let's walk through practical design solutions for your scenario—where one class loads images from disk and shares them, with any number of independent consumer classes processing those images, and the sharing class doesn't depend on whether consumers exist. Since you're using C++ but this is mostly about general design patterns, I'll cover approaches that translate well to C++ with implementation notes.

1. Observer (Publish-Subscribe) Pattern #

This is the approach you already thought of, and it's a perfect fit for your requirement of decoupling the sharing class from consumers.

How it works #

• The image-loading class acts as the Subject: it maintains a list of registered observers (consumer classes) and notifies them whenever a new image is loaded.
• Consumer classes are Observers: they implement a common interface (e.g., ImageObserver) with a method like onImageReceived(const Image&) to handle new images.
• The Subject never needs to know anything specific about Observers—only that they implement the observer interface.

C++ Implementation Snippet #

// Observer Interface
class ImageObserver {
public:
    virtual ~ImageObserver() = default;
    virtual void onImageReceived(const std::shared_ptr<Image>& image) = 0;
};

// Subject (Image Loader)
class ImageLoader {
private:
    std::vector<std::weak_ptr<ImageObserver>> observers; // Use weak_ptr to avoid dangling references
public:
    void registerObserver(std::shared_ptr<ImageObserver> observer) {
        observers.push_back(observer);
    }

    void unregisterObserver(std::shared_ptr<ImageObserver> observer) {
        observers.erase(std::remove_if(observers.begin(), observers.end(),
            [&](const std::weak_ptr<ImageObserver>& ptr) {
                return ptr.lock() == observer;
            }), observers.end());
    }

    void loadImageFromDisk(const std::string& path) {
        auto image = std::make_shared<Image>(path); // Assume Image class exists
        notifyObservers(image);
    }

private:
    void notifyObservers(const std::shared_ptr<Image>& image) {
        for (auto& weak_ptr : observers) {
            if (auto observer = weak_ptr.lock()) {
                observer->onImageReceived(image);
            }
        }
    }
};

Pros #

• Strong decoupling: The ImageLoader has no dependencies on consumer classes—consumers can be added/removed at runtime without changing the loader.
• Automatic notifications: Consumers get immediate updates when new images are available, no polling needed.
• Scalable: Supports any number of consumers (0 or more) seamlessly.

Cons #

• Synchronous overhead (by default): If consumers take time to process images, synchronous notifications will block the ImageLoader until all observers finish. Fix this with async notifications (e.g., spawning threads for each observer or using a task queue).
• Thread safety requires care: If multiple threads are involved, you'll need mutexes to protect the observer list and image access.
• Potential memory leaks: Using raw pointers for observers can lead to dangling references—hence the weak_ptr in the snippet above.

2. Shared Data Pool with Pull Model #

Instead of pushing updates to consumers, the image loader maintains a shared pool of images, and consumers pull images from the pool when they're ready to process them.

How it works #

• The ImageLoader stores loaded images in a thread-safe cache (e.g., a map of image IDs to std::shared_ptr<Image>).
• Consumers query the cache periodically or on demand to retrieve new images (add a "new image available" flag or timestamp to avoid redundant checks).

C++ Implementation Snippet #

class ImageCache {
private:
    std::unordered_map<std::string, std::shared_ptr<Image>> images;
    std::mutex cache_mutex;
    std::atomic<bool> has_new_images = false;

public:
    void addImage(const std::string& id, std::shared_ptr<Image> image) {
        std::lock_guard<std::mutex> lock(cache_mutex);
        images[id] = image;
        has_new_images = true;
    }

    std::shared_ptr<Image> getImage(const std::string& id) {
        std::lock_guard<std::mutex> lock(cache_mutex);
        auto it = images.find(id);
        return (it != images.end()) ? it->second : nullptr;
    }

    bool hasNewImages() {
        return has_new_images.exchange(false); // Reset flag after checking
    }
};

// ImageLoader uses ImageCache
class ImageLoader {
private:
    std::shared_ptr<ImageCache> cache;
public:
    ImageLoader(std::shared_ptr<ImageCache> cache) : cache(cache) {}

    void loadImageFromDisk(const std::string& path) {
        auto image = std::make_shared<Image>(path);
        cache->addImage(path, image);
    }
};

// Consumer example
class ImageProcessor {
private:
    std::shared_ptr<ImageCache> cache;
public:
    ImageProcessor(std::shared_ptr<ImageCache> cache) : cache(cache) {}

    void processNewImages() {
        if (cache->hasNewImages()) {
            auto image = cache->getImage("latest_path");
            if (image) {
                // Process image...
            }
        }
    }
};

Pros #

• Consumer control: Consumers decide when to process images, which is great if processing times vary or if consumers run on different schedules.
• No coupling between loader and consumers: The loader only interacts with the cache, not consumers—consumers can exist or not without affecting the loader.
• Efficient for batch processing: Works well if consumers prefer to process multiple images at once instead of one at a time.

Cons #

• Polling overhead: Consumers need to check the cache periodically, which can waste resources if images are loaded infrequently. Mitigate this with condition variables (cache signals consumers when a new image is added).
• Cache management: You'll need to handle image eviction (removing old images to save memory) and ensure thread safety for cache access.

3. Producer-Consumer Pattern with Message Queue #

This is a variation of the observer pattern but uses an asynchronous queue to decouple image loading from processing.

How it works #

• The ImageLoader (producer) pushes loaded images into a thread-safe queue.
• Consumers (workers) pull images from the queue and process them independently.
• The queue acts as a buffer, allowing the loader to keep loading images even if consumers are busy.

C++ Implementation Snippet #

#include <queue>
#include <mutex>
#include <condition_variable>

class ImageQueue {
private:
    std::queue<std::shared_ptr<Image>> queue;
    std::mutex queue_mutex;
    std::condition_variable cv;
    bool is_closed = false;

public:
    void push(std::shared_ptr<Image> image) {
        std::lock_guard<std::mutex> lock(queue_mutex);
        queue.push(image);
        cv.notify_one();
    }

    bool tryPop(std::shared_ptr<Image>& out_image) {
        std::lock_guard<std::mutex> lock(queue_mutex);
        if (queue.empty()) return false;
        out_image = queue.front();
        queue.pop();
        return true;
    }

    void waitAndPop(std::shared_ptr<Image>& out_image) {
        std::unique_lock<std::mutex> lock(queue_mutex);
        cv.wait(lock, [this] { return !queue.empty() || is_closed; });
        if (is_closed && queue.empty()) return;
        out_image = queue.front();
        queue.pop();
    }

    void close() {
        std::lock_guard<std::mutex> lock(queue_mutex);
        is_closed = true;
        cv.notify_all();
    }
};

// Producer (ImageLoader)
class ImageLoader {
private:
    std::shared_ptr<ImageQueue> queue;
public:
    ImageLoader(std::shared_ptr<ImageQueue> queue) : queue(queue) {}

    void loadImages(const std::vector<std::string>& paths) {
        for (const auto& path : paths) {
            auto image = std::make_shared<Image>(path);
            queue->push(image);
        }
        queue->close(); // Signal no more images
    }
};

// Consumer example
class ImageProcessor {
private:
    std::shared_ptr<ImageQueue> queue;
public:
    ImageProcessor(std::shared_ptr<ImageQueue> queue) : queue(queue) {}

    void run() {
        std::shared_ptr<Image> image;
        while (true) {
            queue->waitAndPop(image);
            if (!image) break; // Queue closed and empty
            // Process image...
        }
    }
};

Pros #

• Full async decoupling: The loader doesn't wait for consumers to finish processing—images are buffered in the queue.
• Scalable consumers: Spin up any number of consumer threads to process images in parallel for better performance.
• Smooth flow control: The queue handles backpressure (if consumers are slow, the queue holds images until they're ready).

Cons #

• Queue management: Handle queue limits (to prevent memory overflow if images are loaded faster than processed) and clean up the queue when done.
• Complexity: Adding thread safety and condition variables increases code complexity compared to simpler patterns.
• No direct notification: Consumers can't know when a specific image is loaded—they just process whatever comes next in the queue.

Which to Choose? #

• Observer Pattern: Best if you need real-time, one-to-many notifications and want strict decoupling between loader and consumers. Use async notifications if processing is slow.
• Shared Data Pool: Ideal if consumers need to control when they process images, or if you need to access images by ID later (not just when they're loaded).
• Producer-Consumer Queue: Perfect for high-throughput scenarios where you want to parallelize processing and handle backpressure gracefully.

All three approaches satisfy your core requirement: the image-loading class doesn't depend on the existence of consumers. Pick the one that aligns best with your performance needs and processing workflow!

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