Android-Semaphore

本篇将介绍使用Semaphore来实现生产者模型

Android-使用Semaphore实现生产者消费者模型

介绍

问题

生产者消费者模型也就是为了解决：生产者和消费者在同一时间段内共用同一个存储空间，当双方生成/消费速度不对等时，数据不同步的问题

解决

那么还是生产者和消费者在同一时间段内共用同一个存储空间，当生产/消费速度不对等时。那么就让当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。

img

Semaphore

介绍

Semaphore也叫信号量，可以控制同时访问特定资源的线程数，通过协调各个线程来保证合理的使用

Semaphore内部维护了一组虚拟的许可，许可的数量可以通过构造方法来进行设定

  • 在访问特定资源前，先需要调用acquire()进行获取许可，如果当前可用许可为0，则会进入阻塞状态，直到有可用许可
  • 访问完特定资源后，需要调用release()进行释放许可

代码

下面将以添加商品和消费商品为例

  1. 首先创建生产者与消费者

    1
    2
    val mProduce = Semaphore(5)
    val mConsumer = Semaphore(5)

    这里给定了信号量最大就是5，也就是说最多缓存5个数据

  2. 在创建一个添加商品的许可和商品缓存区

    1
    2
    val mutex = Semaphore(1)
    val buffer = LinkedList<String>()

    mutex: 表示一次只能添加/消费一个商品

    buffer: 用于缓存商品

  3. 添加商品方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    suspend fun add(name: String) {
        // 尝试申请商品许可，用于添加商品
        if (mProduce.tryAcquire()) {
            delay(1000) // 模拟耗时

            // 申请(添加/消费)商品许可
            mutex.acquire()
            // 添加到缓存区
            buffer.push("${name}雨伞")
            println("添加商品 -> ${name}雨伞")
            // 添加完毕，释放许可
            mutex.release()

            // 防止第一次添加商品，顾客还没有申请消费许可
            try {
                index++
                // 释放消费许可，也就说添加与消费成一一对应
                // 添加一个商品就可以申请一个消费许可
                // 消费一个商品就可以申请一个添加许可
                mConsumer.release()
            } catch (e: Exception) {

            }
        }else{
            // 无许可，等待
            delay(1500)
            println("商品已满")
        }
    }

    这是伪代码，里面使用协程进行模拟耗时操作

  4. 消费商品

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    suspend fun consumer() {
        // 尝试申请消费许可
        if (mConsumer.tryAcquire()) {
            delay(5000)
    		// 申请(添加/消费)商品许可
            mutex.acquire()
            try {
                // 获取缓存区中最前面的商品
                val s = buffer.removeFirst()
                if (s != null) {
                    println("购买商品->${s}")
                } else {
                    mutex.release()
                }

            } catch (e: NoSuchElementException) { }
            mutex.release()

            // 释放商品许可
            try {
                mProduce.release()
            } catch (e: Exception) { }
        } else {
            delay(1500)
            println("没有商品，等待中")
        }
        // 伪代码，无限递过，不停的获取
        consumer()
    }

  5. 在上述代码中，商品添加一个只需1秒，消费一个商品需要5秒。也就是速度不对等，让我们运行代码看看结果如何把

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    fun main(): Unit = runBlocking {
    	var index = 0
        launch {
            while (true) {
                add("$index")
                index+
            }
        }
        consumer()
    }

    运行效果:

    image-20211129150637402

可以看到在，在商品许可无法申请的情况下，商品会进行阻塞，直到有许可为止