akkadotnet/akka.net

static class CollectionExtensions
{
    public static T RemoveFirst<T>(this ICollection<T> items)
    {
        T item = items.FirstOrDefault();
        if (item != null)
        {
            items.Remove(item);
        }
        return item;
    }
}

public class ThrottledMessenger : ReceiveActor
{
    private readonly HashSet<IActorRef> workerPool;
    private readonly HashSet<IActorRef> waitingActors;
    private IActorRef originalSender;
    private object originalMessage;

    public ThrottledMessenger(int parallelWorkers, IEnumerable<IActorRef> refs)
    {
        workerPool = new HashSet<IActorRef>(refs.Take(parallelWorkers));
        waitingActors = new HashSet<IActorRef>(refs.Skip(parallelWorkers));

        ReceiveAny(x =>
        {
            originalSender = Sender;
            originalMessage = x;

            foreach (var aref in workerPool) aref.Tell(x);
            Become(Messenging);
        });
    }

    private void Messenging()
    {
        ReceiveAny(msg =>
        {
            if (workerPool.Remove(Sender))
            {
                originalSender.Tell(msg, Sender);

                var nextActor = waitingActors.RemoveFirst();
                if (nextActor != null)
                {
                    workerPool.Add(nextActor);
                    nextActor.Tell(originalMessage);
                }
                else if (workerPool.Count == 0)
                {
                    Context.Stop(Self);
                }
            }
        });
    }
}