Java의 차단 대기열 예 분석
- PHPz앞으로
- 2023-05-09 21:43:061330검색
1. 블로킹 큐란 무엇인가요?
블로킹 큐는 데이터 구조의 일반 큐와 마찬가지로 선입선출 원칙을 따릅니다. 차단 대기열은 스레드로부터 안전한 대기열이며 다음 두 가지 특성을 갖습니다. 대기열이 가득 차면 대기열에 요소를 계속 삽입하면 다른 스레드가 대기열에서 요소를 가져올 때까지 대기열이 차단됩니다. 큐가 비어 있고 계속해서 큐에서 빼면 다른 스레드가 큐에 요소를 삽입할 때까지 큐가 차단됩니다
추가: 스레드 차단은 이 시점에서 코드가 실행되지 않음을 의미합니다. 즉, 운영 체제가 이를 예약하지 않습니다. 2. 블로킹 큐 코드 사용법
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.BlockingDeque;
public class Test {
public static void main(String[] args) throws InterruptedException {
//不能直接newBlockingDeque，因为它是一个接口，要向上转型
//LinkedBlockingDeque内部是基于链表方式来实现的
BlockingDeque<String> queue=new LinkedBlockingDeque<>(10);//此处可以指定一个具体的数字，这里的的10代表队列的最大容量
queue.put("hello");
String elem=queue.take();
System.out.println(elem);
elem=queue.take();
System.out.println(elem);
}
} 참고: put 메소드에는 블로킹 기능이 있지만, Offer에는 블로킹 기능이 없으므로 일반적으로 Put 메소드를 사용합니다(offer 메소드를 사용할 수 있는 이유). 사용되는 것은 BlockingDeque가 대기열을 상속한다는 것입니다)
BlockingDeque继承了Queue）
打印结果如上所示，当打印了hello后，队列为空，代码执行到elem=queue.take();
인쇄 결과는 위와 같습니다. hello를 인쇄한 후에는 대기열이 비어 있고 elem=queue까지 코드가 계속 실행되지 않습니다. .take(); 이때 스레드는 차단 대기 상태에 들어가며 다른 스레드가 새 요소를 대기열에 넣을 때까지 아무것도 인쇄되지 않습니다
3 생산자-소비자 모델
서버 개발과 백엔드 개발 비교 일반적으로 사용되는 프로그래밍 방법은 일반적으로 디커플링 및 피크 셰이빙 및 밸리 필링에 사용됩니다.
높은 결합도: 두 코드 모듈 사이의 상관 관계가 상대적으로 높습니다.
따라서 우리는 일반적으로 실행 효율성을 높일 수 있는 높은 응집력과 낮은 결합도를 추구합니다. 사용 생산자-소비자 모델은 분리할 수 있습니다
(1) 응용 1: 분리
여기에는 두 개의 서버가 있습니다. A 서버가 B로 데이터를 전송할 때. 직접 전송되면 A가 B에 데이터를 푸시하거나 B가 A에서 데이터를 가져오는데, 둘 다 A와 B가 직접 상호 작용해야 하므로 A와 B는 종속 관계를 갖습니다(A와 B의 결합 정도가 상대적으로 높습니다). ) . A가 C에 데이터를 전송할 수 있도록 C 서버를 추가하는 등 향후에 서버를 확장해야 하는 경우 변경이 더 복잡해지고 효율성이 떨어집니다. 이때 차단 큐인 큐를 추가할 수 있습니다. A가 큐에 데이터를 쓰고 B가 큐에서 데이터를 가져오면 큐는 환승 스테이션(또는 거래 장소)에 해당하고 A는 큐에 해당합니다. 생산자(데이터 제공) B는 소비자(데이터 수신)와 동일합니다. 이때 생산자-소비자 모델이 형성되어 코드가 덜 결합되고 유지 관리가 더 편리하며 실행이 더 효율적입니다. 
컴퓨터에서 생산자는 하나의 스레드 그룹으로 작동하고 소비자는 또 다른 스레드 그룹으로 작동하며 거래 장소는 차단 대기열을 사용할 수 있습니다.
(2) 응용 프로그램 2: 피크 감소 및 밸리 채우기 
실생활에서
댐은 강의 매우 중요한 부분입니다. 댐이 없다면 결과를 상상해 보세요. 홍수 시즌 이후 상류의 물이 매우 많아지면 많은 양의 물이 유입됩니다. 하류 지역에서는 홍수로 인해 작물이 물에 잠길 수 있으며, 하류에는 물이 거의 없어 가뭄이 발생할 수 있습니다. 댐이 있는 경우 홍수기에는 댐에 잉여 물을 저장하고, 수문을 닫아 물을 저장하고, 상류의 물을 일정 비율로 하류로 흐르게 하여 갑작스러운 폭우로 인해 홍수가 발생하는 것을 방지합니다. 하류가 침수되지 않도록 하류. 가뭄 기간 동안 댐은 미리 저장된 물을 방출하고 하류에 물이 너무 부족한 것을 방지하기 위해 물이 특정 속도로 하류로 흐르도록 허용합니다. 이렇게 하면 홍수기의 홍수와 건기의 가뭄을 모두 피할 수 있습니다.
피크: 홍수 기간에 해당
골짜기: 건기 기간에 해당
이 상황은 컴퓨터, 특히 서버 개발에서도 매우 일반적입니다. 게이트웨이는 일반적으로 인터넷의 요청을 일부 제품과 같은 비즈니스 서버로 전달합니다. 서버, 사용자 서버, 가맹점 서버(가맹점 정보 저장), 생방송 서버. 그러나 인터넷에서 들어오는 요청 수는 통제할 수 없기 때문에 갑자기 큰 파도가 오면 게이트웨이가 이를 처리할 수 있더라도 많은 후속 서버가 많은 요청을 받은 후 붕괴됩니다(1개 처리). 요청이 일련의 데이터베이스 작업을 포함하므로 데이터베이스 관련 작업의 효율성이 상대적으로 낮기 때문에 요청이 너무 많으면 처리할 수 없어 충돌이 발생합니다)
所以实际情况中网关和业务服务器之间往往用一个队列来缓冲，这个队列就是阻塞队列（交易场所），用这个队列来实现生产者（网关）消费者（业务服务器）模型，把请求缓存到队列中，后面的消费者（业务服务器）按照自己固定的速率去读请求。这样当请求很多时，虽然队列服务器可能会稍微受到一定压力，但能保证业务服务器的安全。
（3）相关代码
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class TestDemo {
public static void main(String[] args) {
// 使用一个 BlockingQueue 作为交易场所
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
// 此线程作为消费者
Thread customer = new Thread() {
@Override
public void run() {
while (true) {
// 取队首元素
try {
Integer value = queue.take();
System.out.println("消费元素: " + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
customer.start();
// 此线程作为生产者
Thread producer = new Thread() {
@Override
public void run() {
for (int i = 1; i <= 10000; i++) {
System.out.println("生产了元素: " + i);
try {
queue.put(i);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
producer.start();
try {
customer.join();
producer.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
打印如上（此代码是让生产者通过sleep每过1秒生产一个元素，而消费者不使用sleep，所以每当生产一个元素时，消费者都会立马消费一个元素）
4.阻塞队列和生产者消费者模型功能的实现
在学会如何使用BlockingQueue后，那么如何自己去实现一个呢？
主要思路：
1.利用数组
2.head代表队头，tail代表队尾
3.head和tail重合后到底是空的还是满的判断方法：专门定义一个size记录当前队列元素个数，入队列时size加1出队列时size减1，当size为0表示空，为数组最大长度就是满的（也可以浪费一个数组空间用head和tail重合表示空，用tail+1和head重合表示满，但此方法较为麻烦，上一个方法较为直观，因此我们使用上一个方法）
public class Test2 {
static class BlockingQueue {
private int[] items = new int[1000]; // 此处的1000相当于队列的最大容量, 此处暂时不考虑扩容的问题.
private int head = 0;//定义队头
private int tail = 0;//定义队尾
private int size = 0;//数组大小
private Object locker = new Object();
// put 用来入队列
public void put(int item) throws InterruptedException {
synchronized (locker) {
while (size == items.length) {
// 队列已经满了，阻塞队列开始阻塞
locker.wait();
}
items[tail] = item;
tail++;
// 如果到达末尾, 就回到起始位置.
if (tail >= items.length) {
tail = 0;
}
size++;
locker.notify();
}
}
// take 用来出队列
public int take() throws InterruptedException {
int ret = 0;
synchronized (locker) {
while (size == 0) {
// 对于阻塞队列来说, 如果队列为空, 再尝试取元素, 就要阻塞
locker.wait();
}
ret = items[head];
head++;
if (head >= items.length) {
head = 0;
}
size--;
// 此处的notify 用来唤醒 put 中的 wait
locker.notify();
}
return ret;
}
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue queue = new BlockingQueue();
// 消费者线程
Thread consumer = new Thread() {
@Override
public void run() {
while (true) {
try {
int elem = queue.take();
System.out.println("消费元素: " + elem);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
consumer.start();
// 生产者线程
Thread producer = new Thread() {
@Override
public void run() {
for (int i = 1; i < 10000; i++) {
System.out.println("生产元素: " + i);
try {
queue.put(i);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
producer.start();
consumer.join();
producer.join();
}
}
运行结果如上。
注意：
1.wait和notify的正确使用
2.put和take都会产生阻塞情况，但阻塞条件是对立的，wait不会同时触发（put唤醒take阻塞，take唤醒put阻塞）
위 내용은 Java의 차단 대기열 예 분석의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!