こんにちは。デバイスソフトウエア開発部の本間です。
弊社では、様々な IoT 機器を開発・販売しております。 私もこれまで、いくつかの IoT 機器のファームウェア開発に携わってきました。
IoT 機器に求められる機能は、センサデータの収集、異常の検知、クラウドとの連携、機器状態の管理など多岐にわたります。 これらの機能をどのような仕組みで実現するかを考えることは、IoT 機器のシステム設計における面白さの一つだと感じております。
今回は、そのようなシステム設計の一例として、Fluent Bit をイベントルーターとして利用し、複数のプロセスをイベントで連携させる構成を紹介します。

複数プロセスによる設計
組込みシステムと一言でいっても、ハードウェアの構成や性能はシステム毎に様々であり、ソフトウェアの設計もそれらの条件によって大きく変わります。
低リソースだったり、ハードリアルタイム性が求められたりするデバイスでは、ベアメタルや RTOS を採用することも多いかと思います。 一般的なベアメタル環境や多くの RTOS では、汎用 OS のようにユーザープロセスを実行する仕組みを持たないため、 デバイスドライバや制御ロジックなどのプログラムを単一の実行形式へリンクし、その中で関数やモジュール、RTOS のタスクとして責務を分けていく構成となります。
一方、組込み Linux などの汎用 OS が動作する比較的リソースに余裕のあるデバイスでは、プロセスを利用できます。 このようなデバイスでは、センサデータの収集だけでなく、データの監視・記録、ネットワーク通信、ソフトウェア更新など多くの機能を持つことも少なくありません。
そこで、機能ごとにプロセスを分け、それぞれに責務を持たせる設計が選択肢になります。 責務を適切に分けることができれば、機能同士を疎結合にしやすくなり、システム全体として機能追加や保守を行いやすくなります。
ただし、プロセスを分ければシステム全体が自動的に単純になるわけではありません。 それまで関数呼び出しで済んでいた機能間の連携を、今度はプロセス間通信として設計する必要があります。 通信方法やデータ形式など、プロセス間でデータをやり取りするための取り決めが必要になります。
ここからは、組込み Linux などの汎用 OS を搭載した IoT 機器を対象に、プロセス間の連携について考えてみたいと思います。
イベントでプロセスを連携させる
複数のプロセスを連携させる方法は様々です。 UNIX で古くから使われてきたパイプによるフィルタも、強力な手段の1つでしょう。 今回は、その中でもイベント駆動型アーキテクチャ(Event-Driven Architecture、EDA)の考え方を取り入れてみたいと思います。
EDA の考え方では、「ドアを開けた」「ボタンを押した」といった、システム内で起きたことをイベントとして通知し、必要な処理がそのイベントに反応します。 ある処理が別の処理を同期的に呼び出すのではなく、他の処理へ起きたことを非同期に知らせる点が特徴です。
基本的に、EDA は次の3つの要素で構成されます。
| 構成要素 | 役割 |
|---|---|
| プロデューサー | 出来事をイベントとして送信する |
| イベントルーター | イベントの種類を見て配送先を決める |
| コンシューマー | 受信したイベントを処理する |

プロデューサーは最終的な配送先を知る必要がありません。コンシューマーを追加する場合も、プロデューサーを変更せず、イベントルーター側で配送先を追加できます。Pub/Sub 方式の MQTT で言えば、トピックへの publish がイベントの送信、トピックへの subscribe がコンシューマーの追加(イベント購読)に相当すると言えます。
EDA の考え方は、クラウド上のサービス間だけでなく、デバイス内のプロセス間連携にも適用できます。 本記事では、その一例として、イベントルーターを介して複数のプロセスを連携させる構成を作ってみたいと思います。
Fluent Bit をイベントルーターとして使う
本記事でイベントルーターとして使うのが Fluent Bit です。
Fluent Bit は、ログや各種データを収集・処理・転送できる軽量なツールです。テレメトリーエージェントというツールに分類されるそうです。Input・Filter・Output などを組み合わせてデータパイプラインを構成できます。Input するデータにはタグを付けることができ、そのタグを使って配送先を振り分けられます。
Fluent Bit は、MQTT や NATS のようなメッセージングシステムでも、ZeroMQ のようなメッセージングライブラリでもありません。一方で、ルーティングやフィルタリング、バッファリングなど、イベント配送にも利用できる機能を備えています。
IoT 機器のようにプロセス構成が比較的固定されているシステムであれば、イベントの配送経路の設定を各プロセスから切り離し、Fluent Bit の設定へ集約できます。今回は、この特徴を利用して、ホスト内のイベントを事前に設定した宛先へ配送するイベントルーターとして使ってみます。
そのような理由から本記事では Fluent Bit を使ってみますが、イベント連携の用途では、一般的には次のような技術がよく使われるようです。
| 技術・仕組み | 向いているケース |
|---|---|
| ZeroMQ | ブローカーレスの高速なプロセス間通信 |
| MQTT | サーバ・デバイス間の通信、QoS による配送制御 |
| NATS | 高速なプロセス間通信、JetStream による永続化・再配送 |
Fluent Bit でシステムを組んでみる
それでは Fluent Bit で簡単なイベント駆動システムを組んでみましょう。 温度の計測、温度の監視、LED の制御、データの記録をそれぞれプロセスに分け、Fluent Bit を介して連携させます。
サンプルは Linux 上での実行を想定しており、Fluent Bit 5.1.1、Python 3.10 で動作確認しています。温度センサと GPIO の読み書きについては、サンプルを実行しやすいように模擬実装とします。
システム構成
各プロセスの役割は次のとおりです。
| プロセス | 役割 |
|---|---|
temp-sensor |
ADC から値を1秒ごとに読み取り、温度データを生成する |
monitor |
温度を監視し、高温警報状態と LED の設定指示を送信する |
led |
LED の設定指示を受けて GPIO 出力へ反映する |
recorder |
最新の温度と高温警報状態を30秒ごとに保存する |

各プロセスは他のプロセスへ直接データを送らず、Fluent Bit を経由してイベントを送受信します。 例えば、temp-sensor が送信した温度イベントは、Fluent Bit によって monitor と recorder の両方へ配送されます。
プロセス間でやり取りするイベントは次の3つです。
| イベント | タグ | プロデューサー | 配送先 |
|---|---|---|---|
| 温度計測 | sensor.temperature |
temp-sensor |
monitor、recorder |
| 高温警報状態 | alert.high-temperature |
monitor |
recorder |
| LED設定 | led.set |
monitor |
led |
Fluent Bit の設定
fluent-bit.yaml を次のように作成します。
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 |
service: flush: 1 log_level: info pipeline: inputs: - name: forward listen: 127.0.0.1 port: 24224 outputs: # sensor.temperature -> monitor, recorder - name: udp match: sensor.temperature host: 127.0.0.1 port: 26001 format: json_lines - name: udp match: sensor.temperature host: 127.0.0.1 port: 26003 format: json_lines # led.set -> led - name: udp match: led.set host: 127.0.0.1 port: 26002 format: json_lines # alert.high-temperature -> recorder - name: udp match: alert.high-temperature host: 127.0.0.1 port: 26003 format: json_lines # 全イベントをファイルへ保存 - name: file match: '*' path: /tmp file: fluent-debug.log format: out_file |
Inputs の Forward Input は、127.0.0.1:24224 でイベントを受信します。listen の既定値は 0.0.0.0 なので、別のホストから接続されないよう、ループバックアドレスを明示しています。
Outputs では、タグに応じて各 Output へルーティングし、UDP Output からレコードを JSON Lines 形式で各コンシューマーへ送信します。sensor.temperature は monitor と recorder へ、led.set は led へ、alert.high-temperature は recorder へ配送します。最後の File Output では、デバッグ用にすべてのイベントを /tmp/fluent-debug.log へ記録します。
設定ファイルは次のコマンドで検証できます。
|
1 |
$ fluent-bit --dry-run -c fluent-bit.yaml |
Forward Protocol によるイベント送信
プロデューサーから Fluent Bit へのイベント送信には、Python の fluent-logger を使います。まず、fluent-logger パッケージをインストールします。
|
1 2 3 |
$ python3 -m venv .venv $ . .venv/bin/activate $ python -m pip install fluent-logger |
Fluent Bit の Forward Input が受け取るデータは、Forward Protocol に従った MessagePack です。FluentSender.emit() で送信するイベントは、Message mode に従った次の形式になります。
|
1 |
[tag, time, record] |
例えば、FluentSender("sensor") に対して emit("temperature", record) を呼ぶと、タグは sensor.temperature になります。
|
1 2 3 4 5 6 7 8 |
[ "sensor.temperature", time, { "event": "sensor.temperature", "celsius": 30.4 } ] |
Fluent Bit からコンシューマーへの送信には、UDP Output の json_lines 形式を使っています。そのため、コンシューマー側では JSON レコードを1件ずつ簡単に受信できます。
受信処理は json_lines_server.py にまとめます。受信したデータを行ごとに JSON として読み込むだけのシンプルな処理です。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
import json import logging import socket def receive(port): with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server: server.bind(("127.0.0.1", port)) logging.info("listening on UDP 127.0.0.1:%d", port) while True: packet, _ = server.recvfrom(65535) for line in packet.splitlines(): yield json.loads(line) |
各プロセスを実装する
それでは、temp-sensor、monitor、led、recorder の各プロセスを実装していきます。
温度計測プロセス(temp-sensor)
まず、temp_sensor.py で模擬温度を1秒ごとに送信します。
なお、このサンプルでは、実際の温度センサーからの ADC 入力を SAMPLES で模擬しています。実機では、IIO などを使って ADC の値を読み取ります。
|
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 30 31 32 |
import logging import time from itertools import cycle from fluent import sender # ADC からの入力を模擬する SAMPLES = (27.8, 29.2, 30.4, 31.1, 28.9, 29.7, 30.2) logging.basicConfig( format="%(asctime)s.%(msecs)03d [temp-sensor] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO, ) def main(): with sender.FluentSender("sensor", host="127.0.0.1") as publisher: for celsius in cycle(SAMPLES): event = { "event": "sensor.temperature", "celsius": celsius, } if not publisher.emit("temperature", event): raise ConnectionError(publisher.last_error) logging.info("%.1f C", celsius) time.sleep(1) if __name__ == "__main__": main() |
状態監視プロセス(monitor)
monitor.py は温度が30 ℃以上かどうかを判定し、高温警報状態が変わったときだけ alert.high-temperature と led.set を送信します。
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
import logging from fluent import sender from json_lines_server import receive THRESHOLD = 30.0 logging.basicConfig( format="%(asctime)s.%(msecs)03d [monitor] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO, ) def emit(publisher, label, event): if not publisher.emit(label, event): raise ConnectionError(publisher.last_error) def main(): last_active = None with ( sender.FluentSender("alert", host="127.0.0.1") as alert_sender, sender.FluentSender("led", host="127.0.0.1") as led_sender, ): for event in receive(26001): active = event["celsius"] >= THRESHOLD if active == last_active: continue emit(alert_sender, "high-temperature", { "event": "alert.high-temperature", "active": active, }) emit(led_sender, "set", { "event": "led.set", "on": active, }) last_active = active logging.info("alert=%s", active) if __name__ == "__main__": main() |
LED 設定プロセス(led)
led.py は led.set を受信し、状態が変わったときだけ LED を更新します。
なお、このサンプルでは、実際の LED を制御する GPIO 操作をログ出力で代用しています。実機では、libgpiod などで GPIO を操作します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
import logging from json_lines_server import receive logging.basicConfig( format="%(asctime)s.%(msecs)03d [led] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO, ) def main(): last_state = None for event in receive(26002): state = event["on"] if state != last_state: # GPIO への出力を模擬する logging.info("%s", "ON" if state else "OFF") last_state = state if __name__ == "__main__": main() |
記録プロセス(recorder)
最後に、recorder.py で最新の温度と高温警報状態をまとめ、30秒ごとに /tmp/record-data.jsonl へ保存します。イベント識別用の event は状態の更新にだけ使い、保存するレコードからは除きます。
|
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
import json import logging import time from datetime import datetime from json_lines_server import receive OUTPUT = "/tmp/record-data.jsonl" RECORD_INTERVAL = 30 logging.basicConfig( format="%(asctime)s.%(msecs)03d [recorder] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO, ) def main(): temperature = None high_temperature = None next_record_at = None with open(OUTPUT, "a", encoding="utf-8", buffering=1) as stream: for event in receive(26003): if event["event"] == "sensor.temperature": temperature = event["celsius"] elif event["event"] == "alert.high-temperature": high_temperature = event["active"] if temperature is None or high_temperature is None: continue now = time.monotonic() if next_record_at is None: next_record_at = now + RECORD_INTERVAL continue if now < next_record_at: continue next_record_at = now + RECORD_INTERVAL record = { "time": datetime.now().astimezone().isoformat(timespec="milliseconds"), "temperature": temperature, "high_temperature": high_temperature, } print(json.dumps(record), file=stream) if __name__ == "__main__": main() |
動作確認
Fluent Bit と各プロセスを、別々のターミナルで起動します。
|
1 2 3 4 5 |
$ fluent-bit -c fluent-bit.yaml $ python monitor.py $ python led.py $ python recorder.py $ python temp_sensor.py |
それでは、temp-sensor、monitor、led などのログを確認してみましょう。
temp-sensor では、次のように温度を1秒ごとに送信しています。
|
1 2 3 4 5 |
2026-08-28 10:15:20.214 [temp-sensor] 27.8 C 2026-08-28 10:15:21.215 [temp-sensor] 29.2 C 2026-08-28 10:15:22.216 [temp-sensor] 30.4 C 2026-08-28 10:15:23.217 [temp-sensor] 31.1 C 2026-08-28 10:15:24.218 [temp-sensor] 28.9 C |
monitor では、温度に応じて高温警報状態が切り替わっています。
|
1 2 3 4 5 |
2026-08-28 10:15:19.802 [monitor] listening on UDP 127.0.0.1:26001 2026-08-28 10:15:21.005 [monitor] alert=False 2026-08-28 10:15:23.004 [monitor] alert=True 2026-08-28 10:15:25.004 [monitor] alert=False 2026-08-28 10:15:28.004 [monitor] alert=True |
led では、高温警報状態に応じた ON/OFF 指示を受け、LED 状態が切り替わっています。
|
1 2 3 4 5 |
2026-08-28 10:15:19.810 [led] listening on UDP 127.0.0.1:26002 2026-08-28 10:15:22.005 [led] OFF 2026-08-28 10:15:24.005 [led] ON 2026-08-28 10:15:26.005 [led] OFF 2026-08-28 10:15:29.005 [led] ON |
recorder のデータ記録内容は次のようになります。30秒周期でデータ記録されていることが確認できます。
|
1 2 3 4 5 6 |
$ tail -f /tmp/record-data.jsonl {"time": "2026-08-28T10:16:00.004+09:00", "temperature": 28.9, "high_temperature": false} {"time": "2026-08-28T10:16:30.011+09:00", "temperature": 30.2, "high_temperature": true} {"time": "2026-08-28T10:17:00.018+09:00", "temperature": 29.2, "high_temperature": false} {"time": "2026-08-28T10:17:30.025+09:00", "temperature": 31.1, "high_temperature": true} {"time": "2026-08-28T10:18:00.032+09:00", "temperature": 29.7, "high_temperature": false} |
動作確認中に発生した全てのイベントは、デバッグファイルから確認できます。
|
1 2 3 4 5 6 |
$ tail -f /tmp/fluent-debug.log sensor.temperature: [1787912122.000000000, {"event":"sensor.temperature","celsius":30.4}] alert.high-temperature: [1787912123.000000000, {"event":"alert.high-temperature","active":true}] led.set: [1787912123.000000000, {"event":"led.set","on":true}] sensor.temperature: [1787912123.000000000, {"event":"sensor.temperature","celsius":31.1}] sensor.temperature: [1787912124.000000000, {"event":"sensor.temperature","celsius":28.9}] |
このように、各プロセスに責務を分け、Fluent Bit を利用して簡単なイベント駆動システムを構築することができました。
まとめ
今回は、組込み Linux などの汎用 OS 上で機能を複数のプロセスに分け、イベントで連携させる構成を作りました。複数プロセスに分けることで責務を分けやすくなる一方、プロセス間の連携方法についても設計する必要があります。
本記事では Fluent Bit をイベントルーターとして利用してみました。イベントの配送経路を各プロセスから切り離して Fluent Bit の設定にまとめることで、各プロセスは自身の処理に集中できる構成にしています。
サンプルの UDP による配送はベストエフォートですが、TCP Output や Forward Output、永続バッファリングなどを利用して、配送の信頼性を高めることもできます。ただし、これらはコンシューマーでの処理完了まで保証するものではありません。より高度な配送保証や動的な購読、イベントの保存・再生などが必要な場合は、MQTT や NATS なども含め、要件に合った仕組みを選ぶことが重要です。






