如何向多个消费者广播异步StreamReader？

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;

public class DataStreamBroadcaster
{
    private StreamReader _streamReader;
    private event Action<string> DataArrived;
    private List<Action<string>> _subscribers;

    public DataStreamBroadcaster(Stream stream)
    {
        _streamReader = new StreamReader(stream);
        _subscribers = new List<Action<string>>();
    }

    public void Subscribe(Action<string> subscriber)
    {
        _subscribers.Add(subscriber);
    }

    public void Unsubscribe(Action<string> subscriber)
    {
        _subscribers.Remove(subscriber);
    }

    public async Task StartBroadcastingAsync()
    {
        while (true)
        {
            string data = await _streamReader.ReadLineAsync();
            if (data == null)
                break;

            OnDataArrived(data);
        }
    }

    private void OnDataArrived(string data)
    {
        DataArrived?.Invoke(data);
        foreach (var subscriber in _subscribers)
        {
            subscriber.Invoke(data);
        }
    }
}

// 创建一个数据流广播器
var broadcaster = new DataStreamBroadcaster(stream);

// 创建消费者1
Action<string> consumer1 = data =>
{
    Console.WriteLine("Consumer 1 received data: " + data);
};

// 创建消费者2
Action<string> consumer2 = data =>
{
    Console.WriteLine("Consumer 2 received data: " + data);
};

// 订阅消费者1和消费者2
broadcaster.Subscribe(consumer1);
broadcaster.Subscribe(consumer2);

// 启动广播
await broadcaster.StartBroadcastingAsync();

// 取消订阅消费者2
broadcaster.Unsubscribe(consumer2);