我有一个 Django REST API 端点。它接收 JSON 有效负载，例如。
{ "data" : [0,1,2,3] }
这在 views.py 函数中被解码并生成一个新的数据库对象，如下所示（伪代码）：
newobj = MyObj.(col0 = 0, col1= 1, col2 = 2, col3 = 3)
newobj.save()
在测试中，创建 x1000 个新对象的列表，然后进行批量创建要快 20 倍：
Myobj.objects.bulk_create(newobjs, 1000)
那么，问题是当我们有 1000 个 POST 时，如何将单个 POST 保存在 Django 中的某个位置以备批量写入？
【问题讨论】：
标签： python django database django-rest-framework django-orm
例如，您可以使用 Memcached 或 Redis 缓存它。 但是您将需要编写某种服务来检查缓存中有多少新项目，以及是否有更多。 1000 -> 插入它。
所以：
你真的需要它吗？ 如果数据已经存在会怎样？如果数据损坏？用户如何知道这一点？
【讨论】：
当我们有 1000 个 POST 时，将单个 POST 保存在 Django 中的某个位置，以便批量写入
你可以，
CSV文件
但正如维克多所提到的，为什么呢？为什么你这么关心SQL Insert 的速度，反正速度还是蛮快的？
当然，bulk_create要快得多，因为它需要对您的数据库服务器进行一次网络调用并将所有行添加到单个 SQL transaction 中，但只有在您使用它时才有意义实际上有一堆数据要加在一起。 - 最后，您必须将数据保存在某处，这将花费一些处理时间。
因为你的方法有很多缺点：
UNIQUE 或任何其他约束。使用像 Redis 这样的快速缓存来维护条目列表，在您的 api_view 中，您可以调用 cache.get 来获取当前列表，将对象附加到它，然后调用 cache.set 来更新它。在此之后添加一个检查，只要len(list) >= 1000 == True 调用bulk_create。您可能还想考虑使用Elasticsearch 处理如此大量的数据。
【讨论】：
感谢以上回复，答案包含了一些建议，但属于超集，所以这里是一个摘要。
这实际上是关于创建一个 FIFO。 memcached 被证明是不合适的（在尝试之后），因为只有 redis 具有启用此功能的列表功能，很好地解释了 here。
还要注意Django内置缓存不支持redis list api调用。
所以我们需要一个新的 docker-compose.yml 条目来添加 redis：
redis:
image: redis
ports:
- 6379:6379/tcp
networks:
- app-network
然后在views.py中我们添加：（注意redis的使用rpush）
import redis
...
redis_host=os.environ['REDIS_HOST']
redis_port = 6379
redis_password = ""
r = redis.StrictRedis(host=redis_host, port=redis_port, password=redis_password, decode_responses=True)
...
def write_post_to_redis(request):
payload = json.loads(request.body)
r.rpush("df",json.dumps(payload))
所以这会将接收到的有效负载推送到 redis 内存缓存中。我们现在需要读取（或弹出）它并写入 postgres 数据库。所以我们需要一个每n秒唤醒一次并检查的进程。为此，我们需要 Django background_task。首先，安装它：
pipenv install django-background-tasks
并添加到settings.py的已安装应用
INSTALLED_APPS = [
...
'background_task',
然后运行 migrate 添加后台任务表：
python manage.py migrate
现在在views.py中，添加：
from background_task import background
from background_task.models import CompletedTask
并添加将缓存数据写入 postgres 数据库的函数，注意装饰器声明它应该每 5 秒在后台运行一次。还要注意redis的使用lpop。
@background(schedule=5)
def write_cached_samples():
...
payload = json.loads(r.lpop('df'))
# now do your write of payload to postgres
... and delete the completed tasks or we'll have a big db leak
CompletedTask.objects.all().delete()
为了启动进程，将以下内容添加到 urls.py 的基础：
write_cached_samples(repeat=10, repeat_until=None)
最后，因为后台任务需要一个单独的进程，我们在docker-compose.yml中复制django docker容器，但是将asgi server run命令替换为后台进程运行命令。
django_bg:
image: my_django
command: >
sh -c "python manage.py process_tasks"
...
总的来说，我们添加了两个新的 docker 容器，一个用于 redis 内存缓存，一个用于运行 django 后台任务。我们使用 redis lists rpush 和 lpop 函数来创建一个带有 API 接收推送和后台任务的 FIFO弹出。
有一个小问题是 nginx 连接到错误的 django 容器，通过停止并重新启动后台容器来纠正，一些问题是 docker 网络路由初始化错误。
接下来，我将用 Go 替换 Django HTTP API 端点，看看我们获得了多大的速度，因为 Daphne ASGI 服务器每秒只有 100 个请求达到最大 CPU。
【讨论】：