Реализация пользовательских протоколов вывода из нескольких потоков в python

Question

Моя практика в многопоточном Python довольно бедна. Итак, теперь я изучаю, как получать информацию о журналах из нескольких потоков. Я видел много разных подходов, но хочу начать с простого, я думаю. Таким образом, задача состоит в создании нескольких потоков и данных журнала из каждого из них. Чтобы распознать источник журналов, я хочу поместить некоторый пользовательский тег в вывод журнала. Я знаю, что logging lib имеет атрибуты LogRecord досягаемости (thread, threadName и т. Д.), И он работает хорошо. Итак, у меня есть пример (logging-from-multiple-threads) и внесите некоторые изменения. Вот полный код:

импорт каротаж импорт резьб время импорта

logger = logging.getLogger() 
syslog = logging.StreamHandler() 
formatter = logging.Formatter('%(project)s : %(thread)x ' 
           '%(levelname)-8s ' 
           '%(message)s') 
syslog.setFormatter(formatter) 
logger.setLevel(logging.DEBUG) 
logger.addHandler(syslog) 


class ContextFilter(logging.Filter): 

    def __init__(self, project): 
     super(ContextFilter, self).__init__() 
     self.project = project 

    def filter(self, record): 
     record.project = self.project 
     return True 


def worker(args): 
    while not args['stop']: 
     logging.debug('Hi from {}'.format(args['project'])) 
     time.sleep(0.5) 


def main(): 
    projects = ['project_1', 'project_2'] 
    info = {'stop': False} 
    threads = [] 
    for project in projects: 
     info['project'] = project 
     logger.addFilter(ContextFilter(project)) 
     thread = threading.Thread(target=worker, args=(info,)) 
     thread.start() 
     threads.append(thread) 
    while True: 
     try: 
      logging.debug('Hello from main') 
      time.sleep(1.75) 
     except KeyboardInterrupt: 
      info['stop'] = True 
      break 
    for t in threads: 
     t.join() 

if __name__ == '__main__': 
    main() 

А вот результаты вывода:

project_2 : 7fa627e77700 DEBUG Hi from project_2 
project_2 : 7fa6293d0700 DEBUG Hello from main 
project_2 : 7fa627676700 DEBUG Hi from project_2 
project_2 : 7fa627e77700 DEBUG Hi from project_2 
project_2 : 7fa627676700 DEBUG Hi from project_2 
project_2 : 7fa627e77700 DEBUG Hi from project_2 
project_2 : 7fa627676700 DEBUG Hi from project_2 
project_2 : 7fa627e77700 DEBUG Hi from project_2 
project_2 : 7fa627676700 DEBUG Hi from project_2 
project_2 : 7fa6293d0700 DEBUG Hello from main 
project_2 : 7fa627e77700 DEBUG Hi from project_2 

На самом деле, это не то, что я ожидал. Можете ли вы принести мне некоторое представление о том, что не так?

Answer 1

Части вашей проблемы идет от прохождения переменного объекта. Когда вы проходите args=(info,), вы передаете reference to an object (который вы позже модифицируете и переходите к следующему объекту), а не копию объекта. Передача одного и того же объекта на несколько потоков может стать опасной, учитывая возможность для race conditions

Во-первых, мы можем удалить наш ContextFilter. Мы добавляли их в глобальный регистратор и не отслеживали ничто в потоке.

import logging 
import threading 
import time 

logger = logging.getLogger() 
syslog = logging.StreamHandler() 
formatter = logging.Formatter('%(project)s : %(thread)x ' 
           '%(levelname)-8s ' 
           '%(message)s') 
syslog.setFormatter(formatter) 
logger.setLevel(logging.DEBUG) 
logger.addHandler(syslog) 

я нахожу в общем здании threading.Thread классов, чтобы быть более полезным для всех, кроме самых простых задач.

Этот класс имеет собственное состояние running и строит собственный собственный регистрационный адаптер с правильными данными extra.

class Worker(threading.Thread): 
    def __init__(self, info): 
     self.running=False 
     self.info=info 
     self.logger=logging.LoggerAdapter(logger, self.info) 
     super(Worker, self).__init__() 
    def start(self): 
     self.running=True 
     super(Worker, self).start() 
    def stop(self): 
     self.running=False 
    def run(self): 
     while self.running: 
      self.logger.debug('Hi from {}'.format(self.info['project'])) 
      time.sleep(0.5) 

И теперь нам нужно изменить несколько вещей. Нам нужно использовать собственный класс Worker.

Нам не нужно ничего делать с регистратором, класс будет управлять своим собственным LoggerAdapter.

Мы хотим, чтобы каждый раз мы создавали новый информационный объект, это так просто, мы можем просто передать его прямо в вызове функции ({'project': project}) без назначения переменной.

Нам необходимо убедиться, что мы передаем переменную project при регистрации в основном потоке. Возможно, это лучше сделать с другим LoggerAdapter.

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

def main(): 
    projects = ['project_1', 'project_2'] 
    threads = [] 
    for project in projects: 
     thread = Worker({'project': project}) 
     thread.start() 
     threads.append(thread) 
    while True: 
     try: 
      logging.debug('Hello from main', extra={'project':'main'}) 
      time.sleep(1.75) 
     except KeyboardInterrupt: 
      break 
    for t in threads: 
     t.stop() 
    for t in threads: 
     t.join() 

if __name__ == '__main__': 
    main() 

Это дает код результатов, такому в

project_1 : 7f4b44180700 DEBUG Hi from project_1 
project_2 : 7f4b4397f700 DEBUG Hi from project_2 
main : 7f4b45c8d700 DEBUG Hello from main 
project_1 : 7f4b44180700 DEBUG Hi from project_1 
project_2 : 7f4b4397f700 DEBUG Hi from project_2 
project_1 : 7f4b44180700 DEBUG Hi from project_1 
project_2 : 7f4b4397f700 DEBUG Hi from project_2 
project_1 : 7f4b44180700 DEBUG Hi from project_1 
project_2 : 7f4b4397f700 DEBUG Hi from project_2 
main : 7f4b45c8d700 DEBUG Hello from main 
project_1 : 7f4b44180700 DEBUG Hi from project_1 

Есть много способов, код может быть убрано и сделать более удобным для чтения, но это должно по крайней мере дать вам несколько отправных точек, чтобы учиться и начать экспериментировать. Когда вы узнаете больше о потоках, вы также должны прочитать около thread synchronization механизмов. Недавно я начал using Queues для связи между потоками, ведущий код, который легче отлаживать.

Answer 2

Что именно вы ожидали? Если вам интересно, почему «Привет из» не отображается имя проекта, попробуйте:

logging.debug('Hi from {}'.format(args['project'])) 

EDIT: ответить на Ваш комментарий, на самом деле вы получаете журналы из обоих потоков. Но ваш info объект является общим. Когда вы проходите args=(info,)), на самом деле вы передаете ссылку на объект info, а не на копию.

Итак, при первом запуске цикла for вы получаете info['project'] = "project_1", но во второй раз info['project'] перезаписывается «project_2».

Ваши рабочие потоки только читать значения из того же info Словаря ...