プロデューサー・コンシューマーパターン
コルーチンを使ってプロデューサー・コンシューマー型のパイプラインを実装します。
「プロデューサー・コンシューマーパターン」はCoddyKit上の無料Lua Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはLua Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Lua Academyコースには全4レッスンが含まれています。
パターンの説明
コルーチンを使ったプロデューサーとコンシューマーのパターンでは、プロデューサーがデータを生成して各項目をyieldし、コンシューマーが項目を1つずつ受け取ります。互いの実装を知る必要はなく、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コンシューマーをコルーチンにする
コンシューマーもコルーチンにできるため、完全に対称なプロデューサーとコンシューマーの組み合わせを作成できます。3つ目の「ドライバー」関数が両者を調整します。これは、プロデューサーとコンシューマーの処理速度が異なるバッファー付きパイプラインに便利です。
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ファイル行プロデューサー
ファイルを遅延読み込みします。プロデューサーのコルーチンが行を1つずつ読み込み、それぞれを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パイプラインでのエラー伝播
コルーチンパイプラインのエラーは、dead状態または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プロデューサーのマージ
複数のプロデューサーを1つにまとめます。稼働中のすべてのプロデューサーをラウンドロビン方式で順番に処理し、各項目を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を使用
よくある質問
「プロデューサー・コンシューマーパターン」レッスンは無料ですか？
はい。「プロデューサー・コンシューマーパターン」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Lua Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Lua Academyコースには全4レッスンが含まれています。
「プロデューサー・コンシューマーパターン」で何を学びますか？
コルーチンを使ってプロデューサー・コンシューマー型のパイプラインを実装します。 ブラウザで直接実行するハンズオンコードでLua Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
Lua Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのLua Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「プロデューサー・コンシューマーパターン」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このLua Academyレッスンでコードを書いて実行できますか？
はい。すべてのLua Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- コルーチンの作成
- resumeとyield
- コルーチンの状態
- プロデューサー・コンシューマーパターン