O uso de BlockingQueue na concorrência de Java

Este artigo fala principalmente sobre as coleções envolvidas na simultaneidade. Para coleções comuns, consulte [Visão geral da coleção Java]

1. O que é BlockingQueue

BlockingQueue é uma fila de bloqueio. A partir da palavra bloqueio, pode -se ver que o acesso a uma fila de bloqueio pode causar bloqueio em alguns casos. Existem dois casos principais bloqueados:

1. Quando a fila estiver cheia, ela será feita.
2. Quando a fila estiver vazia, ela estará fora da fila.

Portanto, quando um thread tenta fila uma fila já completa, ele será bloqueado, a menos que outro thread faça a operação da fila; Da mesma forma, quando um thread tenta fila uma fila vazia, ele será bloqueado, a menos que outro thread tenha a operação da fila.

Em Java, a interface BlockingQueue está localizada no pacote java.util.concurrent (fornecido na versão Java 5). A partir das características da fila de bloqueio introduzida acima, pode-se observar que a fila de bloqueio é segura por threads.

2. Como usar o BlockingQueue

As filas de bloqueio são usadas principalmente em cenários de produtor/consumidor. A figura a seguir mostra uma produção de threads e um cenário de consumo de threads:

O fio responsável pela produção cria continuamente novos objetos e os insere na fila de bloqueio até que o limite superior desta fila seja atingido. Depois que a fila atingir seu limite superior, o segmento de produção será bloqueado até que o fio consumido consome a fila. Da mesma forma, o fio responsável pelo consumo constantemente consome objetos da fila até que a fila esteja vazia. Quando a fila estiver vazia, o encadeamento de consumo será bloqueado, a menos que um novo objeto na fila seja inserido.

3. Métodos na interface BlockingQueue

Existem quatro conjuntos de métodos para bloquear filas para executar insert , remove e examine , respectivamente. Quando as operações correspondentes a cada conjunto de métodos não podem ser executadas imediatamente, haverá reações diferentes. A tabela a seguir lista esses métodos de maneira classificada:

As características correspondentes desses quatro conjuntos de métodos são:

1. ThrowSexception: Se a operação não puder ser realizada imediatamente, uma exceção será lançada.
2. EspecialValue: Se a operação não puder ser realizada imediatamente, um valor especial será retornado, geralmente verdadeiro ou falso
3. Blocos: Se a operação não puder ser realizada imediatamente, a operação será bloqueada
4. Timesout: Se a operação não puder ser executada imediatamente, a operação será bloqueada no horário especificado. Se o tempo especificado não for executado, um valor especial será retornado, que geralmente é verdadeiro ou falso.

Deve -se notar que não podemos inserir null no BlockingQueue, caso contrário, será relatada NullPointerException .

4. Classe de implementação de Blockingqueue

BlockingQueue é apenas uma interface no pacote java.util.concurrent . Ao usá -lo especificamente, usamos suas classes de implementação. Obviamente, essas classes de implementação também estão localizadas no pacote java.util.concurrent . No Java 6, as classes de implementação do BlockingQueue são principalmente as seguintes:

1. ArrayblockingQueue
2. Atraso
3. LinkedBlockingQueue
4. PriorityBlockingQueue
5. Síncrono

Abaixo, apresentaremos essas classes de implementação separadamente.

4.1 ArrayblockingQueue

O ArrayBlockingQueue é uma fila de bloqueio limitada e sua implementação interna é uma matriz. O significado do limite significa que sua capacidade é limitada, devemos especificar seu tamanho de capacidade quando é inicializado e o tamanho da capacidade não pode ser alterado depois de especificado.

ArrayblockingQueue armazena dados de uma maneira primeiro a sair. O objeto recém -inserido é a cauda e o objeto recém -movido é a cabeça. Aqui está um exemplo de inicialização e uso do ArrayBlockingQueue:

 BlockingQueue fila = novo ArrayBlockingQueue (1024); fila.put ("1"); objeto objeto = fila.take ();

4.2 Atraso

Quais bloqueios de atraso são seus elementos internos. Os elementos no atraso devem implementar a interface java.util.concurrent.Delayed . A definição dessa interface é muito simples:

 A interface pública atrasada se estende comparável a <oulded> {long getDelay (unidade de unidade timeunit);}

O valor de retorno do método getDelay() é o tempo de espera antes que o elemento da fila seja liberado. Se 0 ou um负值for retornado, significa que o elemento expirou e precisa ser liberado. Neste momento, o atraso divulgará esse objeto através do método take() .

Como pode ser visto na definição de interface atrasada acima, ele também herda a interface Comparable . Isso ocorre porque os elementos no atraso precisam ser classificados. De um modo geral, classificamos a prioridade do tempo de expiração do elemento.

Exemplo 1: especifique um tempo de expiração para um objeto

Primeiro, definimos um elemento, que precisa implementar a interface atrasada

 classe pública Atraso os implementos atrasados {privado por muito tempo expirado; Longo atraso privado; nome de string privado; Atraso (String elementName, Longo Atraso) {this. nome = elementName; esse. atraso = atraso; expirado = (atraso + sistema. currenttimemillis ()); } @Override public int compareto (atrasado o) {touchedElement cache = (touchedElement) o; retornar em cache.Getexired ()> expirado? 1: -1; } @Override Public Long GetDelay (unidade TimeUnit) {return (expirou - System. Currenttimemillis ()); } @Override public string tostring () {return "toutEdElement [telto =" + tardar + ", name =" + name + "]"; } public long getExpired () {return expirou; }}

Defina o tempo de expiração deste elemento para 3s

 classe pública touchQueueExample {public static void main (string [] args) lança interruptedException {toutQueue <weoulDelement> fila = new taunhas <> (); TouchedElement ele = new toutEdElement ("cache 3 segundos", 3000); fila.put (ele); Sistema. out.println (fileue.take ()); }}

Execute esta função principal e podemos descobrir que precisamos esperar 3 segundos antes de imprimir esse objeto.

De fato, existem muitos cenários de aplicação para o atraso, como conexões de fechamento cronometradas, objetos de cache, processamento de tempo limite e outros cenários. Vamos fazer o exame do aluno como um exemplo para permitir que todos entendam o uso do atraso mais profundamente.

Exemplo 2: Trate todos os alunos no exame como um atraso, quem terminar as perguntas primeiro os libera

Primeiro, construímos um objeto de estudante

 public classe Student implementa Runnable, atrasado {private String Name; // Nomeie o prolongado tempo de custos privados; // tempo para as perguntas do teste Private Longo acabamento; // Tempo para conclusão Public Student (Nome da String, Longo Custo) {this. nome = nome; esse. CostTime = CostTime; TEMPOTIMENTO = CostTime + System. currenttimemillis (); } @Override public void run () {System. out.println (Nome + "Envie o artigo, tempo" + CostTime /1000); } @Override Public Long GetDelay (unidade TimeUnit) {return (FinalizeTime - System. CurrentTimemillis ()); } @Override public int Compareto (atrasado o) {Student Other = (Student) o; Retorno CostTime> = Other. CostThtime? 1: -1; }}

Em seguida, construa um objeto de professor para levar o exame aos alunos

 Public Class Professor {estático final int student_size = 30; public static void main (string [] args) lança interruptedException {aleather r = new Random (); // Pense em todos os alunos como um atraso na fila de atraso <estudante> alunos = novo detrase <estudante> (); // Construa um pool de threads para permitir que os alunos "façam seus trabalhos de casa" ExecorService Exec = executores.newfixedThreadpool (student_size); para (int i = 0; i <student_size; i ++) {// inicialize o nome e a hora do aluno para fazer o teste estudantes.put (novo aluno ("aluno" + (i + 1), 3000 + r.nextint (10000))); } // Inicie o teste while (! Student.isempty ()) {Exec.execute (estudantes.take ()); } exec.shutdown (); }}

Vamos dar uma olhada nos resultados em execução:

Aluno 2 Envie o artigo, 3
Aluno 1 entregando papéis, levando 5
Aluno 5 Envie o artigo, 7
Aluno 4 Envie o jornal, levando 8
Aluno 3 Envie o artigo, 11

Através dos resultados em execução, podemos descobrir que cada aluno "enviará o artigo" após o tempo de início especificado ( dependendo do método getDelay () ) e o artigo será enviado primeiro ( dependendo do método compareto () ).

Observando seu código -fonte, você pode ver que a implementação interna do atraso usa PriorityQueue e um bloqueio:

4.3 LinkedBlockingQueue

A configuração do tamanho da fila de bloqueio do LinkedBlockingQueue é opcional. Se especificarmos um tamanho ao inicializar, ele será limitado e, se não for especificado, ele será limitado. Diz -se que é ilimitado, mas, na verdade, o tamanho padrão é Integer.MAX_VALUE capacidade. Sua implementação interna é uma lista vinculada.

Como o ArrayBlockingQueue, o LinkedBlockingQueue também armazena dados de uma maneira de primeira vez na primeira saída. O objeto recém -inserido é a cauda e o objeto recém -movido é a cabeça. Aqui está um exemplo de inicialização e fabricação do LinkedBlockingQuee:

 BlockingQueue <String> ilimited = new LinkedBlockingQueue <String> (); BlockingQueue <String> limitado = new LinkedBlockingQueue <String> (1024); limitado.put ("value"); string value = limitado.take ();

4.4 PriorityBlockingQueue

O PriorityBlockingQueue é uma fila sem limites, e suas regras de classificação são as mesmas que java.util.PriorityQueue . Deve -se notar que objetos nulos podem ser inseridos no PriorityBlockingQueue.

Todos os objetos inseridos no PriorityBlockingQueue devem implementar a interface java.lang.Comparable , e as regras de classificação da prioridade da fila são definidas de acordo com a implementação dessa interface.

Além disso, podemos obter um iterador do PriorityBlockingQueue, mas esse iterador não garante iteração na ordem prioritária.

Vamos dar um exemplo para ilustrar. Primeiro, definimos um tipo de objeto, que precisa implementar a interface comparável:

 public classe priorityElement implementa comparável <priorityElement> {private int priority; // define prioridade prioritária (int priority) {// inicialize prioridade this.priority = priority;}@substituirpublic int compare (priorityElement o) {// classificar por tamanho prioritário retorno> = o.getyPriority () () {// classificar por tamanho prioritário> = o.gety? 1: -1;} public int getPriority () {Return Priority;} public void setPriority (int priority) {this.priority = priority;}@substituir string tostring () {return "priorityElement [priority =" + priority + "];}}}}} {

Então nós estabelecemos prioridade aleatoriamente para a fila

 classe pública PriorityBlockingQueeExample {public static void main (string [] args) lança interruptedException {priorityBlockingQueue <lorityElement> fila = new priorityBlockingQueue <> (); for (int i = 0; i <5; i ++) {aleatório = novo aleatório (); PriorityElement ele = New PriorityElement (Random.NextInt (10)); fila.put (ele); } while (! }}}

Confira os resultados em execução:

PriorityElement [prioridade = 3]
PriorityElement [prioridade = 4]
PriorityElement [prioridade = 5]
PriorityElement [prioridade = 8]
PriorityElement [prioridade = 9]

4.5 Síncrono

Apenas um elemento é permitido dentro da fila síncrona. Quando um thread insere um elemento, ele será bloqueado, a menos que o elemento seja consumido por outro thread.

O exposto acima é todo o conteúdo deste artigo. Espero que seja útil para o aprendizado de todos e espero que todos apoiem mais o wulin.com.