如何在纯 C# 中创建队列消息代理

Question

背景

我需要一个排队的消息代理以分布式（在连续帧上）的方式调度消息。在下面显示的示例中，它将处理不超过 10 个订阅者，然后在进一步处理之前等待下一帧。

（为了澄清那些不熟悉 Unity3D 的人，Process() 方法使用 Unity 的内置 StartCoroutine() 方法运行，并且 - 在这种情况下 - 将持续游戏的生命周期 - 等待或处理队列。）

所以我有这么一个比较简单的类：

public class MessageBus : IMessageBus
{
    private const int LIMIT = 10;
    private readonly WaitForSeconds Wait;

    private Queue<IMessage> Messages;
    private Dictionary<Type, List<Action<IMessage>>> Subscribers;

    public MessageBus()
    {
        Wait = new WaitForSeconds(2f);
        Messages = new Queue<IMessage>();
        Subscribers = new Dictionary<Type, List<Action<IMessage>>>();
    }

    public void Submit(IMessage message)
    {
        Messages.Enqueue(message);
    }

    public IEnumerator Process()
    {
        var processed = 0;

        while (true)
        {
            if (Messages.Count == 0)
            {
                yield return Wait;
            }
            else
            {
                while(Messages.Count > 0)
                {
                    var message = Messages.Dequeue();

                    foreach (var subscriber in Subscribers[message.GetType()])
                    {
                        if (processed >= LIMIT)
                        {
                            processed = 0;
                            yield return null;
                        }

                        processed++;
                        subscriber?.Invoke(message);
                    }
                }

                processed = 0;
            }
        }
    }

    public void Subscribe<T>(Action<IMessage> handler) where T : IMessage
    {
        if (!Subscribers.ContainsKey(typeof(T)))
        {
            Subscribers[typeof(T)] = new List<Action<IMessage>>();
        }

        Subscribers[typeof(T)].Add(handler);
    }

    public void Unsubscribe<T>(Action<IMessage> handler) where T : IMessage
    {
        if (!Subscribers.ContainsKey(typeof(T)))
        {
            return;
        }

        Subscribers[typeof(T)].Remove(handler);
    }
}

它的工作原理和行为都符合预期，但有一个问题。

问题

我想（从订阅者的角度）这样使用它：

public void Run()
{
    MessageBus.Subscribe<TestEvent>(OnTestEvent);
}

public void OnTestEvent(TestEvent message)
{
    message.SomeTestEventMethod();
}

但这显然失败了，因为Action<IMessage> 无法转换为Action<TestEvent>。

我可以使用它的唯一方法是这样的：

public void Run()
{
    MessageBus.Subscribe<TestEvent>(OnTestEvent);
}

public void OnTestEvent(IMessage message)
{
    ((TestEvent)message).SomeTestEventMethod();
}

但这感觉不雅且非常浪费，因为每个订阅者都需要自己进行选角。

我尝试过的

我正在尝试这样的“投射”动作：

public void Subscribe<T>(Action<T> handler) where T : IMessage
{
    if (!Subscribers.ContainsKey(typeof(T)))
    {
        Subscribers[typeof(T)] = new List<Action<IMessage>>();
    }

    Subscribers[typeof(T)].Add((IMessage a) => handler((T)a));
}

这适用于 subscribe 部分，但显然不适用于 unsubscribe。我可以在某个新创建的 handler-wrapper-lambdas 中缓存以供退订时使用，但老实说，我认为这不是真正的解决方案。

问题

我怎样才能让它按我的意愿工作？如果可能的话，最好使用一些 C#“魔法”，但我知道它可能需要完全不同的方法。

另外因为这将在游戏中使用，并在其生命周期内运行，如果可能的话，我想要一个无垃圾的解决方案。

Answer 1

所以问题是您试图将不同类型的列表作为值存储在订阅者字典中。

解决此问题的一种方法可能是存储List<Delegate>，然后使用Delegate.DynamicInvoke。

下面是一些总结要点的测试代码：

Dictionary<Type, List<Delegate>> Subscribers = new Dictionary<Type, List<Delegate>>();

void Main()
{
    Subscribe<Evt>(ev => Console.WriteLine($"hello {ev.Message}"));
    IMessage m = new Evt("spender");
    foreach (var subscriber in Subscribers[m.GetType()])
    {
        subscriber?.DynamicInvoke(m);
    }
}

public void Subscribe<T>(Action<T> handler) where T : IMessage
{
    if (!Subscribers.ContainsKey(typeof(T)))
    {
        Subscribers[typeof(T)] = new List<Delegate>();
    }
    Subscribers[typeof(T)].Add(handler);
}

public interface IMessage{}

public class Evt : IMessage
{
    public Evt(string message)
    {
        this.Message = message;
    }
    public string Message { get; }
}