Introdução ao uso de BlockingQueue baseado em java

Autor：Eve Cole Data da Última Atualização：2024-11-18 14:48:02

Recentemente estou mantendo um projeto Java, e no grupo estamos conversando sobre os prós e os contras do Java! Infelizmente, alguns fãs de Java sempre afirmam que o desempenho não é mais tão ruim quanto o C++, e muitas bibliotecas padrão são escritas por mestres, quão estáveis elas são, etc. Simplesmente estudei com atenção, e uma das instruções que me deram foi que para entregar mensagens entre threads basta usar BlockingQueue, que foi encapsulado em java.

Como é suficiente, vamos escrever um código para testá-lo. Simplesmente escreva um pequeno programa e faça alguns testes:

Copie o código do código da seguinte forma:

//Pacote padrão
importar java.util.concurrent.*;

importar base.MyRunnable;

Teste de classe pública
{
público estático void principal(String[] args)
{
BlockingQueue<Integer> fila = new LinkedBlockingQueue<Integer>();
java.lang.Runnable r = new MyRunnable(fila);
Tópico t = novo Tópico(r);
t.start();

enquanto (verdadeiro)
{
tentar
{
enquanto (verdadeiro)
{
para(int i =0;i < 10000;i++)
{
fila.oferta(i);
}
}
}
pegar (Exceção e)
{
e.printStackTrace();
}
}
}
}

//Pacotes que precisam ser adicionados
base de pacote;

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

classe pública MyRunnable implementa Runnable
{
public MyRunnable(BlockingQueue<Integer> fila)
{
this.queue = fila;
}
execução de vazio público ()
{
Data d = nova Data();
tempo de início longo = d.getTime();
System.err.println(hora de início);
contagem interna = 0;
enquanto (verdadeiro)
{
tentar
{
Inteiro i = this.queue.poll();
se (eu! = nulo)
{
contar++;
}
se(contagem == 100000)
{
Data e = nova Data();
tempo de término longo = e.getTime();
System.err.println(contagem);
System.err.println(horário de término);
System.err.print(horário de término - horário de início);
quebrar;
}

}
pegar (Exceção e)
{

}
}
}
fila privada BlockingQueue<Integer>;
}

Para transferir 100.000 dados, são necessários cerca de 50 ms na minha máquina de teste, o que não é ruim! Basta dar uma olhada na implementação subjacente do BlockingQueue

Usei offer e poll no código de teste acima. Vamos dar uma olhada nessas duas funções de implementação.

Copie o código do código da seguinte forma:

enquete E pública() {
contagem final AtomicInteger = this.count;
se (contagem.get() == 0)
retornar nulo;
Ex = nulo;
interno c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
tentar {
if (contagem.get() > 0) {
x = extrair();
c = contagem.getAndDecrement();
se (c > 1)
notEmpty.signal();
}
} finalmente {
takeLock.unlock();
}
se (c == capacidade)
sinalNotFull();
retornar x;
}

É semelhante a um thread de sincronização geral, exceto que mais um sinal é adicionado. Ao aprender programação avançada no ambiente Unix, vi que variáveis de condição são usadas para sincronização entre threads e os threads podem ser sincronizados de maneira competitiva!
A implementação da função poll é semelhante!

Copie o código do código da seguinte forma:

oferta booleana pública (E e) {
if (e == null) lançar new NullPointerException();
contagem final AtomicInteger = this.count;
if (count.get() == capacidade)
retornar falso;
interno c = -1;
final ReentrantLock putLock = this.putLock;
putLock.lock();
tentar {
if (count.get() < capacidade) {
inserir(e);
c = contagem.getAndIncrement();
se (c + 1 <capacidade)
notFull.signal();
}
} finalmente {
putLock.unlock();
}
se(c==0)
sinalNotEmpty();
retornar c >= 0;
}