I need to provide a service (either a Windows Service or an Azure Worker Role) which will handle the parallel execution of jobs. These jobs could be anything from importing data to compiling of reports, or to sending out mass notifications, etc.
Database Design:
The whole process of running these jobs is persistent.
JobDefinition
stores varchar reference to theIJob
concrete type class.JobInstance
is an instance ofJobDefinition
which needs to be executed as a job.JobInstanceEvent
stores the process of executing an instance (the change of states).
Class Structure:
In order to create new jobs the simple IJob
interface needs to be implemented:
public interface IJob
{
bool Execute();
}
JobHandler
is responsible for deserializing the serialized varchar in the database and also saving state changes:
public class JobHandler
{
public string State
{
get
{
return jobInstance.State;
}
set
{
using (MyEntities context = new MyEntities())
{
context.JobInstances.Attach(jobInstance);
jobInstance.State = value;
context.SaveChanges();
}
SaveJobEvent(String.Empty);
}
}
private JobInstance jobInstance;
public JobHandler(JobInstance job)
{
jobInstance = job;
}
public IJob CreateJobObject()
{
Type type = Type.GetType(jobInstance.JobDefinition.FullAssemblyType);
XmlSerializer serializer = new XmlSerializer(type);
using (StringReader reader = new StringReader(jobInstance.SerialisedObject))
{
return serializer.Deserialize(reader) as IJob;
}
}
public void SaveJobEvent(string jobEventMessage)
{
using (MyEntities context = new MyEntities())
{
context.JobInstances.Attach(jobInstance);
jobInstance.JobInstanceEvents.Add(
new JobInstanceEvent()
{
State = jobInstance.State,
Date = DateTime.UtcNow,
Message = jobEventMessage
}
);
context.SaveChanges();
}
}
}
The JobConsumer
is what will run in the service. It performs two primary parallel tasks:
- Adding items to the blocking queue (through polling)
- Executing items from the blocking queue (through thread blocking)
It is able to run multiple jobs concurrently (ParallelExecutionSlots
).
public class JobConsumer
{
public IJobSource JobSource { get; private set; }
public BlockingCollection<JobHandler> JobQueue { get; private set; }
public int ParallelExecutionSlots { get; private set; }
private bool isPopulating;
private CancellationTokenSource consumerCancellationTokenSource;
public JobConsumer(IJobSource jobSource, int parallelExecutionSlots)
{
JobQueue = new BlockingCollection<JobHandler>();
JobSource = jobSource;
ParallelExecutionSlots = parallelExecutionSlots;
consumerCancellationTokenSource = new CancellationTokenSource();
StartPopulatingQueue();
}
public void StartPopulatingQueue()
{
StartPopulatingQueue(500);
}
public void StartPopulatingQueue(int pollingDelay)
{
if (!isPopulating)
{
isPopulating = true;
var task = Task.Factory.StartNew(() =>
{
while (isPopulating)
{
foreach (var job in JobSource.ReadNewJobs())
{
job.State = "Queued";
JobQueue.Add(job);
}
Tasks.Task.Delay(pollingDelay).Wait();
}
})
.ContinueWith(t =>
{
isPopulating = false;
JobConsumerLog.LogException("JobConsumer populating process failed", t.Exception);
},
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted,
TaskScheduler.Current
);
}
}
public void StopPopulatingQueue()
{
isPopulating = false;
}
public void StartConsumer()
{
Task.Factory.StartNew(() =>
{
SemaphoreSlim slotsSemaphore = new SemaphoreSlim(ParallelExecutionSlots);
foreach (var job in JobQueue.GetConsumingEnumerable(consumerCancellationTokenSource.Token))
{
slotsSemaphore.Wait();
JobHandler jobHandler = job;
Task.Factory.StartNew(() =>
{
try
{
ExecuteJob(jobHandler);
}
finally
{
slotsSemaphore.Release();
}
},
TaskCreationOptions.LongRunning
);
}
consumerCancellationTokenSource.Token.ThrowIfCancellationRequested();
}, consumerCancellationTokenSource.Token)
.ContinueWith(t =>
{
JobConsumerLog.LogException("JobConsumer execution process failed", t.Exception);
},
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted,
TaskScheduler.Current
)
.ContinueWith(t =>
{
JobConsumerLog.LogException("JobConsumer execution process stopped", t.Exception);
},
CancellationToken.None,
TaskContinuationOptions.OnlyOnCanceled,
TaskScheduler.Current
);
}
public void StopConsumer()
{
consumerCancellationTokenSource.Cancel();
}
private void ExecuteJob(JobHandler job)
{
job.State = "Executing";
try
{
if (job.CreateJobObject().Execute())
{
job.State = "Successful";
}
else
{
job.State = "Failed";
}
}
catch (Exception ex)
{
job.State = "FailedOnException";
}
}
}
Feedback on the overall design would be much appreciated. I'm looking for a maintainable and extensible solution.
I am using .NET 4.5
(Note: this is a second iteration of the following post)
jobInstance
. Declaring them at the top of the class also helps, especially if you're not prefixing them and you're using them before they're declared. Call that a "readability review" :) IsJobConsumerLog
a static class? I'd just replace that with a constructor-injectedILogger
, otherwise you've got ambient context here, not obvious the thing is logging anything. \$\endgroup\$ – Mathieu Guindon♦ Oct 1 '13 at 23:45