Como: Sincronizar um produtor e um thread de consumidor (guia de programação translation from VPE for Csharp)

Artigo
08/25/2010

O exemplo a seguir demonstra a sincronização de segmentos entre o thread principal e segmentos de trabalho dois usando o lock palavra-chave e o AutoResetEvent e ManualResetEvent classes. Para obter mais informações, consulte bloquear de demonstrativo (referência translation from VPE for Csharp).

O exemplo cria dois auxiliares ou segmentos de trabalho.Um thread produz elementos e as armazena em uma fila genérica que não é thread-safe.Para obter mais informações, consulte Queue<T>.O Outros thread consome itens dessa fila.Além disso, o thread principal periodicamente exibe o Sumário da fila, para que a fila é acessada por três segmentos.The lock palavra-chave é usada para sincronizar o acesso à fila para garantir que o estado da fila não está corrompido.

Além simplesmente impedindo o acesso simultâneo que usa o lock palavra-chave, sincronização adicional é proporcionada por dois objetos de evento. Um é usado para sinalizar a finalização dos segmentos de trabalho e a Outros é usada pelo thread de produtor para sinalizar para o thread de consumidor quando um novo item é adicionado à fila.Esses dois evento objetos são encapsulados em uma classe chamada SyncEvents. Isso permite que os eventos passam para os objetos que representam os threads de consumidor e o produtor com com facilidade.The SyncEvents classe é definida da seguinte forma:

public class SyncEvents
{
    public SyncEvents()
    {

        _newItemEvent = new AutoResetEvent(false);
        _exitThreadEvent = new ManualResetEvent(false);
        _eventArray = new WaitHandle[2];
        _eventArray[0] = _newItemEvent;
        _eventArray[1] = _exitThreadEvent;
    }

    public EventWaitHandle ExitThreadEvent
    {
        get { return _exitThreadEvent; }
    }
    public EventWaitHandle NewItemEvent
    {
        get { return _newItemEvent; }
    }
    public WaitHandle[] EventArray
    {
        get { return _eventArray; }
    }

    private EventWaitHandle _newItemEvent;
    private EventWaitHandle _exitThreadEvent;
    private WaitHandle[] _eventArray;
}

The AutoResetEventclasse é usada para o "novo item" evento porque você deseja que este evento para reiniciar automaticamente sempre que o thread de consumidor responde a este evento. Como alternativa, o ManualResetEventclasse é usada para o evento "sair" porque você deseja que vários threads para responder quando esse evento é sinalizado. Se você usou AutoResetEvent em vez disso, o evento seria reverter para um estado não sinalizado depois apenas um thread respondido o evento. O Outros thread não responderia e falhará nesse caso terminar.

The SyncEvents classe cria dois eventos e armazena-os de duas formas diferentes: sistema autônomo EventWaitHandle, que é a classe base para ambos AutoResetEvent e ManualResetEvente em uma matriz com base em WaitHandle. sistema autônomo você verá a discussão de thread de consumidor, esta matriz é necessária para que o thread de consumidor pode responder a qualquer um dos eventos.

Os threads de consumidor e o produtor são representados por classes denominadas Consumer e Producer. Ambos definir um método chamado ThreadRun. Esses métodos são usados sistema autônomo sistema autônomo pontos de entrada para o operador de segmentos que o Main método cria.

The ThreadRun método definido pela Producer classe parece com isto:

// Producer.ThreadRun
public void ThreadRun()
{
    int count = 0;
    Random r = new Random();
    while (!_syncEvents.ExitThreadEvent.WaitOne(0, false))
    {
        lock (((ICollection)_queue).SyncRoot)
        {
            while (_queue.Count < 20)
            {
                _queue.Enqueue(r.Next(0,100));
                _syncEvents.NewItemEvent.Set();
                count++;
            }
        }
    }
    Console.WriteLine("Producer thread: produced {0} items", count);
}

Esse método faz um loop até que o evento "sair thread" se torna sinalizado.O estado deste evento é testado com o WaitOne método, usando o ExitThreadEvent propriedade definida pela SyncEvents classe. Nesse caso, o estado do evento está selecionado sem bloquear o thread corrente como o primeiro argumento é usado com WaitOne é zero que indica que o método deve retornar imediatamente. If WaitOne Retorna true, e, em seguida, o evento em questão está sinalizado no momento. Nesse caso, a ThreadRun método retorna, que tem o efeito de encerrar o thread de trabalho executar este método.

Até que o evento "sair thread" está sinalizado, oProducer.ThreadStart método tenta manter 20 itens na fila. Um item é apenas um número inteiro entre 0 e 100.A coleção deve ser bloqueada antes de adicionar novos itens para impedir que o consumidor e o principais threads acessem a coleção ao mesmo time.Isso é concluído usando o lock palavra-chave. O argumento passado para lock é o SyncRoot campo expostos por meio do ICollection interface. Este campo é fornecido especificamente para a sincronização thread acesso.Acesso exclusivo para a coleção é concedido para todas as instruções contidas no bloco de código seguinte lock. Para cada novo item que adiciona o produtor para a fila, uma telefonar para o Setmétodo em "novo item" evento é feito. Isso sinaliza o thread de consumidor a surgir de seu estado suspenso para processar o novo item.

The Consumer objeto também define um método chamado ThreadRun. Como a versão do produtor de ThreadRun, esse método é executado por um operador thread criado pela Main método. No entanto, a versão do consumidor de ThreadStart deve responder a dois eventos. The Consumer.ThreadRun método parece com isto:

// Consumer.ThreadRun
public void ThreadRun()
{
    int count = 0;
    while (WaitHandle.WaitAny(_syncEvents.EventArray) != 1)
    {
        lock (((ICollection)_queue).SyncRoot)
        {
            int item = _queue.Dequeue();
        }
        count++;
    } 
    Console.WriteLine("Consumer Thread: consumed {0} items", count);
}

Esse método usa WaitAny para bloquear o thread de consumidor até que qualquer um dos identificadores de espera da matriz fornecida tornam-se signaled. Nesse caso, são dois identificadores no array, que finaliza os threads de trabalho e que indica que um novo item foi adicionado à coleção.WaitAny Retorna o índice do evento que tornou-se signaled. O "novo item" evento é a primeira no array, para que um índice de zero indica um novo item.Nesse caso, procure um índice de 1, o que indica o evento "sair de segmento", e isso é usado para determinar se esse método continua consumir itens.Se o "novo item" evento foi sinalizado, você obtém acesso exclusivo para a coleção comlock e consumir o novo item. Como este exemplo produz e consome milhares de itens, você não exiba cada item consumido.Em vez disso, use Main periodicamente para exibir o Sumário da fila, conforme será demonstrada.

The Main método começa criando a fila cujo Sumário será produzido e consumido e uma instância de SyncEvents, que você observou anteriormente:

Queue<int> queue = new Queue<int>();
SyncEvents syncEvents = new SyncEvents();

Em seguida, Main Configura o Producer e Consumer objetos para uso com segmentos de trabalho. Essa etapa não é, no entanto, criar ou iniciar os threads de trabalho real:

Producer producer = new Producer(queue, syncEvents);
Consumer consumer = new Consumer(queue, syncEvents);
Thread producerThread = new Thread(producer.ThreadRun);
Thread consumerThread = new Thread(consumer.ThreadRun);

Observe que a fila e o objeto de evento de sincronização são passadas para o Consumer e Producer threads sistema autônomo argumentos do construtor. Isso fornece os dois objetos que possuem os recursos compartilhados que precisam para realizar suas respectivas tarefas.Dois novos Thread objetos, em seguida, são criados, usando o ThreadRun método para cada objeto sistema autônomo um argumento. Cada thread de trabalho, quando é iniciado, usará este argumento sistema autônomo o ponto de entrada para o segmento.

Próxima Main Inicia a dois dos segmentos de trabalho com uma telefonar para o Start método, sistema autônomo este:

producerThread.Start();
consumerThread.Start();

Neste ponto, o operador novo dois thread s são criados e começar a execução assíncrono, independente do principal thread que está executando atualmente o Main método. Em fato, a próxima coisa Main faz é suspender o thread principal com uma telefonar para o Sleep método. O método suspende o thread em execução no momento para um determinado número de milissegundos.Ter uma vez esse intervalo decorrido, Main é reativado, no ponto em que ele exibe o Sumário da fila. Main Isso é repetida para quatro iterações, sistema autônomo este:

for (int i=0; i<4; i++)
{
    Thread.Sleep(2500);
    ShowQueueContents(queue);
}

Finalmente, Main sinaliza os threads de trabalho a finalização, chamando o Setmétodo de evento "sair thread" e chamadas de Join método em cada segmento de trabalho para bloquear o thread principal até que cada thread de trabalho responder ao evento e é encerrado.

Há um exemplo final de sincronização de segmentos: the ShowQueueContents método. Esse método, como os threads de consumidor e o producer utiliza lock Para obter acesso exclusivo a fila. Nesse caso, entretanto, o acesso exclusivo é muito importante, pois ShowQueueContents Enumera sobre todos da coleção. Para enumerar uma coleção é uma operação que está especialmente propensa a corrupção de dados por operações assíncrono porque envolve percorrer o Sumário de toda a coleção.

Observe que ShowQueueContents, porque ela é chamada pelo Main, é executado pelo thread principal. Isso significa que esse método quando ele obtém acesso exclusivo para a fila do item, bloqueia o produtor e consumidor threads do acesso à fila.ShowQueueContents bloqueia a fila e enumera o Sumário:

private static void ShowQueueContents(Queue<int> q)
{
    lock (((ICollection)q).SyncRoot)
    {
        foreach (int item in q)
        {
            Console.Write("{0} ", item);
        }
    }
    Console.WriteLine();
}

Segue o exemplo completo.

Exemplo

using System;
using System.Threading;
using System.Collections;
using System.Collections.Generic;

public class SyncEvents
{
    public SyncEvents()
    {

        _newItemEvent = new AutoResetEvent(false);
        _exitThreadEvent = new ManualResetEvent(false);
        _eventArray = new WaitHandle[2];
        _eventArray[0] = _newItemEvent;
        _eventArray[1] = _exitThreadEvent;
    }

    public EventWaitHandle ExitThreadEvent
    {
        get { return _exitThreadEvent; }
    }
    public EventWaitHandle NewItemEvent
    {
        get { return _newItemEvent; }
    }
    public WaitHandle[] EventArray
    {
        get { return _eventArray; }
    }

    private EventWaitHandle _newItemEvent;
    private EventWaitHandle _exitThreadEvent;
    private WaitHandle[] _eventArray;
}
public class Producer 
{
    public Producer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    // Producer.ThreadRun
    public void ThreadRun()
    {
        int count = 0;
        Random r = new Random();
        while (!_syncEvents.ExitThreadEvent.WaitOne(0, false))
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                while (_queue.Count < 20)
                {
                    _queue.Enqueue(r.Next(0,100));
                    _syncEvents.NewItemEvent.Set();
                    count++;
                }
            }
        }
        Console.WriteLine("Producer thread: produced {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

public class Consumer
{
    public Consumer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    // Consumer.ThreadRun
    public void ThreadRun()
    {
        int count = 0;
        while (WaitHandle.WaitAny(_syncEvents.EventArray) != 1)
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                int item = _queue.Dequeue();
            }
            count++;
        } 
        Console.WriteLine("Consumer Thread: consumed {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

public class ThreadSyncSample
{
    private static void ShowQueueContents(Queue<int> q)
    {
        lock (((ICollection)q).SyncRoot)
        {
            foreach (int item in q)
            {
                Console.Write("{0} ", item);
            }
        }
        Console.WriteLine();
    }

    static void Main()
    {
        Queue<int> queue = new Queue<int>();
        SyncEvents syncEvents = new SyncEvents();

        Console.WriteLine("Configuring worker threads...");
        Producer producer = new Producer(queue, syncEvents);
        Consumer consumer = new Consumer(queue, syncEvents);
        Thread producerThread = new Thread(producer.ThreadRun);
        Thread consumerThread = new Thread(consumer.ThreadRun);

        Console.WriteLine("Launching producer and consumer threads...");        
        producerThread.Start();
        consumerThread.Start();

        for (int i=0; i<4; i++)
        {
            Thread.Sleep(2500);
            ShowQueueContents(queue);
        }

        Console.WriteLine("Signaling threads to terminate...");
        syncEvents.ExitThreadEvent.Set();

        producerThread.Join();
        consumerThread.Join();
    }

}

Configuring worker threads... Launching producer and consumer threads... 22 92 64 70 13 59 9 2 43 52 91 98 50 96 46 22 40 94 24 87 79 54 5 39 21 29 77 77 1 68 69 81 4 75 43 70 87 72 59 0 69 98 54 92 16 84 61 30 45 50 17 86 16 59 20 73 43 21 38 46 84 59 11 87 77 5 53 65 7 16 66 26 79 74 26 37 56 92 Signalling threads to terminate... Consumer Thread: consumed 1053771 items Producer thread: produced 1053791 items

Consulte também

Partilhar via

Como: Sincronizar um produtor e um thread de consumidor (guia de programação translation from VPE for Csharp)

Exemplo

Consulte também

Tarefas

Conceitos

Referência

Recursos adicionais