Bagaimana saya harus login sambil menggunakan multiprosesor dengan Python?

Saat ini saya memiliki modul pusat dalam kerangka kerja yang memunculkan beberapa proses menggunakan multiprocessingmodul Python 2.6 . Karena itu digunakan multiprocessing, ada log tingkat-sadar multiprocessing modul LOG = multiprocessing.get_logger(),. Per dokumen , logger ini memiliki kunci proses bersama sehingga Anda tidak memutarbalikkan hal-hal sys.stderr(atau file apa pun yang ditangani) dengan meminta beberapa proses menulis secara bersamaan.

Masalah yang saya miliki sekarang adalah bahwa modul-modul lain dalam kerangka kerja ini tidak sadar multiprosesor. Cara saya melihatnya, saya perlu membuat semua dependensi pada modul pusat ini menggunakan pencatatan multiprocessing-aware. Itu menjengkelkan dalam kerangka kerja, apalagi untuk semua klien kerangka kerja. Apakah ada alternatif yang tidak saya pikirkan?

Satu-satunya cara untuk menangani hal ini secara non-intrusively adalah dengan:

1. Menelurkan setiap pekerja proses sehingga log-nya pergi ke deskriptor file yang berbeda (ke disk atau ke pipa.) Idealnya, semua entri log harus cap waktu.
2. Proses controller Anda kemudian dapat melakukan salah satu dari yang berikut:
   • Jika menggunakan file disk: Menyatukan file log di akhir proses, diurutkan berdasarkan cap waktu
   • Jika menggunakan pipa (disarankan): Menyatukan entri log dengan cepat dari semua pipa, ke file log pusat. (Mis., Secara berkala selectdari deskriptor file pipa, lakukan penggabungan-sort pada entri log yang tersedia, dan siram ke log terpusat. Ulangi.)

Saya baru saja menulis log handler saya sendiri yang hanya memberi makan segalanya untuk proses induk melalui pipa. Saya hanya mengujinya selama sepuluh menit tetapi tampaknya bekerja dengan cukup baik.

( Catatan: Ini adalah hardcoded RotatingFileHandler, yang merupakan kasus penggunaan saya sendiri.)

Pembaruan: @javier sekarang mempertahankan pendekatan ini sebagai paket yang tersedia di Pypi - lihat multiprocessing-logging di Pypi, github di https://github.com/jruere/multiprocessing-logging

Perbarui: Implementasi!

Ini sekarang menggunakan antrian untuk penanganan konkurensi yang benar, dan juga pulih dari kesalahan dengan benar. Saya sekarang telah menggunakan ini dalam produksi selama beberapa bulan, dan versi saat ini di bawah ini berfungsi tanpa masalah.

from logging.handlers import RotatingFileHandler
import multiprocessing, threading, logging, sys, traceback

class MultiProcessingLog(logging.Handler):
    def __init__(self, name, mode, maxsize, rotate):
        logging.Handler.__init__(self)

        self._handler = RotatingFileHandler(name, mode, maxsize, rotate)
        self.queue = multiprocessing.Queue(-1)

        t = threading.Thread(target=self.receive)
        t.daemon = True
        t.start()

    def setFormatter(self, fmt):
        logging.Handler.setFormatter(self, fmt)
        self._handler.setFormatter(fmt)

    def receive(self):
        while True:
            try:
                record = self.queue.get()
                self._handler.emit(record)
            except (KeyboardInterrupt, SystemExit):
                raise
            except EOFError:
                break
            except:
                traceback.print_exc(file=sys.stderr)

    def send(self, s):
        self.queue.put_nowait(s)

    def _format_record(self, record):
        # ensure that exc_info and args
        # have been stringified.  Removes any chance of
        # unpickleable things inside and possibly reduces
        # message size sent over the pipe
        if record.args:
            record.msg = record.msg % record.args
            record.args = None
        if record.exc_info:
            dummy = self.format(record)
            record.exc_info = None

        return record

    def emit(self, record):
        try:
            s = self._format_record(record)
            self.send(s)
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self.handleError(record)

    def close(self):
        self._handler.close()
        logging.Handler.close(self)

QueueHandleradalah asli dalam Python 3.2+, dan melakukan hal ini. Ini mudah direplikasi di versi sebelumnya.

Python docs memiliki dua contoh lengkap: Masuk ke satu file dari beberapa proses

Bagi mereka yang menggunakan Python <3.2, cukup salin QueueHandlerke kode Anda sendiri dari: https://gist.github.com/vsajip/591589 atau impor logutils alternatif .

Setiap proses (termasuk proses induk) menempatkan logging pada Queue, dan kemudian listenerutas atau proses (satu contoh disediakan untuk masing-masing) mengambil mereka dan menulis semuanya ke file - tidak ada risiko korupsi atau kekacauan.

Di bawah ini adalah solusi lain dengan fokus pada kesederhanaan untuk orang lain (seperti saya) yang datang ke sini dari Google. Penebangan harus mudah! Hanya untuk 3,2 atau lebih tinggi.

import multiprocessing
import logging
from logging.handlers import QueueHandler, QueueListener
import time
import random


def f(i):
    time.sleep(random.uniform(.01, .05))
    logging.info('function called with {} in worker thread.'.format(i))
    time.sleep(random.uniform(.01, .05))
    return i


def worker_init(q):
    # all records from worker processes go to qh and then into q
    qh = QueueHandler(q)
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    logger.addHandler(qh)


def logger_init():
    q = multiprocessing.Queue()
    # this is the handler for all log records
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter("%(levelname)s: %(asctime)s - %(process)s - %(message)s"))

    # ql gets records from the queue and sends them to the handler
    ql = QueueListener(q, handler)
    ql.start()

    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    # add the handler to the logger so records from this process are handled
    logger.addHandler(handler)

    return ql, q


def main():
    q_listener, q = logger_init()

    logging.info('hello from main thread')
    pool = multiprocessing.Pool(4, worker_init, [q])
    for result in pool.map(f, range(10)):
        pass
    pool.close()
    pool.join()
    q_listener.stop()

if __name__ == '__main__':
    main()

Namun alternatif lain mungkin berbagai penangan logging berbasis file dalam loggingpaket :

• SocketHandler
• DatagramHandler
• SyslogHandler

(dan lain-lain)

Dengan cara ini, Anda dapat dengan mudah memiliki daemon logging di suatu tempat yang dapat Anda tulis dengan aman dan akan menangani hasilnya dengan benar. (Misalnya, server soket sederhana yang hanya mengurai pesan dan memancarkannya ke penangan file rotasinya sendiri.)

The SyslogHandlerakan mengurus ini untuk Anda juga. Tentu saja, Anda bisa menggunakan contoh Anda sendiri syslog, bukan yang sistem.

Varian yang lain yang membuat log logging dan antrian terpisah.

"""sample code for logging in subprocesses using multiprocessing

* Little handler magic - The main process uses loggers and handlers as normal.
* Only a simple handler is needed in the subprocess that feeds the queue.
* Original logger name from subprocess is preserved when logged in main
  process.
* As in the other implementations, a thread reads the queue and calls the
  handlers. Except in this implementation, the thread is defined outside of a
  handler, which makes the logger definitions simpler.
* Works with multiple handlers.  If the logger in the main process defines
  multiple handlers, they will all be fed records generated by the
  subprocesses loggers.

tested with Python 2.5 and 2.6 on Linux and Windows

"""

import os
import sys
import time
import traceback
import multiprocessing, threading, logging, sys

DEFAULT_LEVEL = logging.DEBUG

formatter = logging.Formatter("%(levelname)s: %(asctime)s - %(name)s - %(process)s - %(message)s")

class SubProcessLogHandler(logging.Handler):
    """handler used by subprocesses

    It simply puts items on a Queue for the main process to log.

    """

    def __init__(self, queue):
        logging.Handler.__init__(self)
        self.queue = queue

    def emit(self, record):
        self.queue.put(record)

class LogQueueReader(threading.Thread):
    """thread to write subprocesses log records to main process log

    This thread reads the records written by subprocesses and writes them to
    the handlers defined in the main process's handlers.

    """

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.daemon = True

    def run(self):
        """read from the queue and write to the log handlers

        The logging documentation says logging is thread safe, so there
        shouldn't be contention between normal logging (from the main
        process) and this thread.

        Note that we're using the name of the original logger.

        """
        # Thanks Mike for the error checking code.
        while True:
            try:
                record = self.queue.get()
                # get the logger for this record
                logger = logging.getLogger(record.name)
                logger.callHandlers(record)
            except (KeyboardInterrupt, SystemExit):
                raise
            except EOFError:
                break
            except:
                traceback.print_exc(file=sys.stderr)

class LoggingProcess(multiprocessing.Process):

    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue

    def _setupLogger(self):
        # create the logger to use.
        logger = logging.getLogger('test.subprocess')
        # The only handler desired is the SubProcessLogHandler.  If any others
        # exist, remove them. In this case, on Unix and Linux the StreamHandler
        # will be inherited.

        for handler in logger.handlers:
            # just a check for my sanity
            assert not isinstance(handler, SubProcessLogHandler)
            logger.removeHandler(handler)
        # add the handler
        handler = SubProcessLogHandler(self.queue)
        handler.setFormatter(formatter)
        logger.addHandler(handler)

        # On Windows, the level will not be inherited.  Also, we could just
        # set the level to log everything here and filter it in the main
        # process handlers.  For now, just set it from the global default.
        logger.setLevel(DEFAULT_LEVEL)
        self.logger = logger

    def run(self):
        self._setupLogger()
        logger = self.logger
        # and here goes the logging
        p = multiprocessing.current_process()
        logger.info('hello from process %s with pid %s' % (p.name, p.pid))


if __name__ == '__main__':
    # queue used by the subprocess loggers
    queue = multiprocessing.Queue()
    # Just a normal logger
    logger = logging.getLogger('test')
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(DEFAULT_LEVEL)
    logger.info('hello from the main process')
    # This thread will read from the subprocesses and write to the main log's
    # handlers.
    log_queue_reader = LogQueueReader(queue)
    log_queue_reader.start()
    # create the processes.
    for i in range(10):
        p = LoggingProcess(queue)
        p.start()
    # The way I read the multiprocessing warning about Queue, joining a
    # process before it has finished feeding the Queue can cause a deadlock.
    # Also, Queue.empty() is not realiable, so just make sure all processes
    # are finished.
    # active_children joins subprocesses when they're finished.
    while multiprocessing.active_children():
        time.sleep(.1)

Semua solusi saat ini terlalu digabungkan ke konfigurasi logging dengan menggunakan handler. Solusi saya memiliki arsitektur dan fitur berikut:

• Anda dapat menggunakan setiap konfigurasi penebangan yang Anda inginkan
• Logging dilakukan di daemon thread
• Shutdown yang aman dari daemon dengan menggunakan manajer konteks
• Komunikasi ke utas pendataan dilakukan oleh multiprocessing.Queue
• Dalam subproses, logging.Logger(dan contoh yang sudah ditentukan) ditambal untuk mengirim semua catatan ke antrian
• Baru : memformat traceback dan pesan sebelum mengirim ke antrian untuk mencegah kesalahan pengawetan

Kode dengan contoh penggunaan dan keluaran dapat ditemukan di Intisari berikut: https://gist.github.com/schlamar/7003737

Karena kami dapat merepresentasikan multiprocess logging karena banyak penerbit dan satu pelanggan (pendengar), menggunakan ZeroMQ untuk mengimplementasikan pesan PUB-SUB memang merupakan pilihan.

Selain itu, modul PyZMQ , binding Python untuk ZMQ, mengimplementasikan PUBHandler , yang merupakan objek untuk mempublikasikan pesan logging melalui soket zmq.PUB.

Ada solusi di web , untuk logging terpusat dari aplikasi terdistribusi menggunakan PyZMQ dan PUBHandler, yang dapat dengan mudah diadopsi untuk bekerja secara lokal dengan berbagai proses penerbitan.

formatters = {
    logging.DEBUG: logging.Formatter("[%(name)s] %(message)s"),
    logging.INFO: logging.Formatter("[%(name)s] %(message)s"),
    logging.WARN: logging.Formatter("[%(name)s] %(message)s"),
    logging.ERROR: logging.Formatter("[%(name)s] %(message)s"),
    logging.CRITICAL: logging.Formatter("[%(name)s] %(message)s")
}

# This one will be used by publishing processes
class PUBLogger:
    def __init__(self, host, port=config.PUBSUB_LOGGER_PORT):
        self._logger = logging.getLogger(__name__)
        self._logger.setLevel(logging.DEBUG)
        self.ctx = zmq.Context()
        self.pub = self.ctx.socket(zmq.PUB)
        self.pub.connect('tcp://{0}:{1}'.format(socket.gethostbyname(host), port))
        self._handler = PUBHandler(self.pub)
        self._handler.formatters = formatters
        self._logger.addHandler(self._handler)

    @property
    def logger(self):
        return self._logger

# This one will be used by listener process
class SUBLogger:
    def __init__(self, ip, output_dir="", port=config.PUBSUB_LOGGER_PORT):
        self.output_dir = output_dir
        self._logger = logging.getLogger()
        self._logger.setLevel(logging.DEBUG)

        self.ctx = zmq.Context()
        self._sub = self.ctx.socket(zmq.SUB)
        self._sub.bind('tcp://*:{1}'.format(ip, port))
        self._sub.setsockopt(zmq.SUBSCRIBE, "")

        handler = handlers.RotatingFileHandler(os.path.join(output_dir, "client_debug.log"), "w", 100 * 1024 * 1024, 10)
        handler.setLevel(logging.DEBUG)
        formatter = logging.Formatter("%(asctime)s;%(levelname)s - %(message)s")
        handler.setFormatter(formatter)
        self._logger.addHandler(handler)

  @property
  def sub(self):
      return self._sub

  @property
  def logger(self):
      return self._logger

#  And that's the way we actually run things:

# Listener process will forever listen on SUB socket for incoming messages
def run_sub_logger(ip, event):
    sub_logger = SUBLogger(ip)
    while not event.is_set():
        try:
            topic, message = sub_logger.sub.recv_multipart(flags=zmq.NOBLOCK)
            log_msg = getattr(logging, topic.lower())
            log_msg(message)
        except zmq.ZMQError as zmq_error:
            if zmq_error.errno == zmq.EAGAIN:
                pass


# Publisher processes loggers should be initialized as follows:

class Publisher:
    def __init__(self, stop_event, proc_id):
        self.stop_event = stop_event
        self.proc_id = proc_id
        self._logger = pub_logger.PUBLogger('127.0.0.1').logger

     def run(self):
         self._logger.info("{0} - Sending message".format(proc_id))

def run_worker(event, proc_id):
    worker = Publisher(event, proc_id)
    worker.run()

# Starting subscriber process so we won't loose publisher's messages
sub_logger_process = Process(target=run_sub_logger,
                                 args=('127.0.0.1'), stop_event,))
sub_logger_process.start()

#Starting publisher processes
for i in range(MAX_WORKERS_PER_CLIENT):
    processes.append(Process(target=run_worker,
                                 args=(stop_event, i,)))
for p in processes:
    p.start()

Saya juga suka jawaban zzzeek tetapi Andre benar bahwa antrian diperlukan untuk mencegah gangguan. Saya beruntung dengan pipa, tetapi memang melihat kekacauan yang agak diharapkan. Menerapkannya ternyata lebih sulit dari yang saya kira, terutama karena berjalan di Windows, di mana ada beberapa batasan tambahan tentang variabel dan hal-hal global (lihat: Bagaimana Proses Multi-Python Diimplementasikan pada Windows? )

Tapi, akhirnya aku berhasil. Contoh ini mungkin tidak sempurna, jadi komentar dan saran dipersilahkan. Itu juga tidak mendukung pengaturan formatter atau apa pun selain root logger. Pada dasarnya, Anda harus menginstal ulang logger di setiap proses kumpulan dengan antrian dan mengatur atribut lainnya pada logger.

Sekali lagi, saran tentang cara membuat kode lebih baik dipersilakan. Saya pasti belum tahu semua trik Python :-)

import multiprocessing, logging, sys, re, os, StringIO, threading, time, Queue

class MultiProcessingLogHandler(logging.Handler):
    def __init__(self, handler, queue, child=False):
        logging.Handler.__init__(self)

        self._handler = handler
        self.queue = queue

        # we only want one of the loggers to be pulling from the queue.
        # If there is a way to do this without needing to be passed this
        # information, that would be great!
        if child == False:
            self.shutdown = False
            self.polltime = 1
            t = threading.Thread(target=self.receive)
            t.daemon = True
            t.start()

    def setFormatter(self, fmt):
        logging.Handler.setFormatter(self, fmt)
        self._handler.setFormatter(fmt)

    def receive(self):
        #print "receive on"
        while (self.shutdown == False) or (self.queue.empty() == False):
            # so we block for a short period of time so that we can
            # check for the shutdown cases.
            try:
                record = self.queue.get(True, self.polltime)
                self._handler.emit(record)
            except Queue.Empty, e:
                pass

    def send(self, s):
        # send just puts it in the queue for the server to retrieve
        self.queue.put(s)

    def _format_record(self, record):
        ei = record.exc_info
        if ei:
            dummy = self.format(record) # just to get traceback text into record.exc_text
            record.exc_info = None  # to avoid Unpickleable error

        return record

    def emit(self, record):
        try:
            s = self._format_record(record)
            self.send(s)
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self.handleError(record)

    def close(self):
        time.sleep(self.polltime+1) # give some time for messages to enter the queue.
        self.shutdown = True
        time.sleep(self.polltime+1) # give some time for the server to time out and see the shutdown

    def __del__(self):
        self.close() # hopefully this aids in orderly shutdown when things are going poorly.

def f(x):
    # just a logging command...
    logging.critical('function number: ' + str(x))
    # to make some calls take longer than others, so the output is "jumbled" as real MP programs are.
    time.sleep(x % 3)

def initPool(queue, level):
    """
    This causes the logging module to be initialized with the necessary info
    in pool threads to work correctly.
    """
    logging.getLogger('').addHandler(MultiProcessingLogHandler(logging.StreamHandler(), queue, child=True))
    logging.getLogger('').setLevel(level)

if __name__ == '__main__':
    stream = StringIO.StringIO()
    logQueue = multiprocessing.Queue(100)
    handler= MultiProcessingLogHandler(logging.StreamHandler(stream), logQueue)
    logging.getLogger('').addHandler(handler)
    logging.getLogger('').setLevel(logging.DEBUG)

    logging.debug('starting main')

    # when bulding the pool on a Windows machine we also have to init the logger in all the instances with the queue and the level of logging.
    pool = multiprocessing.Pool(processes=10, initializer=initPool, initargs=[logQueue, logging.getLogger('').getEffectiveLevel()] ) # start worker processes
    pool.map(f, range(0,50))
    pool.close()

    logging.debug('done')
    logging.shutdown()
    print "stream output is:"
    print stream.getvalue()

cukup terbitkan di suatu tempat instance Anda dari logger. dengan begitu, modul dan klien lain dapat menggunakan API Anda untuk mendapatkan logger tanpa harus melakukannya import multiprocessing.

Saya menyukai jawaban zzzeek. Saya hanya akan mengganti Pipa untuk Antrian karena jika beberapa utas / proses menggunakan ujung pipa yang sama untuk menghasilkan pesan log mereka akan kacau.

Bagaimana dengan mendelegasikan semua pencatatan ke proses lain yang membaca semua entri log dari Antrian?

LOG_QUEUE = multiprocessing.JoinableQueue()

class CentralLogger(multiprocessing.Process):
    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue
        self.log = logger.getLogger('some_config')
        self.log.info("Started Central Logging process")

    def run(self):
        while True:
            log_level, message = self.queue.get()
            if log_level is None:
                self.log.info("Shutting down Central Logging process")
                break
            else:
                self.log.log(log_level, message)

central_logger_process = CentralLogger(LOG_QUEUE)
central_logger_process.start()

Cukup bagikan LOG_QUEUE melalui salah satu mekanisme multiproses atau bahkan warisan dan semuanya bekerja dengan baik!

Saya punya solusi yang mirip dengan ironhacker kecuali saya menggunakan logging.exception di beberapa kode saya dan menemukan bahwa saya perlu memformat pengecualian sebelum meneruskannya kembali ke Antrean karena traceback tidak dapat dipilih:

class QueueHandler(logging.Handler):
    def __init__(self, queue):
        logging.Handler.__init__(self)
        self.queue = queue
    def emit(self, record):
        if record.exc_info:
            # can't pass exc_info across processes so just format now
            record.exc_text = self.formatException(record.exc_info)
            record.exc_info = None
        self.queue.put(record)
    def formatException(self, ei):
        sio = cStringIO.StringIO()
        traceback.print_exception(ei[0], ei[1], ei[2], None, sio)
        s = sio.getvalue()
        sio.close()
        if s[-1] == "\n":
            s = s[:-1]
        return s

Di bawah ini adalah kelas yang dapat digunakan di lingkungan Windows, membutuhkan ActivePython. Anda juga bisa mewarisi untuk penangan logging lainnya (StreamHandler dll.)

class SyncronizedFileHandler(logging.FileHandler):
    MUTEX_NAME = 'logging_mutex'

    def __init__(self , *args , **kwargs):

        self.mutex = win32event.CreateMutex(None , False , self.MUTEX_NAME)
        return super(SyncronizedFileHandler , self ).__init__(*args , **kwargs)

    def emit(self, *args , **kwargs):
        try:
            win32event.WaitForSingleObject(self.mutex , win32event.INFINITE)
            ret = super(SyncronizedFileHandler , self ).emit(*args , **kwargs)
        finally:
            win32event.ReleaseMutex(self.mutex)
        return ret

Dan berikut ini adalah contoh yang menunjukkan penggunaan:

import logging
import random , time , os , sys , datetime
from string import letters
import win32api , win32event
from multiprocessing import Pool

def f(i):
    time.sleep(random.randint(0,10) * 0.1)
    ch = random.choice(letters)
    logging.info( ch * 30)


def init_logging():
    '''
    initilize the loggers
    '''
    formatter = logging.Formatter("%(levelname)s - %(process)d - %(asctime)s - %(filename)s - %(lineno)d - %(message)s")
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    file_handler = SyncronizedFileHandler(sys.argv[1])
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

#must be called in the parent and in every worker process
init_logging() 

if __name__ == '__main__':
    #multiprocessing stuff
    pool = Pool(processes=10)
    imap_result = pool.imap(f , range(30))
    for i , _ in enumerate(imap_result):
        pass

Inilah solusi sederhana saya / pemecahan masalah ... bukan yang paling komprehensif, tetapi mudah dimodifikasi dan lebih sederhana untuk dibaca dan dipahami menurut saya daripada jawaban lain yang saya temukan sebelum menulis ini:

import logging
import multiprocessing

class FakeLogger(object):
    def __init__(self, q):
        self.q = q
    def info(self, item):
        self.q.put('INFO - {}'.format(item))
    def debug(self, item):
        self.q.put('DEBUG - {}'.format(item))
    def critical(self, item):
        self.q.put('CRITICAL - {}'.format(item))
    def warning(self, item):
        self.q.put('WARNING - {}'.format(item))

def some_other_func_that_gets_logger_and_logs(num):
    # notice the name get's discarded
    # of course you can easily add this to your FakeLogger class
    local_logger = logging.getLogger('local')
    local_logger.info('Hey I am logging this: {} and working on it to make this {}!'.format(num, num*2))
    local_logger.debug('hmm, something may need debugging here')
    return num*2

def func_to_parallelize(data_chunk):
    # unpack our args
    the_num, logger_q = data_chunk
    # since we're now in a new process, let's monkeypatch the logging module
    logging.getLogger = lambda name=None: FakeLogger(logger_q)
    # now do the actual work that happens to log stuff too
    new_num = some_other_func_that_gets_logger_and_logs(the_num)
    return (the_num, new_num)

if __name__ == '__main__':
    multiprocessing.freeze_support()
    m = multiprocessing.Manager()
    logger_q = m.Queue()
    # we have to pass our data to be parallel-processed
    # we also need to pass the Queue object so we can retrieve the logs
    parallelable_data = [(1, logger_q), (2, logger_q)]
    # set up a pool of processes so we can take advantage of multiple CPU cores
    pool_size = multiprocessing.cpu_count() * 2
    pool = multiprocessing.Pool(processes=pool_size, maxtasksperchild=4)
    worker_output = pool.map(func_to_parallelize, parallelable_data)
    pool.close() # no more tasks
    pool.join()  # wrap up current tasks
    # get the contents of our FakeLogger object
    while not logger_q.empty():
        print logger_q.get()
    print 'worker output contained: {}'.format(worker_output)

Ada paket bagus ini

Paket: https://pypi.python.org/pypi/multiprocessing-logging/

kode: https://github.com/jruere/multiprocessing-logging

Install:

pip install multiprocessing-logging

Kemudian tambahkan:

import multiprocessing_logging

# This enables logs inside process
multiprocessing_logging.install_mp_handler()

Salah satu alternatifnya adalah menulis penebangan mutliprocessing ke file yang dikenal dan mendaftarkan atexitpawang untuk bergabung pada proses-proses tersebut membacanya kembali di stderr; namun, Anda tidak akan mendapatkan aliran waktu-nyata ke pesan keluaran di stderr dengan cara itu.

Jika Anda memiliki kebuntuan yang terjadi dalam kombinasi kunci, utas dan garpu dalam loggingmodul, yang dilaporkan dalam laporan bug 6721 (lihat juga pertanyaan SO terkait ).

Ada solusi perbaikan kecil yang diposting di sini .

Namun, itu hanya akan memperbaiki deadlock potensial di logging. Itu tidak akan memperbaiki hal-hal yang mungkin kacau. Lihat jawaban lain yang disajikan di sini.

Ide paling sederhana seperti yang disebutkan:

• Ambil nama file dan id proses dari proses saat ini.
• Siapkan a [WatchedFileHandler][1]. Alasan untuk pawang ini dibahas secara rinci di sini , tetapi singkatnya ada beberapa kondisi ras yang lebih buruk dengan pawang penebangan lainnya. Yang ini memiliki jendela terpendek untuk kondisi lomba.
  • Pilih jalur untuk menyimpan log seperti / var / log / ...

Untuk siapa pun yang mungkin membutuhkan ini, saya menulis dekorator untuk paket multiprocessing_logging yang menambahkan nama proses saat ini ke log, sehingga menjadi jelas siapa yang mencatat apa.

Itu juga menjalankan install_mp_handler () sehingga menjadi tidak berguna untuk menjalankannya sebelum membuat kumpulan.

Ini memungkinkan saya untuk melihat pekerja mana yang membuat pesan log mana.

Inilah cetak biru dengan contoh:

import sys
import logging
from functools import wraps
import multiprocessing
import multiprocessing_logging

# Setup basic console logger as 'logger'
logger = logging.getLogger()
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter(u'%(asctime)s :: %(levelname)s :: %(message)s'))
logger.setLevel(logging.DEBUG)
logger.addHandler(console_handler)


# Create a decorator for functions that are called via multiprocessing pools
def logs_mp_process_names(fn):
    class MultiProcessLogFilter(logging.Filter):
        def filter(self, record):
            try:
                process_name = multiprocessing.current_process().name
            except BaseException:
                process_name = __name__
            record.msg = f'{process_name} :: {record.msg}'
            return True

    multiprocessing_logging.install_mp_handler()
    f = MultiProcessLogFilter()

    # Wraps is needed here so apply / apply_async know the function name
    @wraps(fn)
    def wrapper(*args, **kwargs):
        logger.removeFilter(f)
        logger.addFilter(f)
        return fn(*args, **kwargs)

    return wrapper


# Create a test function and decorate it
@logs_mp_process_names
def test(argument):
    logger.info(f'test function called via: {argument}')


# You can also redefine undecored functions
def undecorated_function():
    logger.info('I am not decorated')


@logs_mp_process_names
def redecorated(*args, **kwargs):
    return undecorated_function(*args, **kwargs)


# Enjoy
if __name__ == '__main__':
    with multiprocessing.Pool() as mp_pool:
        # Also works with apply_async
        mp_pool.apply(test, ('mp pool',))
        mp_pool.apply(redecorated)
        logger.info('some main logs')
        test('main program')

Kepada anak-anak saya yang mengalami masalah yang sama dalam beberapa dekade dan menemukan pertanyaan ini di situs ini saya meninggalkan jawaban ini.

Kesederhanaan vs rumit. Cukup gunakan alat lain. Python memang luar biasa, tetapi itu tidak dirancang untuk melakukan beberapa hal.

Cuplikan berikut untuk daemon logrotate berfungsi untuk saya dan tidak terlalu rumit. Jadwalkan untuk menjalankan setiap jam dan

/var/log/mylogfile.log {
    size 1
    copytruncate
    create
    rotate 10
    missingok
    postrotate
        timeext=`date -d '1 hour ago' "+%Y-%m-%d_%H"`
        mv /var/log/mylogfile.log.1 /var/log/mylogfile-$timeext.log
    endscript
}

Ini adalah bagaimana saya menginstalnya (symlink tidak berfungsi untuk logrotate):

sudo cp /directpath/config/logrotate/myconfigname /etc/logrotate.d/myconfigname
sudo cp /etc/cron.daily/logrotate /etc/cron.hourly/logrotate