Java 기반의 BlockingQueue 사용 소개

저자：Eve Cole 업데이트 시간：2024-11-18 14:48:02

최근 Java 프로젝트를 관리하고 있는데 그룹에서 Java의 장점과 단점에 대해 이야기를 나누고 있습니다! 불행하게도 Java의 일부 궁극적인 팬은 항상 성능이 더 이상 C++만큼 나쁘지 않으며 많은 표준 라이브러리가 마스터에 의해 작성되고 얼마나 안정적인지 등을 주장합니다. 그냥 주의 깊게 공부했는데, 스레드 간에 메시지를 전달하려면 Java에 캡슐화된 BlockingQueue를 사용하면 충분하다는 지시 사항 중 하나가 주어졌습니다.

충분하므로 테스트할 코드를 작성해 보겠습니다. 간단히 작은 프로그램을 작성하고 몇 가지 테스트를 수행해 보겠습니다.

다음과 같이 코드 코드를 복사합니다 .

//기본 패키지
import java.util.concurrent.*;

import base.MyRunnable;

공개 수업 테스트
{
공개 정적 무효 메인(문자열[] 인수)
{
BlockingQueue<Integer> 대기열 = new LinkedBlockingQueue<Integer>();
java.lang.Runnable r = new MyRunnable(queue);
스레드 t = 새 스레드(r);
t.start();

동안(사실)
{
노력하다
{
동안(사실)
{
for(int i =0;i < 10000;i++)
{
queue.offer(i);
}
}
}
잡기 (예외 e)
{
e.printStackTrace();
}
}
}
}

//추가해야 할 패키지
패키지 베이스;

import java.lang.Runnable;
import java.util.concurrent.*;
import java.util.*;

공개 클래스 MyRunnable은 Runnable을 구현합니다.
{
공개 MyRunnable(BlockingQueue<Integer> 대기열)
{
this.queue = 대기열;
}
공개 무효 실행()
{
날짜 d = 새로운 날짜();
긴 시작 시간 = d.getTime();
System.err.println(시작시간);
정수 개수 = 0;
동안(사실)
{
노력하다
{
정수 i = this.queue.poll();
if(i != null)
{
카운트++;
}
if(개수 == 100000)
{
날짜 e = 새로운 날짜();
긴 종료 시간 = e.getTime();
System.err.println(count);
System.err.println(종료시간);
System.err.print(종료시간 - 시작시간);
부서지다;
}

}
잡기 (예외 e)
{

}
}
}
개인 BlockingQueue<Integer> 대기열;
}

100,000개의 데이터를 전송하는 데 제 테스트 머신에서는 약 50ms가 소요되는데, 이는 나쁘지 않습니다! BlockingQueue의 기본 구현을 살펴보세요.

위의 테스트 코드에서는 Offer와 Polling을 사용했습니다. 첫 번째는 Offer입니다.

다음과 같이 코드 코드를 복사합니다 .

공개 E 여론조사() {
최종 AtomicInteger 개수 = this.count;
if (count.get() == 0)
null을 반환;
E x = 널;
int c = -1;
최종 ReentrantLock takeLock = this.takeLock;
takeLock.lock();
노력하다 {
if (count.get() > 0) {
x = 추출();
c = count.getAndDecrement();
만약 (c > 1)
notEmpty.signal();
}
} 마지막으로 {
takeLock.unlock();
}
if (c == 용량)
signalNotFull();
x를 반환;
}

신호가 하나 더 추가된다는 점만 빼면 일반적인 동기화 스레드와 비슷합니다. Unix 환경에서 고급 프로그래밍을 배울 때 스레드 간 동기화를 위해 조건 변수를 사용하고 스레드를 경쟁적으로 동기화할 수 있다는 것을 봤습니다!
설문조사 기능의 구현도 비슷합니다!

다음과 같이 코드 코드를 복사합니다 .

공개 부울 제안(E e) {
if (e == null) 새로운 NullPointerException()을 던집니다.
최종 AtomicInteger 개수 = this.count;
if (count.get() == 용량)
거짓을 반환;
int c = -1;
최종 ReentrantLock putLock = this.putLock;
putLock.lock();
노력하다 {
if (count.get() < 용량) {
삽입(e);
c = count.getAndIncrement();
if (c + 1 < 용량)
notFull.signal();
}
} 마지막으로 {
putLock.unlock();
}
만약(c==0)
signalNotEmpty();
c >= 0을 반환합니다.
}