Очередь Tasks с отложенным выполнением каждой задачи

Question

Пишу веб приложение, которое уведомляет пользователя об изменениях произошедших в базе данных сайта. Веб сервер этого сайта отслеживает изменения и рассылает их всем подключенным клиентам. Мне нужно лишь принять изменение, обработать и отправить юзеру

Собственно проблема: измениться может запись о товаре, мне нужно уведомить лишь при изменении одного параметра - цены. На сервере реализованы атомарные сохранения - изменения летят в клиентов, при каждой введенной новой цифре.

Мне нужно реализовать механизм задержки перед отправкой по следующей схеме:

получаю изменение по товару с id=1 -> жду какое-то время -> если новых изменений по конкретно этотму товару не было -> отправляю изменение
если же изменения были -> обновляю таймер ожидания

*если товар другой, то для него запускается такая-же логика, но свой таймер

Основная логика приложения:

public static async Task Main()
    {
        // событие, которое висит на соединении и отлавливает уведомления от сервера 
        Connection.On<NotificationData>("Entity", async message=>
        {
        // здесь обработка уведомления
        // ...
        // результатом получаю id и разницу в цене

        // в итоге я понимаю что изменилась цена и у какого товара (его id) она изменилась
           var info = $"На товар c id = {id} цена выросла на {priceDiff} рублей"

           await SendMessage(info)
           // этим SendMessage() я как раз отправляю информацию пользователям
        }
    }

Посоветуйте, как лучше реализовать такую схему ожидания. О каких технологиях нужно почитать для этого? Я пока даже не знаю, что и где искать(

Answer 1

Для решения этой задачи подойдет примитивная очередь сообщений:

class MyQueue<TKey, TData> where TKey : IEquatable<TKey> {
    public static MyQueue<TKey, TData> Run() {
        var queue = new MyQueue<TKey, TData>();
        ThreadStart ts = new ThreadStart(queue.Worker);
        Thread thread = new Thread(ts);
        thread.Start();
        queue.workerThread = thread;
        return queue;
    }
    private MyQueue() { }
    readonly IDictionary<TKey, MyTask> queue = new Dictionary<TKey, MyTask>();
    readonly object syncRoot = new object();
    Thread workerThread;
    public void AddOrUpdate(TKey key, TData data, Action<TData> action) {
        lock(syncRoot) {
            MyTask task;
            bool found = queue.TryGetValue(key, out task);
            if(found)
                task.Update(data);
            else {
                task = new MyTask(data, action);
                queue.Add(key, task);
            }
        }
    }
    public void Stop() {
        workerThread.Abort();
    }
    public void Worker() {

        while(true) {
            Thread.Sleep(100);
            lock(syncRoot) {
                IList<TKey> keys = new List<TKey>(queue.Keys);
                foreach(TKey key in keys) {
                    MyTask task = queue[key];
                    if(task.ScheduledOn < DateTime.Now) {
                        task.Action(task.Data);
                        queue.Remove(key);
                    }
                }
            }
        }
    }
    class MyTask {
        public MyTask(TData data, Action<TData> action) {
            this.Action = action;
            Update(data);
        }
        public Action<TData> Action { get; }
        public DateTime ScheduledOn { get; private set; }
        public TData Data { get; private set; }
        public void Update(TData data) {
            this.Data = data;
            ScheduledOn = DateTime.Now.AddSeconds(1);
        }
    }
}

static void Main(string[] args) {
    var queue = MyQueue<int, string>.Run();
    queue.AddOrUpdate(1, "First Message", ExecuteTask);
    queue.AddOrUpdate(1, "Updated Message", ExecuteTask);
    Thread.Sleep(2000);
    queue.Stop();
}
static void ExecuteTask(string msg) {
    Console.WriteLine(msg);
}

БЛОГ НА HUSL

Очередь Tasks с отложенным выполнением каждой задачи

Ответы (1 шт):