c# - 通过 websocket 连接同步集合

Question

我目前正在开发一个客户端-服务器系统，我正在尝试获取一个集合以跨 websocket 进行同步。一切都在 C# + .Net 4.5 中，我想知道是否有通过 websocket 同步数据的特定最佳实践。这是一种单向同步:

服务器:BindingCollection ----- Websocket -----> 客户端:BindingCollection

集合最多可以有 1000 个对象，每个对象有 20 个字段，所以每次都发送全部似乎有点浪费。

Answer 1

我会使用观察者模式，只发送更改的对象进行同步。

于是终于抽空写了一个小例子。我正在使用一个内存中的通用存储库，它在更改时调用事件。然后将更改发送给所有客户端，这样您就不必发送完整的列表/集合。

一个简单的监控模型

using System;

namespace SynchronizingCollection.Common.Model
{
    public class MyModel
    {
        public Guid Id { get; set; }
        public string Name { get; set; }
        public int Age { get; set; }
    }
}

通用存储库

注意在添加/更新/删除某些内容时调用的事件 OnChange。该事件在 XSockets 长时间运行的 Controller (单例)中“订阅”请参见“RepoMonitor”类

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;

namespace SynchronizingCollection.Server.Repository
{
    /// <summary>
    /// A static generic thread-safe repository for in-memory storage
    /// </summary>
    /// <typeparam name="TK">Key Type</typeparam>
    /// <typeparam name="T">Value Type</typeparam>
    public static class Repository<TK, T>
    {
        /// <summary>
        /// When something changes
        /// </summary>
        public static event EventHandler<OnChangedArgs<TK,T>> OnChange;

        private static ConcurrentDictionary<TK, T> Container { get; set; }

        static Repository()
        {
            Container = new ConcurrentDictionary<TK, T>();
        }

        /// <summary>
        /// Adds or updates the entity T with key TK
        /// </summary>
        /// <param name="key"></param>
        /// <param name="entity"></param>
        /// <returns></returns>
        public static T AddOrUpdate(TK key, T entity)
        {
            var obj = Container.AddOrUpdate(key, entity, (s, o) => entity);
            if(OnChange != null)
                OnChange.Invoke(null,new OnChangedArgs<TK, T>(){Key = key,Value = entity, Operation =  Operation.AddUpdate});
            return obj;
        }

        /// <summary>
        /// Removes the entity T with key TK
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static bool Remove(TK key)
        {
            T entity;
            var result = Container.TryRemove(key, out entity);
            if (result)
            {
                if (OnChange != null)
                    OnChange.Invoke(null, new OnChangedArgs<TK, T>() { Key = key, Value = entity, Operation = Operation.Remove});
            }
            return result;
        }

        /// <summary>
        /// Removes all entities matching the expression f
        /// </summary>
        /// <param name="f"></param>
        /// <returns></returns>
        public static int Remove(Func<T, bool> f)
        {
            return FindWithKeys(f).Count(o => Remove(o.Key));
        }        

        /// <summary>
        /// Find all entities T matching the expression f
        /// </summary>
        /// <param name="f"></param>
        /// <returns></returns>
        public static IEnumerable<T> Find(Func<T, bool> f)
        {
            return Container.Values.Where(f);
        }

        /// <summary>
        /// Find all entities T matching the expression f and returns a Dictionary TK,T
        /// </summary>
        /// <param name="f"></param>
        /// <returns></returns>
        public static IDictionary<TK, T> FindWithKeys(Func<T, bool> f)
        {
            var y = from x in Container
                    where f.Invoke(x.Value)
                    select x;
            return y.ToDictionary(x => x.Key, x => x.Value);
        }

        /// <summary>
        /// Returns all entities as a Dictionary TK,T
        /// </summary>
        /// <returns></returns>
        public static IDictionary<TK, T> GetAllWithKeys()
        {
            return Container;
        }

        /// <summary>
        /// Returns all entities T from the repository
        /// </summary>
        /// <returns></returns>
        public static IEnumerable<T> GetAll()
        {
            return Container.Values;
        }

        /// <summary>
        /// Get a single entity T with the key TK
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static T GetById(TK key)
        {
            return Container.ContainsKey(key) ? Container[key] : default(T);
        }

        /// <summary>
        /// Get a single entity T as a KeyValuePair TK,T with the key TK
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static KeyValuePair<TK, T> GetByIdWithKey(TK key)
        {
            return Container.ContainsKey(key) ? new KeyValuePair<TK, T>(key, Container[key]) : new KeyValuePair<TK, T>(key, default(T));
        }

        /// <summary>
        /// Checks if the repository has a key TK
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public static bool ContainsKey(TK key)
        {
            return Container.ContainsKey(key);
        }
    }
}

事件参数和枚举以了解刚刚发生了什么变化

using System;

namespace SynchronizingCollection.Server.Repository
{
    /// <summary>
    /// To send changes in the repo
    /// </summary>
    /// <typeparam name="TK"></typeparam>
    /// <typeparam name="T"></typeparam>
    public class OnChangedArgs<TK,T> : EventArgs
    {
        public Operation Operation { get; set; }
        public TK Key { get; set; }
        public T Value { get; set; }
    }
}

namespace SynchronizingCollection.Server.Repository
{
    /// <summary>
    /// What kind of change was performed
    /// </summary>
    public enum Operation
    {
        AddUpdate,
        Remove
    }
}

向客户端发送更改的 Controller ...

using System;
using SynchronizingCollection.Common.Model;
using SynchronizingCollection.Server.Repository;
using XSockets.Core.XSocket;
using XSockets.Core.XSocket.Helpers;
using XSockets.Plugin.Framework;
using XSockets.Plugin.Framework.Attributes;

namespace SynchronizingCollection.Server
{
    /// <summary>
    /// Long running controller that will send information to clients about the collection changes
    /// </summary>
    [XSocketMetadata(PluginRange = PluginRange.Internal, PluginAlias = "RepoMonitor")]
    public class RepositoryMonitor : XSocketController
    {
        public RepositoryMonitor()
        {
            Repository<Guid, MyModel>.OnChange += RepositoryOnChanged;
        }

        private void RepositoryOnChanged(object sender, OnChangedArgs<Guid, MyModel> e)
        {
            switch (e.Operation)
            {
                case Operation.Remove:
                    this.InvokeTo<Demo>(p => p.SendUpdates, e.Value,"removed");
                break;                    
                case Operation.AddUpdate:
                    this.InvokeTo<Demo>(p => p.SendUpdates, e.Value, "addorupdated");
                break;                    
            }
        }       
    }
}

客户端调用以添加/删除/更新集合的 XSockets Controller 。

using System;
using SynchronizingCollection.Common.Model;
using SynchronizingCollection.Server.Repository;
using XSockets.Core.XSocket;

namespace SynchronizingCollection.Server
{
    public class Demo : XSocketController
    {
        public bool SendUpdates { get; set; }

        public Demo()
        {
            //By default all clients get updates
            SendUpdates = true;
        }

        public void AddOrUpdateModel(MyModel model)
        {
            Repository<Guid, MyModel>.AddOrUpdate(model.Id, model);
        }

        public void RemoveModel(MyModel model)
        {
            Repository<Guid, MyModel>.Remove(model.Id);
        }        
    }
}

还有一个用 C# 编写的演示客户端，可以添加和删除 10 个不同的对象...但是使用 JavaScript API 也很容易。特别是使用 knockoutjs 在客户端操作集合。

using System;
using System.Threading;
using SynchronizingCollection.Common.Model;
using XSockets.Client40;

namespace SynchronizingCollection.Client
{
    class Program
    {
        static void Main(string[] args)
        {
            var c = new XSocketClient("ws://127.0.0.1:4502","http://localhost","demo");

            c.Controller("demo").OnOpen += (sender, connectArgs) => Console.WriteLine("Demo OPEN");

            c.Controller("demo").On<MyModel>("addorupdated", model => Console.WriteLine("Updated " + model.Name));
            c.Controller("demo").On<MyModel>("removed", model => Console.WriteLine("Removed " + model.Name));

            c.Open();

            for (var i = 0; i < 10; i++)
            {
                var m = new MyModel() {Id = Guid.NewGuid(), Name = "Person Nr" + i, Age = i};
                c.Controller("demo").Invoke("AddOrUpdateModel", m);

                Thread.Sleep(2000);

                c.Controller("demo").Invoke("RemoveModel", m);
                Thread.Sleep(2000);
            }

            Console.ReadLine();
        }
    }
}

您可以从我的保管箱下载项目:https://www.dropbox.com/s/5ljbedovx6ufkww/SynchronizingCollection.zip?dl=0

问候乌菲