Многопроцессорность Django и подключения к базе данных

Ответ 1

Многопроцессорная копирует объекты соединения между процессами, потому что он обрабатывает процессы и, следовательно, копирует все дескрипторы файлов родительского процесса. При этом соединение с SQL-сервером является просто файлом, вы можете увидеть его в linux под /proc//fd/.... любой открытый файл будет разделяться между разветвленными процессами. Вы можете найти больше о forking здесь.

Мое решение было просто просто подключить db непосредственно перед запуском процессов, каждый процесс воссоздает соединение, когда оно понадобится (проверено в django 1.4):

from django import db
db.connections.close_all()
def db_worker():      
    some_paralell_code()
Process(target = db_worker,args = ())

Pgbouncer/pgpool не связан с потоками в смысле многопроцессорности. Это скорее решение для не закрытия соединения по каждому запросу = ускорение подключения к postgres при высокой нагрузке.

Update:

Чтобы полностью удалить проблемы с подключением к базе данных, просто переместите всю логику, связанную с базой данных, на db_worker - я хотел передать QueryDict в качестве аргумента... Лучшая идея - просто передать список идентификаторов... См. QueryDict и values_list ('id', flat = True) и не забудьте включить его в список! (QueryDict) перед тем, как перейти к db_worker. Благодаря этому мы не копируем соединение с базой данных моделей.

def db_worker(models_ids):        
    obj = PartModelWorkerClass(model_ids) # here You do Model.objects.filter(id__in = model_ids)
    obj.run()


model_ids = Model.objects.all().values_list('id', flat=True)
model_ids = list(model_ids) # cast to list
process_count = 5
delta = (len(model_ids) / process_count) + 1

# do all the db stuff here ...

# here you can close db connection
from django import db
db.connections.close_all()

for it in range(0:process_count):
    Process(target = db_worker,args = (model_ids[it*delta:(it+1)*delta]))   

Ответ 2

При использовании нескольких баз данных вы должны закрыть все подключения.

from django import db
for connection_name in db.connections.databases:
    db.connections[connection_name].close()

ИЗМЕНИТЬ

Пожалуйста, используйте то же самое, что и @lechup, упомянутый для закрытия всех подключений (не уверен, с какой версии django этот метод был добавлен):

from django import db
db.connections.close_all()

Ответ 3

Для Python 3 и Django 1.9 это то, что сработало для меня:

import multiprocessing
import django
django.setup() # Must call setup

def db_worker():
    for name, info in django.db.connections.databases.items(): # Close the DB connections
        django.db.connection.close()
    # Execute parallel code here

if __name__ == '__main__':
    multiprocessing.Process(target=db_worker)

Обратите внимание, что без django.setup() я не мог заставить это работать. Я предполагаю, что что-то нужно снова инициализировать для многопроцессорности.

Ответ 4

Привет, я столкнулся с этой проблемой и смог ее решить, выполнив следующие действия (мы внедряем ограниченную систему задач)

task.py

from django.db import connection

def as_task(fn):
    """  this is a decorator that handles task duties, like setting up loggers, reporting on status...etc """ 
    connection.close()  #  this is where i kill the database connection VERY IMPORTANT
    # This will force django to open a new unique connection, since on linux at least
    # Connections do not fare well when forked 
    #...etc

ScheduledJob.py

from django.db import connection

def run_task(request, job_id):
    """ Just a simple view that when hit with a specific job id kicks of said job """ 
    # your logic goes here
    # ...
    processor = multiprocessing.Queue()
    multiprocessing.Process(
        target=call_command,  # all of our tasks are setup as management commands in django
        args=[
            job_info.management_command,
        ],
        kwargs= {
            'web_processor': processor,
        }.items() + vars(options).items()).start()

result = processor.get(timeout=10)  # wait to get a response on a successful init
# Result is a tuple of [TRUE|FALSE,<ErrorMessage>]
if not result[0]:
    raise Exception(result[1])
else:
   # THE VERY VERY IMPORTANT PART HERE, notice that up to this point we haven't touched the db again, but now we absolutely have to call connection.close()
   connection.close()
   # we do some database accessing here to get the most recently updated job id in the database

Честно говоря, для предотвращения условий гонки (с несколькими одновременными пользователями) было бы лучше всего вызвать database.close() как можно быстрее после того, как вы перекроете процесс. Возможно, все еще есть вероятность того, что другой пользователь где-то на линии полностью отправит запрос в db, прежде чем у вас будет возможность сбросить базу данных.

Всем честно, скорее всего, будет безопаснее и умнее, чтобы ваша вилка не вызывала команду напрямую, а вместо этого вызывала script в операционной системе, чтобы порожденная задача выполнялась самостоятельно django shell!

Ответ 5

(не большое решение, но возможное обходное решение)

Если вы не можете использовать сельдерей, возможно, вы могли бы реализовать свою собственную систему очередей, в основном добавляя задачи в какую-то таблицу задач и имея регулярный cron, который выбирает их и обрабатывает? (с помощью команды управления)

Ответ 6

Если вам нужно только I/O parallelism и не обрабатывать parallelism, вы можете избежать этой проблемы, переключив свои процессы на потоки. Заменить

from multiprocessing import Process

с

from threading import Thread

Объект Thread имеет тот же интерфейс, что и Procsess

Ответ 7

Вы можете предоставить больше ресурсов Postgre, в Debian/Ubuntu вы можете редактировать:

nano /etc/postgresql/9.4/main/postgresql.conf

заменив 9.4 вашей версией postgre.

Вот несколько полезных строк, которые должны быть обновлены с помощью значений примера, чтобы имена отображались сами:

max_connections=100
shared_buffers = 3000MB
temp_buffers = 800MB
effective_io_concurrency = 300
max_worker_processes = 80

Будьте осторожны, чтобы не увеличивать слишком много этих параметров, поскольку это может привести к ошибкам с Postgre, пытаясь получить больше ресурсов, чем доступно. Примеры, приведенные выше, отлично работают на машине Debian 8GB Ram, оборудованной четырьмя ядрами.