Lua Academy · 课时

生产者-消费者模式

Q: 「生产者-消费者模式」这节课中我会学到什么？

使用协程实现生产者-消费者流水线。 你通过在浏览器中直接运行的动手代码来练习 Lua Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

使用协程实现生产者-消费者流水线。

生产者-消费者模式 是 CoddyKit 上的免费 Lua Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Lua Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Lua Academy 课程共包含 4 节课。

模式解析

使用协程实现生产者-消费者模式：生产者生成数据，并逐个 yield 项目；消费者一次接收一个项目。双方都无需了解对方的实现——它们通过 yield/resume 通信。这样可以将数据生成与数据处理解耦。

local function makeProducer(items)
  return coroutine.create(function()
    for _, item in ipairs(items) do
      coroutine.yield(item)
    end
  end)
end

local function consume(producer)
  while true do
    local ok, item = coroutine.resume(producer)
    if not ok or item == nil then break end
    print("Processing:", item)
  end
end

consume(makeProducer({10, 20, 30, 40}))

无限生产者

生产者协程可以 yield 一个无限序列。消费者负责决定何时停止。这非常适合数据流来源，例如传感器读数、日志事件和生成的序列。

local function primes()
  return coroutine.wrap(function()
    local function isPrime(n)
      if n<2 then return false end
      for i=2, math.floor(math.sqrt(n)) do
        if n%i==0 then return false end
      end
      return true
    end
    local n = 2
    while true do
      if isPrime(n) then coroutine.yield(n) end
      n = n + 1
    end
  end)
end

local gen = primes()
for i = 1, 10 do
  io.write(gen() .. " ")
end
print()  -- 2 3 5 7 11 13 17 19 23 29

作为协程的消费者

消费者也可以是一个协程，从而创建完全对称的生产者-消费者组合。第三个“驱动器”函数负责协调双方。这对于带缓冲的流水线很有用，因为生产者和消费者的处理速率可能不同。

local function producer()
  return coroutine.create(function(consumer)
    for i = 1, 5 do
      coroutine.resume(consumer, i * 10)
      coroutine.yield()  -- wait
    end
  end)
end

local consumer = coroutine.create(function()
  while true do
    local val = coroutine.yield()
    print("Received:", val)
  end
end)

coroutine.resume(consumer)  -- start, waits for data
local pro = producer()
for i = 1, 5 do
  coroutine.resume(pro, consumer)
end

反压

反压是指消费者向生产者发出减速信号。使用协程时，这一过程很自然：生产者在每个项目之后执行 yield，等待消费者请求下一个项目。如果消费者速度较慢，生产者就会等待。

local function slowConsumer(gen)
  for v in gen do
    -- Simulate slow processing
    print("Processing", v, "...")
    -- In real code: os.execute("sleep 0.1")
  end
end

local function fastProducer(n)
  return coroutine.wrap(function()
    for i = 1, n do
      print("Producing", i)
      coroutine.yield(i)
    end
  end)
end

-- Producer automatically waits for consumer:
slowConsumer(fastProducer(4))

批量处理

批处理生产者会先收集 N 个项目，再 yield 一个批次。消费者接收完整批次并进行处理。对于高频生产者，这可以减少协调开销。

local function batchProducer(items, batchSize)
  return coroutine.wrap(function()
    local batch = {}
    for _, item in ipairs(items) do
      batch[#batch+1] = item
      if #batch >= batchSize then
        coroutine.yield(batch)
        batch = {}
      end
    end
    if #batch > 0 then coroutine.yield(batch) end
  end)
end

local data = {1,2,3,4,5,6,7,8,9,10}
for batch in batchProducer(data, 3) do
  print("Batch:", table.concat(batch, ","))
end
-- Batch: 1,2,3
-- Batch: 4,5,6
-- Batch: 7,8,9
-- Batch: 10

文件行生产者

惰性读取文件：生产者协程一次读取一行，并将每行 yield 出去。消费者可以逐行处理，而无需将整个文件加载到内存中。

local function lineProducer(path)
  return coroutine.wrap(function()
    local f = io.open(path, "r")
    if not f then return end
    for line in f:lines() do
      coroutine.yield(line)
    end
    f:close()
  end)
end

-- Process file line by line without loading all at once
local wordCount = 0
for line in lineProducer("data.txt") do
  for _ in line:gmatch("%S+") do
    wordCount = wordCount + 1
  end
end
print("Words:", wordCount)

转换阶段

在生产者和消费者之间插入一个转换阶段。转换阶段从一个协程读取数据，并将处理后的值 yield 给另一个协程。通过连接这些阶段，可以构建数据流水线。

local function transform(source, fn)
  return coroutine.wrap(function()
    for v in source do
      coroutine.yield(fn(v))
    end
  end)
end

local numbers = coroutine.wrap(function()
  for i = 1, 6 do coroutine.yield(i) end
end)

local doubled = transform(numbers, function(n) return n*2 end)
local filtered = coroutine.wrap(function()
  for v in doubled do
    if v > 4 then coroutine.yield(v) end
  end
end)

for v in filtered do io.write(v.." ") end
print()  -- 6 8 10 12

流水线中的错误传播

协程流水线中的错误会通过已结束状态或 pcall 传播。请将流水线阶段包装在 pcall 中，以便优雅地捕获和处理错误，而不会导致整个流水线崩溃。

local function safeStage(source, fn)
  return coroutine.wrap(function()
    for v in source do
      local ok, result = pcall(fn, v)
      if ok then
        coroutine.yield(result)
      else
        print("Error processing", v, ":", result)
      end
    end
  end)
end

local data = coroutine.wrap(function()
  for _, v in ipairs({4, -1, 9, 0, 16}) do coroutine.yield(v) end
end)

local results = safeStage(data, function(n)
  assert(n > 0, "non-positive")
  return math.sqrt(n)
end)

for v in results do print(v) end

合并生产者

将多个生产者合并为一个：在所有仍存活的生产者之间轮流调度，每次 yield 一个项目。所有生产者都耗尽后停止。

local function merge(...)
  local producers = {...}
  return coroutine.wrap(function()
    while #producers > 0 do
      local alive = {}
      for _, co in ipairs(producers) do
        local ok, v = coroutine.resume(co)
        if ok and v ~= nil then
          coroutine.yield(v)
          alive[#alive+1] = co
        end
      end
      producers = alive
    end
  end)
end

local function src(items)
  return coroutine.create(function()
    for _,v in ipairs(items) do coroutine.yield(v) end
  end)
end

for v in merge(src({1,3,5}),src({2,4,6})) do
  io.write(v.." ")
end
print()  -- 1 2 3 4 5 6

实际应用场景

日志处理流水线：生产者读取日志行，筛选阶段只保留错误行，转换阶段提取关键信息，消费者按错误类型进行汇总。每个阶段都是独立且可测试的协程。

local function makeFilter(src, pred)
  return coroutine.wrap(function()
    for line in src do
      if pred(line) then coroutine.yield(line) end
    end
  end)
end

local logs = coroutine.wrap(function()
  local entries = {
    "INFO: started",
    "ERROR: timeout",
    "DEBUG: connecting",
    "ERROR: auth failed",
  }
  for _, e in ipairs(entries) do coroutine.yield(e) end
end)

local errors = makeFilter(logs, function(l) return l:match("^ERROR") end)
for line in errors do print(line) end
-- ERROR: timeout
-- ERROR: auth failed

快速检查

生产者-消费者协程模式如何处理反压？

回顾：生产者-消费者

总结：

生产者 yield 项目；消费者通过调用 resume 来驱动生产者
无限生产者：while true do yield() end
流水线：在生产者和消费者之间连接转换协程
反压是自动实现的：生产者会等待消费者
将批处理、筛选和合并阶段实现为协程包装器
在流水线阶段使用 pcall 隔离错误

常见问题解答

「生产者-消费者模式」课时是免费的吗？

是的 — 「生产者-消费者模式」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Lua Academy 课程的其余内容，请升级到 CoddyKit PRO。 Lua Academy 课程共包含 4 节课。

「生产者-消费者模式」这节课中我会学到什么？

使用协程实现生产者-消费者流水线。你通过在浏览器中直接运行的动手代码来练习 Lua Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Lua Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Lua Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「生产者-消费者模式」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Lua Academy 课中编写并运行代码吗？

能。每节 Lua Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

创建协程
resume 和 yield
协程状态
生产者-消费者模式

← 返回 Lua Academy