当前位置：首页 > news >正文

【构建并发程序】8-并发队列之阻塞队列

news 来源：原创 2024/5/21 17:34:59

1 BlockingQueue的各个版本

并发队列操作	异常版本	特殊值版本	计时等待版本	阻塞版本
出队	remove()	pool()	pool(t:Long,u:TimeUnit)	take()
入队	add(x:T)	offer(x:T)	offer(x:T,t:Long,u:TimeUnit)	put(x:T)
检查下一个元素	element	peek	无	无

1.1 往队列中添加元素

add：若队列满了，报异常
offer (特殊值版本)：若队列满了，还添加报False
offer (计时等待版本)：若队列满了，则等待如果超出设置的等待时间，还添加报False
put：若队列满了，则线程堵塞一直等待空间，以期添加任务元素

1.2 往队列中删除元素

remove：若队列为空，报异常
pool (特殊值版本)：若删除了返回对应的值，如果队列为空删除失败则返回null
pool (计时等待版本)：若删除了返回对应的值，如果队列为空删除失败则等待如果超出设置的等待时间则返回null
take：若队列为空，则线程堵塞一直等待空间，以期删除任务元素。

2 阻塞队列之“有界”与“无界”

ArrayBlockingQueue：是一个有界的阻塞队列。
在创建ArrayBlockingQueue队列时，我们需要设置它的容量，即该队列能够容纳的元素数量的最大值。如果生产者创建元素的速度比消费者处理元素的速度快，那么就应该使用有界队列。否则，不断增长的无界队列会消耗光该程序的所有可用内存。

LinkedBlockingQueue：是一个无界的队列
如果消费者处理元素的速度要比生产者创建元素的速度快时，就可以使用该无界队列。

2.1 代码示例

LinkedBlockungQueue无界队列的例子；Main线程中对并发队列进行入队操作，其它线程对并发队列进行出队操作；

class three_并发队列_File(var root: String) {


  // 将队列声明为messages的私有变量
  private val messages = new LinkedBlockingQueue[String]()
  // 一个名为logger的独立守护线程，调用了该队列中的take方法；
  // take是出队的阻塞版本(他会阻止线程logger调用它，知道队列中含有消息为止)

  val logger: Thread = new Thread {
    //setDaemon(true)
    // 将(logger:Thread)设置为守护线程；所有线程都结束后，守护线程就结束(即main线程结束后，所有线程都会结束)
    // 举个例子：
    //  如果我的队列中没有值，也就是我把 ”//fileSystem.logMessage("Testing log!")“注释掉，那么
    // 此时程序就会卡在第一个循环中的 messages.take() {表示的是：若队列为空，则线程堵塞一直等待空间，以期删除任务元素}；
    // 但如果我设置成守护线程，则代表者我main线程结束，管你 卡不卡住，此线程都会退出


    // messages.take():并发队列中有值，第一次循环会将其取出来，但由于我只入队了一次值，所以第二次循环时就等待堵塞了；
    override def run(): Unit = {
      while (true) {
        val msg = messages.take()
        //val msg = messages.poll()
        Logger.getRootLogger.info(msg+" -- "+logger.getName)
      }
    }
  }

  logger.start()

  //  入队操作，我们也可以使用add或put方法，因为这个队列是无界的，因此这些方法永远都不会抛出异常
  //或者阻塞调用它们的线程
  def logMessage(msg: String): Boolean = messages.offer(msg)

}

object FileSystemTest extends App {

  val fileSystem = new three_并发队列_File(".")
  fileSystem.logMessage("Testing log!")

  Logger.getRootLogger.info("main 线程结束")

}