时间戳

首页 > 解决方案 > 线程 - 具有 Counter 对象的生产者/消费者

python - 线程 - 具有 Counter 对象的生产者/消费者

问题描述

我在下面编写了一个代码片段，旨在实现以下目标。我有一个在所有消费者之间共享的计数器对象，这些消费者在完成他们的任务后，递增计数器并耐心等待。这个想法是创建一个新的 IP 地址，然后其他线程可以在他们的新任务中使用，这将由生产者完成，生产者还检查计数器是否具有某些值，如果是，它将创建一个新的ip 并通知所有人，否则通知所有人而不创建新 ip。但由于某种原因，我从来没有收到过NOT SETTING A NEW IP ADDRESS消息。有人能告诉我为什么吗？

非常感谢：

import logging
import threading
import time
import random

class Counter(object):

    def __init__(self, start=0):
        self.lock = threading.RLock()
        self.value = start

    def increment(self):
        logging.debug('Waiting for lock')
        with self.lock:
            logging.debug('Acquired lock. Current counter value: {}'.format(self.value + 1))
            self.value = self.value + 1

def consumer(cond, counter):
    """wait for the condition and use the resource"""
    logging.debug('Starting consumer thread')
    while True:
        time_sleep = random.randint(5, 10) / 5
        time.sleep(time_sleep)
        with cond:
            counter.increment()
            logging.debug('Resource is available to consumer. Doing some werk on counter {}'.format(counter.value))
            logging.debug('Done some werk. Waiting for other threads to finish their werk and for producer to signal continuation.')
            cond.wait()


def producer(cond, counter):
    """set up the resource to be used by the consumer"""
    logging.debug('Starting producer thread')

    for i in range(4):
        logging.debug('Producer sleeping for 3 seconds')
        time.sleep(3)
        with cond:
            if counter.value % 2 == 0:
                logging.debug('Setting a new ip address')
                cond.notifyAll()
            if counter.value % 2 != 0:
                logging.debug('NOT SETTING A NEW IP ADDRESS')
                cond.notifyAll()


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

condition = threading.Condition()
c = Counter()

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

c1.start()
c2.start()

p.start()

标签： pythonpython-3.xmultithreading

解决方案

尝试在生产者线程中添加锁。

其他线程应该等待生产者完成其任务。

这可以防止计数器在生产者线程执行期间被其他线程覆盖。

def producer(cond, counter):
    """set up the resource to be used by the consumer"""
    logging.debug('Starting producer thread')
    with self.lock
    for i in range(4):
        logging.debug('Producer sleeping for 3 seconds')
        time.sleep(3)
        with cond:
            if counter.value % 2 == 0:
                logging.debug('Setting a new ip address')
                cond.notifyAll()
            if counter.value % 2 != 0:
                logging.debug('NOT SETTING A NEW IP ADDRESS')
                cond.notifyAll()

推荐阅读