№34 потоки и многопоточность / для начинающих

Синхронизация потоков

Другой способ синхронизации потоков – объект Condition. Поскольку Condition использует Lock, его можно привязать к общему ресурсу. Это позволяет потокам ожидать обновления ресурса.

В приведенном ниже примере поток consumer() будет ждать, пока не будет установлено Condition, прежде чем продолжить. Поток producer() отвечает за установку Condition и уведомление других потоков о том, что они могут продолжить выполнение.

import logging
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s (%(threadName)-2s) %(message)s',
                    )

def consumer(cond):
    """wait for the condition and use the resource"""
    logging.debug('Starting consumer thread')
    t = threading.currentThread()
    with cond:
        cond.wait()
        logging.debug('Resource is available to consumer')

def producer(cond):
    """set up the resource to be used by the consumer"""
    logging.debug('Starting producer thread')
    with cond:
        logging.debug('Making resource available')
        cond.notifyAll()

condition = threading.Condition()
c1 = threading.Thread(name='c1', target=consumer, args=(condition,))
c2 = threading.Thread(name='c2', target=consumer, args=(condition,))
p = threading.Thread(name='p', target=producer, args=(condition,))

c1.start()
time.sleep(2)
c2.start()
time.sleep(2)
p.start()

Потоки используют with для блокировки, связанной с Condition. Использование методов acquire() и release()в явном виде также работает.

$ python threading_condition.py

2013-02-21 06:37:49,549 (c1) Starting consumer thread
2013-02-21 06:37:51,550 (c2) Starting consumer thread
2013-02-21 06:37:53,551 (p ) Starting producer thread
2013-02-21 06:37:53,552 (p ) Making resource available
2013-02-21 06:37:53,552 (c2) Resource is available to consumer
2013-02-21 06:37:53,553 (c1) Resource is available to consumer

2.3 Остановка и прерывание потоков

Для остановки потока в Java версии 1 использовался метод stop(). Однако в версии Java 1.1 этот метод сделали deprecated, потому что использование метода stop() не гарантирует корректного завершения работы потока и стабильной работы программы в целом. Поэтому при написании программ использовать его настоятельно не рекомендуется. 

Вместо метода stop() следует использовать метод interrupt(). В отличие от метода stop(), который принудительно останавливал поток, метод interrupt() предлагает потоку остановить свое выполнение путем установки флага interrupted в true внутри потока. Этот флаг отображает статус прерывания и имеет начальное значение false. Когда поток прерывается другим потоком, происходит одно из двух:

• Если поток ожидает выполнения прерываемого метода блокирования, таких как Thread.sleep(), Thread.join() или Object.wait(), то ожидание прерывается и метод генерирует InterruptedException. Флаг interrupted устанавливается в false.
• Флаг interrupted устанавливается в true.

Есть три метода для работы с прерыванием потока:

Листинг 5:

public class Thread {     public void interrupt() { … }     public boolean isInterrupted() { … }     public static boolean interrupted() { … } … }

• Работа метода interrupt() описана выше.
• isInterrupted() возвращает значение флага и не изменяет его.
• interrupted() возвращает значение флага и устанавливает его значение в false. Если флаг interrupted установлен в true и вызывается этот метод, то первый раз метод вернет true, а последующие вызовы вернут false.

Существуют два вида операций: блокирующие и неблокирующие. Неблокирующие операции не приостанавливают выполнения потока. К блокирующим операциям можно отнести вызовы методов sleep(), wait(), join() и, например, некоторые методы класса Socket. Если поток был прерван, пока он выполнял неблокирующие вычисления, они не будут прерваны незамедлительно. Однако поток уже отмечен как прерванный, поэтому любая следующая блокирующая операция немедленно прервется и выбросит InterruptedException. 

Для обработки прерывания в потоке, который не использует блокирующие операции, следует добавить проверку флага interrupted пример в листинге 6.

Листинг 6:

public void run() {     while (Thread.currentThread().isInterrupted()) {         someHeavyComputations();     }}

Обработка InterruptedException.

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

Первый способ обработки InterruptedException — объявление этого исключения в вышестоящем методе. Также при перехвате метода InterruptedException можно произвести какие-то действия (например, очистку ресурсов или переменных) и повторно пробросить InterruptedException. 

Во втором случае, когда InterruptedException объявить невозможно, при генерации и перехвате InterruptedException флаг interrupted устанавливается в false, и вызывающие методы не увидят, что было совершено прерывание потока. Однако можно восстановить флаг прерывания, вызвав Thread.currentThread().interrupt() при обработке прерывания. 

Также восстановление флага interrupted может быть полезным, когда первый поток имеет ссылку на второй поток, и первый хочет узнать состояние флага второго. 

Стоит внимательно следить за обработкой этого исключения когда код выполняется в threadpool. InterruptedException может быть «интересен»  не только коду, но и потоку, который выполняет этот код. 

Листинг 7:

try {     Object o = queue.take(); } catch InterruptedException e) { }

Этот код некорректен, потому что поглощает (swallows) прерывание. Если этот код выполняется в tread pool, то воркер (thread pool worker) tread pool`а должен завершить исполнение, но этого не произойдёт, потому что исключение будет поглощено, и флаг будет сброшен. Корректный код будет выглядеть так:

Листинг 8:

try {     Object o = queue.take(); } catch InterruptedException e) {     Thread.currentThread().interrupt(); }

В блоке catch происходит перехват исключения и установка флага в true. 

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

Рекомендации для библиотек классов

При разработке библиотек классов для многопоточности необходимо учитывать следующие рекомендации.

• Старайтесь не создавать потребность в синхронизации. Особенно это относится к коду, который используется наиболее часто. Например, алгоритм можно скорректировать таким образом, чтобы он допускал конфликты, а не устранял их. Ненужная синхронизация снижает производительность и может привести к взаимоблокировке и конфликтам.

• Сделайте статические данные ( в Visual Basic) по умолчанию потокобезопасными.

• Данные экземпляров не должны быть потокобезопасными по умолчанию. Добавление блокировок для создания потокобезопасного кода снижает производительность, увеличивает конфликт блокировки и создает условия для возникновения взаимоблокировок. В обычных моделях приложений пользовательский код одновременно выполняется только одним потоком, что уменьшает необходимость потокобезопасности. По этой причине библиотеки классов .NET не являются потокобезопасными по умолчанию.

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

Взаимоблокировки и состояние гонки

Многопоточность позволяет решить проблемы с пропускной способностью и скоростью реагирования, но при этом возникают новые проблемы: взаимоблокировки и конфликты.

Взаимоблокировки

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

Многие методы классов управляемых потоков предоставляют значения времени ожидания для обнаружения взаимоблокировок. Например, следующий код пытается получить блокировку для объекта с именем . Если блокировка не будет получена в течение 300 миллисекунд, Monitor.TryEnter возвратит .

Состояние гонки

Конфликт — это ошибка, которая возникает, когда результат программы зависит от того, какой из двух или более потоков первым достигнет определенного блока кода. Выполнение программы часто дает различные результаты, и предсказать результат выполнения конкретного запуска невозможно.

Простой пример состояния гонки — увеличение поля. Предположим, что класс содержит закрытое поле static (Shared в Visual Basic), которое увеличивается всякий раз при создании класса с помощью кода, например (в C#) или (в Visual Basic). Для этой операции необходимо загрузить значение из в регистр, увеличить или уменьшить это значение и сохранить его в .

В многопоточных приложениях поток, загружающий и увеличивающий значение, может быть вытеснен другим потоком, который выполняет все три эти действия; если первый поток возобновляет выполнение и сохраняет его значение, он переопределяет , не принимая во внимание тот факт, что в промежутке значение изменилось. Конкретно этого состояния гонки можно легко избежать, применяя методы класса Interlocked, например Interlocked.Increment

Сведения о других технологиях синхронизации данных между несколькими потоками см. в разделе Синхронизация данных для многопоточности

Конкретно этого состояния гонки можно легко избежать, применяя методы класса Interlocked, например Interlocked.Increment. Сведения о других технологиях синхронизации данных между несколькими потоками см. в разделе Синхронизация данных для многопоточности.

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

Способы реализации параллельных вычислений в программах на Python.

Что такое параллелизм?

Параллелизм дает возможность работать над несколькими вычислениями одновременно в одной программе. Такого поведения в Python можно добиться несколькими способами:

• Используя многопоточность , позволяя нескольким потокам работать по очереди.
• Используя несколько ядер процессора . Делать сразу несколько вычислений, используя несколько ядер процессора. Это и называется параллелизмом.
• Используя асинхронный ввод-вывод с модулем . Запуская какую то задачу, продолжать делать другие вычисления, вместо ожидания ответа от сетевого подключения или от операций чтения/записи.

Разница между потоками и процессами.

Поток — это независимая последовательность выполнения каких то вычислений. Поток делит выделенную память ядру процессора, а так же его процессорное время со всеми другими потоками, которые создаются программой в рамках одного ядра процессора. Программы на языке Python имеют, по умолчанию, один основной поток. Можно создать их больше и позволить Python переключаться между ними. Это переключение происходит очень быстро и кажется, что они работают параллельно.

Понятие процесс в — представляет собой так же независимую последовательность выполнения вычислений. В отличие от потоков , процесс имеет собственное ядро и следовательно выделенную ему память, которое не используется совместно с другими процессами. Процесс может клонировать себя, создавая два или более экземпляра в одном ядре процессора.

Асинхронный ввод-вывод не является ни потоковым (), ни многопроцессорным (). По сути, это однопоточная, однопроцессная парадигма и не относится к параллельным вычислениям.

У Python есть одна особенность, которая усложняет параллельное выполнение кода. Она называется GIL, сокращенно от Global Interpreter Lock. GIL гарантирует, что в любой момент времени работает только один поток. Из этого следует, что с потоками невозможно использовать несколько ядер процессора.

GIL был введен в Python потому, что управление памятью CPython не является потокобезопасным. Имея такую блокировку Python может быть уверен, что никогда не будет условий гонки.

Что такое условия гонки и потокобезопасность?

• Состояние гонки возникает, когда несколько потоков могут одновременно получать доступ к общей структуре данных или местоположению в памяти и изменять их, в следствии чего могут произойти непредсказуемые вещи…

  Пример из жизни: если два пользователя одновременно редактируют один и тот же документ онлайн и второй пользователь сохранит данные в базу, то перезапишет работу первого пользователя. Чтобы избежать условий гонки, необходимо заставить второго пользователя ждать, пока первый закончит работу с документом и только после этого разрешить второму пользователю открыть и начать редактировать документ.

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

Алгоритм планирования доступа потоков к общим данным.

Как уже говорилось, потоки используют одну и ту же выделенную память. Когда несколько потоков работают одновременно, то нельзя угадать порядок, в котором потоки будут обращаются к общим данным. Результат доступа к совместно используемым данным зависит от алгоритма планирования. который решает, какой поток и когда запускать. Если такого алгоритма нет, то конечные данные могут быть не такими как ожидаешь.

Например, есть общая переменная . Теперь предположим, что есть два потока, и . Они выполняют следующие операции:

a = 2

# функция 1 потока
def thread_one():
    global a
    a = a + 2

# функция 2 потока
def thread_two():
    global a
    a = a * 3

Если поток получит доступ к общей переменной первым и вторым, то результат будет 12:

1. 2 + 2 = 4;
2. 4 * 3 = 12.

или наоборот, сначала запустится , а затем , то мы получим другой результат:

1. 2 * 3 = 6;
2. 6 + 2 = 8.

Таким образом очевидно, что порядок выполнения операций потоками имеет значение

Без алгоритмов планирования доступа потоков к общим данным такие ошибки очень трудно найти и произвести отладку. Кроме того, они, как правило, происходят случайным образом, вызывая беспорядочное и непредсказуемое поведение.

Есть еще худший вариант развития событий, который может произойти без встроенной в Python блокировки потоков GIL . Например, если оба потока начинают читать глобальную переменную одновременно… Оба потока увидят, что , а дальше, в зависимости от того какой поток произведет вычисления последним, в конечном итоге и будет равна переменная (4 или 6). Не то, что ожидалось!

Комментарии

Многие приложения создают потоки, которые тратят много времени на спящий режим, ожидая возникновения события. Другие потоки могут войти в спящее состояние только для периодического опроса на наличие изменений или сведений о состоянии обновления. Пул потоков позволяет более эффективно использовать потоки, предоставляя приложению пул рабочих потоков, управляемых системой. Примеры операций, использующих потоки пула потоков, включают следующее.

• При создании Task Task<TResult> объекта или для асинхронного выполнения некоторой задачи по умолчанию задача запланирована на выполнение в потоке пула потоков.

• Асинхронные таймеры используют пул потоков. Потоки из пула потоков выполняют обратные вызовы из System.Threading.Timer класса и инициируют события из System.Timers.Timer класса.

• При использовании зарегистрированных дескрипторов ожидания системный поток отслеживает состояние дескрипторов ожидания. По завершении операции ожидания рабочий поток из пула потоков выполняет соответствующую функцию обратного вызова.

• При вызове QueueUserWorkItem метода для постановки в очередь метода для выполнения в потоке пула потоков. Это можно сделать, передав методу WaitCallback делегат. Делегат имеет сигнатуру

  где — это объект, содержащий данные, которые будут использоваться делегатом. Фактические данные могут быть переданы делегату путем вызова метода.

Примечание

Потоки в пуле управляемых потоков являются фоновыми потоками. То есть их IsBackground свойства имеют значения . Это означает, что ThreadPool поток не будет работать с приложением после выхода из всех основных потоков.

Важно!

Когда пул потоков повторно использует поток, он не очищает данные в локальном хранилище потока или в полях, помеченных ThreadStaticAttribute атрибутом. Таким образом, когда метод проверяет локальное хранилище потока или поля, помеченные ThreadStaticAttribute атрибутом, найденные значения могут остаться в более раннем использовании потока пула потоков.

Кроме того, можно ставить в очередь рабочие элементы, не связанные с операцией ожидания, в пул потоков. Чтобы запросить обработку рабочего элемента потоком в пуле потоков, вызовите QueueUserWorkItem метод. Этот метод принимает в качестве параметра ссылку на метод или делегат, который будет вызываться потоком, выбранным из пула потоков. Невозможно отменить рабочий элемент после его постановки в очередь.

Таймеры очереди и зарегистрированные операции ожидания также используют пул потоков. Функции обратного вызова помещаются в очередь пула потоков.

Для каждого процесса существует один пул потоков. Начиная с .NET Framework 4, размер пула потоков по умолчанию для какого-либо процесса зависит от нескольких факторов, таких как размер виртуального адресного пространства. Процесс может вызвать метод GetMaxThreads для определения количества потоков. Число потоков в пуле потоков можно изменить с помощью SetMaxThreads метода. Каждый поток использует размер стека по умолчанию и выполняется с приоритетом по умолчанию.

Примечание

неуправляемый код, в котором размещается платформа .NET Framework, может изменять размер пула потоков с помощью функции, определенной в файле mscoree. h.

Пул потоков предоставляет новые рабочие потоки или потоки завершения ввода-вывода по запросу, пока не достигнет максимального значения для каждой категории. При достижении максимального значения пул потоков может создавать дополнительные потоки в этой категории или ожидать завершения некоторых задач. Начиная с .NET Framework 4, пул потоков создает и уничтожает рабочие потоки в целях оптимизации пропускной способности, которая определяется как количество задач, выполняемых в единицу времени. Слишком малое количество потоков может препятствовать оптимальному использованию доступных ресурсов, тогда как слишком большое их количество может усиливать конкуренцию за ресурсы.

Примечание

Если потребность низкая, фактическое количество потоков из пула потоков может быть ниже минимальных значений.

Для получения этих минимальных значений можно использовать метод GetMinThreads.

Внимание!

SetMinThreadsДля увеличения минимального числа потоков можно использовать метод. Однако необоснованное увеличение этих значений может привести к снижению производительности. Если одновременно запускается слишком много задач, все они могут выполняться слишком медленно. В большинстве случаев пул потоков работает наилучшим образом, если он использует собственный алгоритм выделения потоков.

Ограничение одновременного доступа к ресурсам

Как разрешить доступ к ресурсу нескольким worker одновременно, но при этом ограничить их количество. Например, пул соединений может поддерживать фиксированное число одновременных подключений, или сетевое приложение может поддерживать фиксированное количество одновременных загрузок. Semaphore является одним из способов управления соединениями.

import logging
import random
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s (%(threadName)-2s) %(message)s',
                    )

class ActivePool(object):
    def __init__(self):
        super(ActivePool, self).__init__()
        self.active = []
        self.lock = threading.Lock()
    def makeActive(self, name):
        with self.lock:
            self.active.append(name)
            logging.debug('Running: %s', self.active)
    def makeInactive(self, name):
        with self.lock:
            self.active.remove(name)
            logging.debug('Running: %s', self.active)

def worker(s, pool):
    logging.debug('Waiting to join the pool')
    with s:
        name = threading.currentThread().getName()
        pool.makeActive(name)
        time.sleep(0.1)
        pool.makeInactive(name)

pool = ActivePool()
s = threading.Semaphore(2)
for i in range(4):
    t = threading.Thread(target=worker, name=str(i), args=(s, pool))
    t.start()

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

$ python threading_semaphore.py

2013-02-21 06:37:53,629 (0 ) Waiting to join the pool
2013-02-21 06:37:53,629 (1 ) Waiting to join the pool
2013-02-21 06:37:53,629 (0 ) Running: 
2013-02-21 06:37:53,629 (2 ) Waiting to join the pool
2013-02-21 06:37:53,630 (3 ) Waiting to join the pool
2013-02-21 06:37:53,630 (1 ) Running: 
2013-02-21 06:37:53,730 (0 ) Running: 
2013-02-21 06:37:53,731 (2 ) Running: 
2013-02-21 06:37:53,731 (1 ) Running: 
2013-02-21 06:37:53,732 (3 ) Running: 
2013-02-21 06:37:53,831 (2 ) Running: 
2013-02-21 06:37:53,833 (3 ) Running: []

Блокировки как менеджеры контекста

Блокировки реализуют API context manager и совместимы с оператором with. Использование оператора with позволяет обойтись без блокировки.

import threading
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-10s) %(message)s',
                    )

def worker_with(lock):
    with lock:
        logging.debug('Lock acquired via with')
        
def worker_no_with(lock):
    lock.acquire()
    try:
        logging.debug('Lock acquired directly')
    finally:
        lock.release()

lock = threading.Lock()
w = threading.Thread(target=worker_with, args=(lock,))
nw = threading.Thread(target=worker_no_with, args=(lock,))

w.start()
nw.start()

Функции worker_with() и worker_no_with() управляют блокировкой эквивалентными способами.

$ python threading_lock_with.py

(Thread-1  ) Lock acquired via with
(Thread-2  ) Lock acquired directly

Определение текущего потока

Использование аргументов для идентификации потока является трудоемким процессом. Каждый экземпляр Thread имеет имя со значением, присваиваемым по умолчанию. Оно может быть изменено, когда создается поток.

Именование потоков полезно в серверных процессах с несколькими служебными потоками, обрабатывающими различные операции.

import threading
import time

def worker():
    print threading.currentThread().getName(), 'Starting'
    time.sleep(2)
    print threading.currentThread().getName(), 'Exiting'

def my_service():
    print threading.currentThread().getName(), 'Starting'
    time.sleep(3)
    print threading.currentThread().getName(), 'Exiting'

t = threading.Thread(name='my_service', target=my_service)
w = threading.Thread(name='worker', target=worker)
w2 = threading.Thread(target=worker) # используем имя по умолчанию

w.start()
w2.start()
t.start()

Программа выводит имя текущего потока в каждой строке. «Thread-1» — это безымянный поток w2.

$ python -u threading_names.py

worker Thread-1 Starting
my_service Starting
Starting
Thread-1worker Exiting
 Exiting
my_service Exiting

Большинство программ не используют print для отладки. Модуль logging поддерживает добавление имени потока в каждое сообщение журнала с помощью % (threadName)s. Включение имен потоков в журнал облегчает отслеживание этих сообщений.

import logging
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format=' (%(threadName)-10s) %(message)s',
                    )

def worker():
    logging.debug('Starting')
    time.sleep(2)
    logging.debug('Exiting')

def my_service():
    logging.debug('Starting')
    time.sleep(3)
    logging.debug('Exiting')

t = threading.Thread(name='my_service', target=my_service)
w = threading.Thread(name='worker', target=worker)
w2 = threading.Thread(target=worker) # use default name

w.start()
w2.start()
t.start()

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

$ python threading_names_log.py

 (worker    ) Starting
 (Thread-1  ) Starting
 (my_service) Starting
 (worker    ) Exiting
 (Thread-1  ) Exiting
 (my_service) Exiting

2.2 Свойства потоков, запуск потоков, присоединение других потоков

Все методы программы выполняются в каком-либо потоке. Поток, который вызывает метод main, является главным потоком приложения и имеет имя main. 

В Java поток представлен классом Thread. Создать и запустить поток можно двумя способами: 

1) Создать наследника от класса Thread и переопределить метод run().

Листинг 1:

public class MyThread extends Thread {     public void run() {         long sum = 0;         for (int i = 0; i < 1000; ++i) {             sum += i;         }         System.out.println(sum);     } }

MyThread t = new MyThread();

2) Реализовать интерфейс Runnable и передать объект полученного класса в конструктор класса Thread.

Листинг 2:

Runnable r = new MyRunnable() { () ->     System.out.println(“Hello!”); }

Thread t = new Thread(r);

Для запуска потока необходимо использовать метод Thread.start().  Если вызвать метод run(), то он выполнится в вызывающем потоке:

Листинг 3:

Thread t = new Thread(r);

t.run(); //код r выполняется в текущем потоке

t.start(); //код r выполняется в новом потоке

Не следует запускать поток из конструктора класса. Некоторые фреймворки, такие как Spring, создают динамические подклассы для поддержки перехвата методов. В конечном счете, мы получим два потока, запущенных из двух экземпляров. 

Объект текущего потока можно получить, вызвав статический метод: Thread.currentThread().

Имена потокам можно задавать через метод setName() или через параметр конструктора. Рекомендуется давать потокам осмысленные имена, это пригодится при отладке. Не рекомендуется давать потокам одинаковые имена, хотя имена потоков не валидируются JVM. 

Стандартный формат имен потоков, которые были созданы одиночно — thread-N, где N порядковый номер потока. Для пула потоков, стандартное наименование — pool-N-thread-M, где N обозначает последовательный номер пула (каждый раз, когда вы создаете новый пул, глобальный счетчик N увеличивается), а M — порядковый номер потока в пуле.

У потоков есть приоритет, который можно задать целым числом от 1 до 10. Чем больше число, тем выше приоритет потока. Поток main имеет приоритет 5. А приоритет новых потоков равен приоритету потока-родителя, его можно изменить при помощи метода setPriority(int). Поток с большим приоритетом будет иметь больше процессорного времени на выполнение. Если два потока имеют одинаковый приоритет, то решение о том, какой из них будет выполняться первым, зависит от алгоритма планировщика: (Round-Robin, First Come First Serve).

Есть несколько констант для приоритета потоков:

• Thread.MIN_PRIORITY — минимальный приоритет, значение 1;
• Thread.NORM_PRIORITY — приоритет по умолчанию, значение 5;
• Thread.MAX_PRIORITY — максимальный приоритет потока, значение 10.

Листинг 4:

public class Main {     public static void main(String[] args) {         System.out.println(Thread.currentThread().getName());         Thread.currentThread().setPriority(8);         Thread thread = new Thread() {             public void run() {                 Thread.currentThread().setName(«My name»);                 System.out.println(Thread.currentThread().getName());                 System.out.println(Thread.currentThread().getPriority());             }         };         thread.start();     } }

В Java есть такое понятие, как поток-демон. Работа JVM заканчивается, когда закончил выполняться последний поток не-демон, несмотря на работающие потоки-демоны. Для работы с этим свойством существуют два метода: setDaemon() и isDaemon(). 

Класс ThreadGroup. Все потоки находятся в группах, представленных экземплярами класса ThreadGroup. Группа указывается при создании потока. Если группа не была указана, то поток помещается в ту же группу, в которой находится поток-родитель. Методы activeCount() и enumerate() возвращают, соответственно, количество и полный список всех активных потоков в группе. 

Способы приостановления выполнения потока на указанное количество времени: Thread.sleep(long millis) и TimeUnit.<UNIT>.sleep(long timeout). Они приостанавливают выполнение текущего потока на указанный период времени. Вызов методов требует обработки исключения InterruptedException. 

Нестатический метод join() позволяет одному потоку дождаться выполнения другого. Если текущий поток t1 вызывает у другого потока t2h2t2.join(), то поток th2 останавливается до тех пор, пока поток t2 не завершит свою работу. Вызвать метод join() можно также и с аргументом, указывающим лимит времени ожидания (в миллисекундах или в миллисекундах с нано секундами). Если целевой поток t2 не закончит работу за указанный период времени, метод join() все равно вернет управление инициатору t1. 

Характеристики пула потоков

Потоки из пула являются фоновыми. Для каждого потока используется размер стека по умолчанию, поток запускается с приоритетом по умолчанию и находится в многопотоковом подразделении. Когда поток в пуле завершает свою задачу, он возвращается в очередь потоков в состоянии ожидания. С этого момента его можно использовать вновь. Повторное использование позволяет приложениям избежать дополнительных затрат на создание новых потоков для каждой задачи.

Для каждого процесса существует только один пул потоков.

Исключения в потоках из пула потоков

Необработанные исключения в потоках из пула приводят к завершению процесса. Есть три исключения из этого правила:

• Исключение System.Threading.ThreadAbortException возникает в потоке пула вследствие вызова Thread.Abort.
• Исключение System.AppDomainUnloadedException возникает в потоке пула вследствие выгрузки домена приложения.
• Среда CLR или процесс ведущего приложения прерывает выполнение потока.

См. дополнительные сведения об исключениях в управляемых потоках.

Максимальное число потоков в пуле потоков

Число операций, которое можно поставить в очередь в пуле потоков, ограничено только доступной памятью. Однако пул потоков имеет ограничение на число потоков, которое можно активировать в процессе одновременно. Если все потоки в пуле заняты, дополнительные рабочие элементы помещаются в очередь и ожидают их освобождения. Размер по умолчанию пула потоков для процесса зависит от нескольких факторов, таких как размер виртуального адресного пространства. Процесс может вызвать метод ThreadPool.GetMaxThreads для определения количества потоков.

Вы можете управлять максимальным количеством потоков с помощью методов ThreadPool.GetMaxThreads и ThreadPool.SetMaxThreads.

Примечание

В коде, содержащем среду CLR, этот размер можно задать с помощью метода .

Минимальные значения пула потоков

Пул потоков предоставляет новые рабочие потоки или потоки завершения ввода-вывода по запросу, пока не будет достигнут заданный минимум для каждой категории. Для получения этих минимальных значений можно использовать метод ThreadPool.GetMinThreads.

Примечание

Если потребность низкая, фактическое количество потоков из пула потоков может быть ниже минимальных значений.

При достижении минимума пул потоков может создавать дополнительные потоки или ожидать завершения некоторых задач. Пул потоков создает и уничтожает рабочие потоки в целях оптимизации пропускной способности, которая определяется как количество задач, завершаемых за единицу времени. Слишком малое количество потоков может препятствовать оптимальному использованию доступных ресурсов, тогда как слишком большое их количество может усиливать конкуренцию за ресурсы.

Внимание!

Для увеличения минимального количества бездействующих потоков можно использовать метод ThreadPool.SetMinThreads. Однако необоснованное увеличение этих значений может привести к снижению производительности. Если одновременно запускается слишком много задач, все они могут выполняться слишком медленно. В большинстве случаев пул потоков работает наилучшим образом, если он использует собственный алгоритм выделения потоков.

Объекты потоков

Самый простой способ использовать поток — создать его с помощью целевой функции и запустить с помощью метода start().

import threading

def worker():
    """thread worker function"""
    print 'Worker'
    return

threads = []
for i in range(5):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

Результат работы программы – пять строк со строкой «Worker»:

$ python threading_simple.py

Worker
Worker
Worker
Worker
Worker

В приведенном ниже примере в качестве аргумента потоку передается число для вывода.

import threading

def worker(num):
    """thread worker function"""
    print 'Worker: %s' % num
    return

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

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

$ python -u threading_simpleargs.py

Worker: 0
Worker: 1
Worker: 2
Worker: 3
Worker: 4

Остановка потока

Бывают ситуации, когда требуется остановить поток, который работает в фоне. Допустим у нас поток у которого в функции run бесконечный цикл. В основной программе нам нужно его остановить. Тут самое простое — это создать некую переменную stop:

• В бесконечном цикле делать постоянно её проверку и если она True, то завершать его.
• Не использовать функции, которые могут блокировать выполнение на длительное время. Всегда использовать timeout.

Вот пример такой программы:

import threading 
stop = False
def myfunc():
    global stop
    while stop == False:
        pass
thr1 = threading.Thread(target = myfunc) 
thr1.start() 
stop = True
while thr1.is_alive() == True: 
    pass
print('поток завершился')

Здесь используем глобальную переменную stop. Когда нам нужно остановить поток, мы ей присваиваем значение True, а дальше просто ждём его завершения.