仓颉编程实练-并发-使用ArrayBlockingQueue实现工作者模式

main.cj:

package demo1

import std.collection.ArrayList
import std.collection.concurrent.*;
import std.sync.*

struct Task {
    public Task(let id: UInt64) {}
}

main(): Int64 {
    let taskQueue = ArrayBlockingQueue<Task>(10)
    let queueIsEnd = AtomicBool(false)

    let tList = ArrayList<Future<Unit>>()
    for (_ in 0..3) {
        let t = spawn {
            while(true) {
                if (queueIsEnd.load()) {
                    println("queue is end, thread end.")
                    break
                }

                let op = taskQueue.remove(2 * Duration.second) // 读取任务，并设置2s超时
                match (op) {
                    case Some(task) => 
                        let taskId = task.id
                        if (taskId == 0) { // 获得结尾空任务
                            queueIsEnd.store(true)
                            println("got empty task: ${taskId}, thread end.")
                            break
                        }
                        println("Worker 执行任务: ${taskId}")
                    case None => 
                        println("超时，继续等待")
                }
            }
        }
        tList.add(t)
    }

    for (i in 1..10) {
        taskQueue.add(Task(UInt64(i)))
    }
    taskQueue.add(Task(0)) // 放入结尾空任务

    for (t in tList) {
        t.get()
    }

    println("main thread done.")
    return 0
}

运行：

> cjpm run
Worker 执行任务: 1
Worker 执行任务: 3
Worker 执行任务: 5Worker 执行任务: 4
Worker 执行任务: 6

Worker 执行任务: 2
Worker 执行任务: 9
got empty task: 0, thread end.      
Worker 执行任务: 8
queue is end, thread end.
Worker 执行任务: 7
queue is end, thread end.
main thread done.

cjpm run finished

说明：

1. 实现了并发的工作者模式，主线程向任务队列中添加任务，并发worker线程消费这些任务，直到所有任务被处理完毕，所有的worker线程收到通知后退出。
2. 使用ArrayBlockingQueue进行线程间通信，worker读取时使用超时机制
3. 增加一个空任务，并使用AtomicBool(原子操作)作为线程间通知任务队列已经为空的标志

可以对比一下golang的类似实现：

package main

import (
	"fmt"
	"sync"
)

// Task 表示要执行的任务
type Task struct {
	ID int
}

func main() {
	taskQueue := make(chan Task, 10)

	var wg sync.WaitGroup

	// 并发的worker
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			for task := range taskQueue {
				fmt.Printf("Worker 执行任务 %d\n", task.ID)
			}
		}()
	}

	// 添加一些任务到任务队列中
	for i := 1; i <= 10; i++ {
		task := Task{ID: i}
		taskQueue <- task
	}

	// 关闭任务队列，等待所有工作者完成任务
	close(taskQueue)
	wg.Wait()
	println("main thread done.")
}