阻塞队列 | 有界缓冲区 | 生产者-消费者

1public class BlockingQueue
2{
3    private int[] items;
4    private int maxSize;
5    private int currentSize;
6    private int tail;
7    private int head;
8
9    public BlockingQueue(int size)
10    {
11        this.maxSize = size;
12        this.currentSize = 0;
13        this.items = new int[size];
14        this.tail = 0;
15        this.head = 0;
16    }
17
18    private void enqueue()
19    {
20    }
21
22    private int dequeue()
23    {
24    }
25}

1mutex.WaitOne();
2while (currentSize == maxSize)
3{
4    // 释放互斥锁，让其他线程有机会获取
5    mutex.ReleaseMutex();
6
7    // 重新获取互斥锁后再检查条件
8    mutex.WaitOne();
9}
10
11if (tail == maxSize) tail = 0;
12
13items[tail] = data;
14currentSize++;
15tail++;
16mutex.ReleaseMutex();

1int result;
2mutex.WaitOne();
3
4while (currentSize == 0)
5{
6    mutex.ReleaseMutex();
7    mutex.WaitOne();
8}
9
10if (head == maxSize) head = 0;
11
12currentSize--;
13result = items[head];
14head++;
15mutex.ReleaseMutex();
16return result;

1public class BlockingQueue
2{
3    private int[] items;
4    private int maxSize;
5    private int currentSize;
6    private int tail;
7    private int head;
8    private Mutex mutex;
9
10    public BlockingQueue(int size)
11    {
12        this.maxSize = size;
13        this.currentSize = 0;
14        this.tail = 0;
15        this.head = 0;
16        this.items = new int[size];
17        this.mutex = new Mutex();
18    }
19
20    public int dequeue()
21    {
22        int result;
23        mutex.WaitOne();
24
25        while (currentSize == 0)
26        {
27            mutex.ReleaseMutex();
28            mutex.WaitOne();
29        }
30
31        if (head == maxSize) head = 0;
32
33        currentSize--;
34        result = items[head];
35        head++;
36        mutex.ReleaseMutex();
37        return result;
38    }
39
40    public void enqueue(int data)
41    {
42        mutex.WaitOne();
43        while (currentSize == maxSize)
44        {
45            // 释放互斥锁，让其他线程有机会获取
46            mutex.ReleaseMutex();
47
48            // 重新获取互斥锁后再检查条件
49            mutex.WaitOne();
50        }
51
52        if (tail == maxSize) tail = 0;
53
54        items[tail] = data;
55        currentSize++;
56        tail++;
57        mutex.ReleaseMutex();
58    }
59}

1using System;
2using System.Threading;
3
4class Demonstration
5{
6    static void Main()
7    {
8        BlockingQueueTest bqt = new BlockingQueueTest();
9        bqt.run();
10    }
11
12    public class BlockingQueue
13    {
14        private int[] items;
15        private int maxSize;
16        private int currentSize;
17        private int tail;
18        private int head;
19        private Mutex mutex;
20
21        public BlockingQueue(int size)
22        {
23            this.maxSize = size;
24            this.currentSize = 0;
25            this.tail = 0;
26            this.head = 0;
27            this.items = new int[size];
28            this.mutex = new Mutex();
29        }
30
31        public int dequeue()
32        {
33            int result;
34            mutex.WaitOne();
35
36            while (currentSize == 0)
37            {
38                mutex.ReleaseMutex();
39                mutex.WaitOne();
40            }
41
42            if (head == maxSize) head = 0;
43
44            currentSize--;
45            result = items[head];
46            head++;
47            mutex.ReleaseMutex();
48            return result;
49        }
50
51        public void enqueue(int data)
52        {
53            mutex.WaitOne();
54            while (currentSize == maxSize)
55            {
56                // 释放互斥锁，让其他线程有机会获取
57                mutex.ReleaseMutex();
58
59                // 重新获取互斥锁后再检查条件
60                mutex.WaitOne();
61            }
62
63            if (tail == maxSize) tail = 0;
64
65            items[tail] = data;
66            currentSize++;
67            tail++;
68            mutex.ReleaseMutex();
69        }
70    }
71
72    class BlockingQueueTest
73    {
74        private BlockingQueue bq = new BlockingQueue(1);
75
76        private void producerThread()
77        {
78            int data = 1;
79            while (true)
80            {
81                bq.enqueue(data);
82                Console.WriteLine("线程ID " + Thread.CurrentThread.ManagedThreadId + " 生产了 = " + data);
83                data++;
84                // Thread.Sleep(100);
85            }
86        }
87
88        private void consumerThread()
89        {
90            while (true)
91            {
92                int data = bq.dequeue();
93                Console.WriteLine("线程ID " + Thread.CurrentThread.ManagedThreadId + " 消费了 = " + data);
94                // Thread.Sleep(100);
95            }
96        }
97
98        public void run()
99        {
100            ThreadStart producerThreadStart = new ThreadStart(this.producerThread);
101            Thread producer = new Thread(producerThreadStart);
102            producer.Start();
103
104            Thread producer2 = new Thread(producerThreadStart);
105            producer2.Start();
106
107            Thread producer3 = new Thread(producerThreadStart);
108            producer3.Start();
109
110            ThreadStart consumerThreadStart = new ThreadStart(this.consumerThread);
111            Thread consumer = new Thread(consumerThreadStart);
112            consumer.Start();
113
114            Thread consumer2 = new Thread(consumerThreadStart);
115            consumer2.Start();
116
117            Thread consumer3 = new Thread(consumerThreadStart);
118            consumer3.Start();
119
120            // 模拟运行1秒
121            Thread.Sleep(1000);
122        }
123    }
124}

1public int dequeue()
2{
3    int result;
4    while (currentSize == 0) { }
5
6    mutex.WaitOne();
7    if (head == maxSize) head = 0;
8
9    currentSize--;
10    result = items[head];
11    head++;
12    mutex.ReleaseMutex();
13    return result;
14}

1public int dequeue()
2{
3    int result;
4    while (currentSize == 0) { }
5
6    mutex.WaitOne();
7    if (head == maxSize) head = 0;
8
9    currentSize--;
10    result = items[head];
11    head++;
12    mutex.ReleaseMutex();
13    return result;
14}

代码示例

1        public void enqueue(int data)
2        {
3            while (currentSize == maxSize) { }
4
5            mutex.WaitOne();
6            if (tail == maxSize) tail = 0;
7
8            items[tail] = data;
9            currentSize++;
10            tail++;
11            mutex.ReleaseMutex();
12        }

1using System;
2using System.Threading;
3
4class Demonstration
5{
6    static void Main()
7    {
8        BlockingQueueBrokenTest bqbt = new BlockingQueueBrokenTest();
9        bqbt.run();
10    }
11}
12
13public class BlockingQueueBroken
14{
15    private int[] items;
16    private int maxSize;
17    private int currentSize;
18    private Mutex mutex;
19    private int tail;
20    private int head;
21
22    public BlockingQueueBroken(int size)
23    {
24        this.maxSize = size;
25        this.currentSize = 0;
26        this.items = new int[size];
27        this.mutex = new Mutex();
28        this.tail = this.head = 0;
29    }
30
31    public int dequeue()
32    {
33        int result;
34        while (currentSize == 0) { }
35
36        mutex.WaitOne();
37        if (head == maxSize) head = 0;
38
39        currentSize--;
40        result = items[head];
41        items[head] = -1;
42        head++;
43        mutex.ReleaseMutex();
44        return result;
45    }
46
47    public void enqueue(int data)
48    {
49        while (currentSize == maxSize) { }
50
51        mutex.WaitOne();
52        if (tail == maxSize) tail = 0;
53
54        items[tail] = data;
55        currentSize++;
56        tail++;
57        mutex.ReleaseMutex();
58    }
59}
60
61class BlockingQueueBrokenTest
62{
63    private BlockingQueueBroken bqb = new BlockingQueueBroken(5);
64
65    private void producerThread()
66    {
67        int data = 1;
68        while (true)
69        {
70            bqb.enqueue(data);
71            Console.WriteLine("线程ID " + Thread.CurrentThread.ManagedThreadId + " 生产了 = " + data);
72            data++;
73            // Thread.Sleep(5);
74        }
75    }
76
77    private void consumerThread()
78    {
79        while (true)
80        {
81            int data = bqb.dequeue();
82            Console.WriteLine("线程ID " + Thread.CurrentThread.ManagedThreadId + " 消费了 = " + data);
83            // Thread.Sleep(10);
84        }
85    }
86
87    public void run()
88    {
89        ThreadStart producerThreadStart = new ThreadStart(this.producerThread);
90        Thread producer = new Thread(producerThreadStart);
91        producer.Start();
92
93        Thread producer2 = new Thread(producerThreadStart);
94        producer2.Start();
95
96        Thread producer3 = new Thread(producerThreadStart);
97        producer3.Start();
98
99        ThreadStart consumerThreadStart = new ThreadStart(this.consumerThread);
100        Thread consumer = new Thread(consumerThreadStart);
101        consumer.Start();
102
103        Thread consumer2 = new Thread(consumerThreadStart);
104        consumer2.Start();
105
106        // 模拟运行5秒
107        Thread.Sleep(5000);
108    }
109}