Django微服务架构中，如何仅向Kafka发消息而非写入数据库？

方案1：给模型加Mixin扩展save（推荐，可控性强） #

写一个抽象Mixin，不用完全重写原生save方法，只添加跳过DB写入的逻辑，同时处理ManyToMany的变更事件：

from django.db import models
from django.db.models.signals import m2m_changed
# 导入你的Kafka生产者模块
import your_kafka_producer

class KafkaSaveMixin(models.Model):
    class Meta:
        abstract = True

    def save(self, skip_db=True, *args, **kwargs):
        # 捕获实例初始状态，用于记录变更内容
        initial_data = None
        if not self._state.adding:
            initial_data = {f.name: getattr(self, f.name) for f in self._meta.fields if not f.auto_created}

        # 仅当需要强制写入DB时，调用原生save
        if not skip_db:
            super().save(*args, **kwargs)
        
        # 发送Kafka事件，包含模型操作信息
        operation = 'create' if self._state.adding else 'update'
        your_kafka_producer.send(
            topic='db_write_events',
            value={
                'model_label': self._meta.label,
                'operation': operation,
                'instance_data': self.__dict__,
                'initial_data': initial_data
            }
        )

    # 处理ManyToMany字段的变更
    def _handle_m2m_change(self, sender, instance, action, reverse, model, pk_set, **kwargs):
        if action in ['post_add', 'post_remove', 'post_clear']:
            your_kafka_producer.send(
                topic='db_write_events',
                value={
                    'model_label': instance._meta.label,
                    'operation': f'm2m_{action}',
                    'instance_id': instance.pk,
                    'm2m_field_name': sender.name,
                    'related_model': model._meta.label,
                    'affected_pks': list(pk_set)
                }
            )

# 在业务模型中使用Mixin
class Product(KafkaSaveMixin, models.Model):
    name = models.CharField(max_length=100)
    categories = models.ManyToManyField('Category')

# 为ManyToMany字段注册信号
m2m_changed.connect(Product._handle_m2m_change, sender=Product.categories.through)

方案2：全局替换Model.save（适合批量改造） #

如果不想逐个修改模型，可以用猴子补丁替换原生的save方法，全局拦截写入操作：

from django.db import models
from django.db.models.signals import m2m_changed
import your_kafka_producer

# 保存原生save方法
original_save = models.Model.save

def kafka_only_save(self, skip_db=True, *args, **kwargs):
    if not skip_db:
        return original_save(self, *args, **kwargs)
    
    # 发送Kafka事件逻辑
    operation = 'create' if self._state.adding else 'update'
    initial_data = {f.name: getattr(self, f.name) for f in self._meta.fields if not f.auto_created} if not self._state.adding else None
    your_kafka_producer.send(
        topic='db_write_events',
        value={
            'model_label': self._meta.label,
            'operation': operation,
            'instance_data': self.__dict__,
            'initial_data': initial_data
        }
    )

# 替换原生save方法
models.Model.save = kafka_only_save

# 全局处理ManyToMany变更
def global_m2m_handler(sender, instance, action, reverse, model, pk_set, **kwargs):
    if action in ['post_add', 'post_remove', 'post_clear']:
        your_kafka_producer.send(
            topic='db_write_events',
            value={
                'model_label': instance._meta.label,
                'operation': f'm2m_{action}',
                'instance_id': instance.pk,
                'm2m_field_name': sender.name,
                'related_model': model._meta.label,
                'affected_pks': list(pk_set)
            }
        )

m2m_changed.connect(global_m2m_handler)

核心要点 #

• 读取不受影响：所有get()、filter()等查询操作依然直接访问SQL数据库，完全保留原生逻辑。
• ManyToMany兼容：通过m2m_changed信号捕获关联变更，避免重写save导致的关联数据丢失问题。
• 灵活切换：如果需要临时直接写入数据库，调用instance.save(skip_db=False)即可。

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