3

I have used Observer Pattern for my application.

I have a subject which has one System.Timers.Timer object in it named 'tmr'. The tick event of this timer fires after every 60 seconds. On this tick event I will notify all my observers that are attached to my subject. I have used a for-loop to iterate throught my Observers List & then fire the Observers Update method.

Assume i have 10 observers attached to my subject.

Each observer takes 10 seconds to complete its processing.

Now notification being done in a for-loop causes the last Observer's Update method to be called after 90seconds. i.e. Next Observer Update method is called only after previous one has completed its processing.

But this is not what i wanted in my application. I need all my observers Update method to be fired instantly when the timer tick occurs. So that no observer has to wait. I hope this can be done by Threading.

So, I modified code to,

// Fires the updates instantly
    public void Notify()
    {
      foreach (Observer o in _observers)
      {
        Threading.Thread oThread = new Threading.Thread(o.Update);
        oThread.Name = o.GetType().Name;
        oThread.Start();
      }
    }

But I have two doubts in my mind,

  1. If there are 10 observers And my timer interval is 60 seconds Then the statement new Thread() will fire 600 times.

    Is it efficient and recommended to create new threads on every timer tick ?

  2. What if my observers are taking too much time to complete their update logic i.e. goes more than 60seconds. Means the timer tick occurs before the observers are updated. How can i control this ?

I can post sample code.. if required...

The code i used..

using System;
using System.Collections.Generic;
using System.Timers;
using System.Text;
using Threading = System.Threading;
using System.ComponentModel;

namespace singletimers
{
  class Program
  {


    static void Main(string[] args)
    {
      DataPullerSubject.Instance.Attach(Observer1.Instance);
      DataPullerSubject.Instance.Attach(Observer2.Instance);
      Console.ReadKey();
    }
  }

  public sealed class DataPullerSubject
  {
    private static volatile DataPullerSubject instance;
    private static object syncRoot = new Object();
    public static DataPullerSubject Instance
    {
      get
      {
        if (instance == null)
        {
          lock (syncRoot)
          {
            if (instance == null)
              instance = new DataPullerSubject();
          }
        }

        return instance;
      }
    }

    int interval = 10 * 1000;
    Timer tmr;
    private List<Observer> _observers = new List<Observer>();

    DataPullerSubject()
    {
      tmr = new Timer();
      tmr.Interval = 1; // first time to call instantly
      tmr.Elapsed += new ElapsedEventHandler(tmr_Elapsed);
      tmr.Start();
    }

    public void Attach(Observer observer)
    {
      _observers.Add(observer);
    }

    public void Detach(Observer observer)
    {
      _observers.Remove(observer);
    }

    // Fires the updates instantly
    public void Notify()
    {
      foreach (Observer o in _observers)
      {
        Threading.Thread oThread = new Threading.Thread(o.Update);
        oThread.Name = o.GetType().Name;
        oThread.Start();
      }
    }

    private void tmr_Elapsed(object source, ElapsedEventArgs e)
    {
      tmr.Interval = interval;
      tmr.Stop(); // stop the timer until all notification triggered
      this.Notify();
      tmr.Start();//start again
    }
  }


  public abstract class Observer
  {
    string data;
    public abstract void Update();
    public virtual void GetDataFromDBAndSetToDataSet(string param)
    {
      Console.WriteLine("Processing for: " + param);
      data = param + new Random().Next(1, 2000);
      Threading.Thread.Sleep(10 * 1000);//long work
      Console.WriteLine("Data set for: " + param);
    }
  }


  public sealed class Observer1 : Observer
  {
    private static volatile Observer1 instance;
    private static object syncRoot = new Object();
    public static Observer1 Instance
    {
      get
      {
        if (instance == null)
        {
          lock (syncRoot)
          {
            if (instance == null)
              instance = new Observer1();
          }
        }

        return instance;
      }
    }
    Observer1()
    {
    }
    public override void Update()
    {
      base.GetDataFromDBAndSetToDataSet("Observer1");
    }

  }

  public sealed class Observer2 : Observer
  {
    private static volatile Observer2 instance;
    private static object syncRoot = new Object();
    public static Observer2 Instance
    {
      get
      {
        if (instance == null)
        {
          lock (syncRoot)
          {
            if (instance == null)
              instance = new Observer2();
          }
        }

        return instance;
      }
    }
    Observer2()
    {
    }
    public override void Update()
    {
      base.GetDataFromDBAndSetToDataSet("Observer2");
    }

  }
}

Thanks & kind regards.

1

3 Answers 3

1
  • Using new Thread is discouraged. Use Task or Task<T>
  • Your best attempt at creating an Observable pattern framework will probably only come close to Rx. Use that which solves the problems you mentioned (i.e. if processing takes too much time). Rx will give you huge flexibility in defining your observable scenarios.
0

1) You can use Threads from the ThreadPool via ThreadPool.QueueUserWorkItem or you can use Tasks

2) You have to synchronize your Methods.

0

Alternatively, the observers could implement Update in a non-blocking way. That is, Update always returns immediately. Then it is the responsibility of the Observer objects to perform their work in a new thread if necessary.

I'm not sure if this helps in your scenario - I don't know what your 'Observers' are, but then maybe you don't know either?

3
  • @Garen: I didnt understand what you ment by "non-blocking way". In my Observers I get data from database. So here its possible that it will take more than couple of minutes. DOing so each observers update their public field.
    – thinkmmk
    Apr 13, 2011 at 15:52
  • @thinkmmk - I mean that the call to Update returns immediately, and the Observers are responsible for starting the Update operation in a separate thread. the Observers are also responsible for either queueing or ignoring Update calls when an Update is already in progress. But it depends on your Observers as to whether this would be a good decision - you want to avoid duplicating the same logic in different 'Observer' classes.
    – GarethOwen
    Apr 14, 2011 at 6:54
  • I have edited my original post with code.. now i m struggling for synchronization i.e. obesever 'x' shouldnt be called again unless it has finished its last work.
    – thinkmmk
    Apr 19, 2011 at 9:52

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.