Partage de technologie

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

file d'attente bloquante

1. Concept

La file d'attente de blocage est un type particulier de file d'attente. Elle suit également le principe du « premier entré, premier sorti ».

La file d'attente de blocage peut être une structure de données thread-safe et présente les caractéristiques suivantes :

• Lorsque la file d'attente est pleine, continuer à entrer dans la file d'attente sera bloqué jusqu'à ce que d'autres threads prennent des éléments de la file d'attente..
• Lorsque la file d'attente est vide, continuer à retirer la file d'attente sera également bloqué jusqu'à ce que d'autres threads insèrent des éléments dans la file d'attente.

Un scénario d'application typique de blocage de file d'attente est le « modèle producteur-consommateur ». Il s'agit d'un modèle de développement très typique.
producteur

Le modèle de consommation résout le problème du couplage fort entre producteurs et consommateurs via un conteneur.

Les producteurs et les consommateurs ne communiquent pas directement entre eux, mais via des files d'attente de blocage. Par conséquent, une fois que le producteur a produit les données, il n'a pas besoin d'attendre que le consommateur les traite, mais les jette directement dans la file d'attente de blocage. ne demande pas de données au producteur, mais est extraite directement de la file d'attente de blocage.

Il y a deux étapes majeures dans la mise en œuvre du modèle producteur-consommateur :

1. producteur d'outils
2. réaliser le consommateur

2. File d'attente bloquante dans la bibliothèque standard

Il existe une file d'attente de blocage intégrée à la bibliothèque standard Java. Si nous devons utiliser des files d'attente de blocage dans certains programmes, nous pouvons utiliser directement la file d'attente de blocage de la bibliothèque standard.
Peut.

• BlockingQueue est une interface. La véritable classe d'implémentation est LinkedBlockingQueue.
• La méthode put est utilisée pour bloquer l’entrée dans la file d’attente et la méthode take est utilisée pour bloquer la sortie de la file d’attente.
• BlockingQueue propose également des méthodes telles que offer, poll et peek, mais ces méthodes n'ont pas de caractéristiques de blocage.

Le pseudo-code pour bloquer la file d'attente est le suivant :

BlockingQueue<String> queue = new LinkedBlockingQueue<>();

// ⼊队列
queue.put("abc");

// 出队列. 如果没有 put 直接 take, 就会阻塞.
String elem = queue.take();
1
2
3
4
5
6
7

3. Modèle producteur-consommateur

public static void main(String[] args) throws InterruptedException {

	BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>();
	
	Thread customer = new Thread(() -> {
		while (true) {
			try {
				int value = blockingQueue.take();
				System.out.println("消费元素: " + value);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}, "消费者");
	
customer.start();

	Thread producer = new Thread(() -> {
		Random random = new Random();
		while (true) {
			try {
				int num = random.nextInt(1000);
				System.out.println("生产元素: " + num);
				blockingQueue.put(num);
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}, "生产者");
	
	producer.start();
	customer.join();
	producer.join();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

4. Blocage de la mise en œuvre de la file d'attente

1. Ceci est réalisé grâce à la méthode de la « file d’attente circulaire ».
2. Utiliser synchronisé pour le contrôle du verrouillage.
3. Lorsque put insère un élément, il détermine si la file d'attente est pleine et attend. (Notez que l'attente doit être effectuée en boucle. Lorsqu'elle est réveillée, la file d'attente peut ne pas être pleine car plusieurs threads peuvent être réveillés en même temps.) .
4. Lorsque take supprime l'élément, il détermine si la file d'attente est vide et attend (c'est aussi une attente en boucle).

public class BlockingQueue {

	private int[] items = new int[1000];
	private volatile int size = 0;
	private volatile int head = 0;
	private volatile int tail = 0;
	
	public void put(int value) throws InterruptedException {
		synchronized (this) {
		// 此处最好使⽤ while.
		// 否则 notifyAll 的时候, 该线程从 wait 中被唤醒,
		// 但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能⼜已经队列满了
		// 就只能继续等待
			while (size == items.length) {
				wait();
			}
			items[tail] = value;
			tail = (tail + 1) % items.length;
			size++;
			notifyAll();
		}
	}
	
	public int take() throws InterruptedException {
		int ret = 0;
		synchronized (this) {
			while (size == 0) {
				wait();
			}
			ret = items[head];
			head = (head + 1) % items.length;
			size--;
			notifyAll();
		}
		return ret;
	}
	
	public synchronized int size() {
		return size;
	}
	
	// 测试代码
	public static void main(String[] args) throws InterruptedException {
		BlockingQueue blockingQueue = new BlockingQueue();
		Thread customer = new Thread(() -> {
		while (true) {
			try {
				int value = blockingQueue.take();
				System.out.println(value);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}, "消费者");	
	
		customer.start();
		Thread producer = new Thread(() -> {
			Random random = new Random();
				while (true) {					try {
						blockingQueue.put(random.nextInt(10000));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}, "生产者");
			
		producer.start();
		customer.join();
		producer.join();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

Résumer

1. La file d'attente de blocage équivaut à un tampon, équilibrant les capacités de traitement des producteurs et des consommateurs (Peak clipping et remplissage).
2. Le blocage des files d’attente peut également découpler les producteurs et les consommateurs.