Azureで音声バイナリを.NET Streamとして扱う

Posted at 2026-10-07

Azureで音声バイナリを.NET Streamとして扱う

Event HubsとSystem.IO.Pipelinesによる復元力のある設計

音声をリアルタイムで取得して、.NETアプリケーションで処理したいことがある。

たとえば次のようなケース。

Teams通話の録音
SIPやVoIPの音声処理
Speech to Text
音声をLLMやAIへ渡す
WAVなどの音声ファイルとして保存する
リアルタイム音声監視
複数の後続処理へ音声を配信する

単純に音声データを Stream に流すだけなら難しくない。

問題になるのは、

処理途中でVMやプロセスが落ちた場合、音声を失わず復旧できるか

という点。

この記事では、Azure Event Hubsと.NETのSystem.IO.Pipelinesを組み合わせて、音声バイナリを「復元可能なStream」として扱う構成を考える。

音声は結局バイナリデータ

マイク、Teams、SIPなどから受け取る音声は、アプリケーションから見ると基本的にはバイナリデータになる。

たとえばPCMなら、一定時間ごとに次のようなデータを受け取る。

byte[] audioBytes;

20ms単位で取得するなら、音声は次のような連続したPacketになる。

Packet 000001
Packet 000002
Packet 000003
Packet 000004
Packet 000005
...

通常の.NETアプリケーションなら、そのままStreamへ書き込める。

await outputStream.WriteAsync(audioBytes);

この構成自体は非常に単純。

ただし、Streamだけでは耐障害性がない。

Streamだけでは復旧できない

たとえば次の構成を考える。

音声入力
   ↓
.NET Application
   ↓
Stream
   ↓
Speech / WAV / AI

アプリケーションが正常に動いている間は問題ない。

途中でプロセスが停止すると、

Packet 1001
Packet 1002
Packet 1003
       ↓
     障害
Packet 1004
Packet 1005

1004以降を失う可能性がある。

TCPやWebSocketを使っていても、

接続が切断される前にアプリケーションが処理できなかったデータ

を後から再取得できるとは限らない。

必要なのは単なる通信路ではなく、

一度受け取った音声Packetを保持し、必要なら再読込できる仕組み

になる。

Azure Event Hubsを音声Packetのバッファとして使う

この用途で使いやすいAzureサービスの一つがAzure Event Hubs。

Event Hubsへ音声を文字列として送る必要はない。

バイナリデータをそのままEventDataとして送信できる。

var eventData = new EventData(audioBytes);

await producer.SendAsync(
    new[] { eventData });

Base64などへの変換も基本的には不要。

構成は次のようになる。

Teams / SIP / Microphone
          │
          │ PCM / Opus / μ-law
          ▼
 ┌─────────────────┐
 │ .NET Audio Input│
 └────────┬────────┘
          │
          │ byte[]
          ▼
 ┌─────────────────┐
 │ Azure Event Hubs│
 └────────┬────────┘
          │
          ▼
 ┌────────────────────┐
 │ .NET Audio Consumer│
 └─────────┬──────────┘
           │
           ▼
      Audio Stream

Event Hubsが、通常のStreamと大きく違うところは、イベントを一定期間保持できること。

Consumerが停止しても、再起動後に過去のイベントを読み直せる。

音声PacketにはSequence番号を持たせる

音声処理で特に重要なのが順番。

次のようなデータ構造にしておくと扱いやすい。

public record AudioPacket(
    Guid CallId,
    long SequenceNo,
    DateTimeOffset Timestamp,
    byte[] Audio);

例：

CallId      : abc123
SequenceNo  : 1001
Timestamp   : 10:23:45.020
Audio       : byte[]

次のPacketは、

CallId      : abc123
SequenceNo  : 1002
Timestamp   : 10:23:45.040
Audio       : byte[]

となる。

20msごとのPCMなら、

1001 → 20ms
1002 → 20ms
1003 → 20ms
1004 → 20ms

という連続した音声として再構成できる。

Partition KeyにはCallIdを使う

Event Hubsでは、同一Partition内のイベント順序を利用できる。

音声の場合、同じ通話のPacketを同じPartitionへ送る必要がある。

PartitionKey = CallId

という設計が自然。

var options = new SendEventOptions
{
    PartitionKey = callId.ToString()
};

await producer.SendAsync(
    events,
    options);

これによって、

というイベントが同じPartitionへ送られる。

別の通話は別のPartitionへ分散可能。

Partition 0
Call A
1001
1002
1003

Partition 1
Call B
2001
2002
2003

Event Hubsから.NET Streamへ戻す

Consumer側では、Event Hubsから受け取った音声をStreamへ書けばよい。

概念的には次のようになる。

await foreach (PartitionEvent partitionEvent
    in consumer.ReadEventsAsync())
{
    ReadOnlyMemory<byte> audio =
        partitionEvent.Data.EventBody.ToMemory();

    await outputStream.WriteAsync(audio);
}

これで、

Event Hubs
    ↓
byte[]
    ↓
Stream

という変換ができる。

後続処理から見れば普通の.NETの音声Streamとして扱える。

.NETならSystem.IO.Pipelinesが使いやすい

リアルタイム音声では、単純なMemoryStreamよりSystem.IO.Pipelinesが使いやすい場合がある。

構成は次の形。

Azure Event Hubs
       ↓
EventProcessorClient
       ↓
    PipeWriter
===================
 System.IO.Pipelines
===================
    PipeReader
       ↓
Speech / AI / Recorder

書き込み側。

Pipe pipe = new();

PipeWriter writer = pipe.Writer;

await writer.WriteAsync(audioBytes);

読み込み側。

PipeReader reader = pipe.Reader;

while (true)
{
    ReadResult result = await reader.ReadAsync();

    ReadOnlySequence<byte> buffer = result.Buffer;

    // 音声処理

    reader.AdvanceTo(buffer.End);

    if (result.IsCompleted)
        break;
}

System.IO.Pipelinesは、

Producer
Consumer

を分離しやすい。

音声取得速度と音声処理速度に差がある場合にも扱いやすい。

Azure Event HubsとSystem.IO.Pipelinesの役割は違う

この2つは競合する技術ではない。

役割が違う。

Azure Event Hubs
        │
        │ Azure上の耐障害ストリーム
        ▼
System.IO.Pipelines
        │
        │ プロセス内の高速ストリーム
        ▼
Audio Processing

役割を分けると次のようになる。

機能	Event Hubs	System.IO.Pipelines
プロセス間通信	○	×
VM間通信	○	×
Azureサービス間通信	○	×
バイナリ	○	○
再読込	○	×
障害復旧	○	×
高速メモリ処理	△	◎
.NET Stream処理	△	◎
Backpressure	△	◎

Event Hubsが外側。

Pipelinesが内側。

と考えると分かりやすい。

障害復旧

Event Hubsを使う大きな理由がここ。

たとえばConsumerがPacket 1003まで処理したとする。

1001 処理済み
1002 処理済み
1003 処理済み

----- checkpoint -----

1004
1005
1006

ここでVMが停止。

          ×
     VM Crash

再起動後、checkpointから処理を再開する。

1004
1005
1006

これにより、単純なNetwork Streamより復旧しやすい。

Blob Storageも組み合わせる

Event Hubsを録音データの永久保存先として考える必要はない。

音声保存にはBlob Storageを使う方が自然。

                       ┌→ Speech to Text
                       │
Audio → Event Hubs ────┼→ AI
                       │
                       └→ WAV生成
                              ↓
                         Blob Storage

役割を分ける。

Event Hubs

リアルタイム配送
一時保持
再処理
障害復旧

Blob Storage

長期保存
録音ファイル
監査
後日再解析

複数のConsumerにも配信できる

音声処理は一つとは限らない。

                     ┌→ Speech to Text
                     │
Audio → Event Hubs ──┼→ 録音
                     │
                     ├→ AI解析
                     │
                     └→ モニタリング

Consumer Groupを分ければ、それぞれ独立して読み取れる。

Consumer Group : speech
Consumer Group : recording
Consumer Group : ai

音声取得側が各処理を直接呼び出す必要がなくなる。

これはシステム設計上かなり大きい。

Azure Web PubSubという選択肢

リアルタイム性だけを考えるならAzure Web PubSubも候補になる。

WebSocketを使ってバイナリデータをリアルタイム配信できる。

Audio
 ↓
WebSocket
 ↓
Azure Web PubSub
 ↓
Client

用途としては、

ブラウザへ音声をリアルタイム配信
リアルタイムモニタ
双方向通信

などに向いている。

Event Hubsとの違いは設計思想。

Web PubSub
     ↓
「今、このデータを届ける」

Event Hubs
     ↓
「イベントを記録して、Consumerが読む」

今回のテーマである、

音声を途中で失いたくない

という要件ではEvent Hubsの方が扱いやすい。

Windows VM内だけならもっと簡単

Azure上のWindows VM内部で完結する場合、Event Hubsを使わない選択肢もある。

.NETには次の仕組みがある。

System.IO.Pipelines
System.Threading.Channels
NamedPipeServerStream
NamedPipeClientStream

単一プロセス内部なら、

System.IO.Pipelines

複数プロセスなら、

NamedPipe

も使える。

ただしVM自体が落ちた場合にはデータを保持できない。

耐障害性が必要なら永続的な仕組みと組み合わせる必要がある。

最終的な構成

自分なら次の形にする。

Teams / SIP / Audio Device
            │
            │
            ▼
       Audio Receiver
            │
            │
       AudioPacket
            │
            ▼
     Azure Event Hubs
            │
      ┌─────┴─────┐
      │           │
      ▼           ▼
 Recording     Realtime AI
 Consumer       Consumer
      │           │
      ▼           ▼
System.IO      System.IO
.Pipelines     .Pipelines
      │           │
      ▼           ▼
 WAV / Blob    Speech / LLM

音声Packetは、

CallId
SequenceNo
Timestamp
AudioFormat
AudioBytes

程度の情報を持たせる。

重要なのは「Streamを保存する」のではない

この構成で考え方を間違えやすいポイントがある。

Azureに、

.NET Stream

そのものを保存するわけではない。

保存するのは、

Audio Packet

になる。

Packet
Packet
Packet
Packet
Packet

を受信側で、

Stream

へ再構成する。

概念としては、

Durable Packet Stream

        ↓

System.IO.Stream

になる。

まとめ

音声を.NETでリアルタイム処理するだけならStreamで十分。

障害復旧まで考えると、それだけでは足りない。

Azure Event Hubsを間に置くことで、

音声Packet
   ↓
Event Hubs
   ↓
再読込可能
   ↓
.NET Stream

という構成にできる。

さらにVM内ではSystem.IO.Pipelinesを使う。

Azure Event Hubs
       ↓
EventProcessorClient
       ↓
System.IO.Pipelines
       ↓
Speech / AI / WAV

という組み合わせになる。

ポイントは次の3つ。

音声をbyte[]のPacketとして扱う
CallIdとSequenceNoを持たせる
Event Hubsを「復元可能なStream」として使い、VM内ではSystem.IO.Pipelinesへ変換する

音声処理を単なるSocket通信として考えるより、

音声をイベントストリームとして扱う

と考えた方が、Azure上では耐障害性、スケール、再処理を設計しやすくなる。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up