/*
Remoting 异步队列实现,流程如下
1.并发若干客户端程序通过调用 Remoting CocurrentQueue Server 提供的公开远程方法 Enqueue 将数据元素入队尾
2.Remoting CocurrentQueue Server 发现队列不为空,则并发若干线程陆续 Dequeue 队首数据元素并处理
注意:
1.队列的数据元素定义需自行实现
2.对出列数据元素的处理程序需自行实现
不受 Remoting 生命周期租约过期的约束
*/
//AsyncQueue.cs
#define c4 //C# 4.0+
//#define c2
namespace Microshaoft
{
using System;
using System.Threading;
using System.Diagnostics;
using System.Collections.Generic;
#if c4
using System.Collections.Concurrent;
#endif
using Microshaoft;
public class AsyncQueue<T>
where T : class
{
public delegate void QueueEventHandler(T element);
public event QueueEventHandler OnDequeue;
public delegate void QueueLogEventHandler(string logMessage);
//public event QueueLogEventHandler OnQueueLog;
public event QueueLogEventHandler OnQueueRunningThreadStart;
public event QueueLogEventHandler OnQueueRunningThreadEnd;
public event QueueLogEventHandler OnDequeueThreadStart;
public event QueueLogEventHandler OnDequeueThreadEnd;
public event QueueLogEventHandler OnDequeueAllThreadsEnd;
public delegate void ExceptionEventHandler(Exception exception);
public event ExceptionEventHandler OnException;
#if c2
private Queue<T> _queue = new Queue<T>();
#elif c4
private ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
#endif
private object _syncQueueLockObject = new object();
//private object _syncQueueRunningLockObject = new object();
private long _isQueueRunning = 0;
private long _concurrentDequeueThreadsCount = 0; //Microshaoft 用于控制并发线程数
private PerformanceCounter _enqueuePerformanceCounter;
private PerformanceCounter _dequeuePerformanceCounter;
private PerformanceCounter _dequeueProcessedPerformanceCounter;
private PerformanceCounter _queueLengthPerformanceCounter;
private PerformanceCounter _dequeueThreadStartPerformanceCounter;
private PerformanceCounter _dequeueThreadEndPerformanceCounter;
private PerformanceCounter _dequeueThreadsCountPerformanceCounter;
private PerformanceCounter _queueRunningThreadStartPerformanceCounter;
private PerformanceCounter _queueRunningThreadEndPerformanceCounter;
private PerformanceCounter _queueRunningThreadsCountPerformanceCounter;
private bool _isAttachedPerformanceCounters = false;
public void AttachPerformanceCounters(string instanceNamePrefix)
{
string category = "Microshaoft AsyncConurrentQueue Counters";
string counter = string.Empty;
Process process = Process.GetCurrentProcess();
//int processID = 0;//process.Id;
string processName = process.ProcessName;
//string processStartTime = "";//process.StartTime;
string instanceName = string.Empty;
instanceName = string.Format
(
"{0}-{1}"
, instanceNamePrefix
, processName
//, processID
//, processStartTime.ToString("yyyy-MM-dd HH:mm:ss.fff")
);
CounterCreationDataCollection ccdc = new CounterCreationDataCollection();
if (PerformanceCounterCategory.Exists(category))
{
PerformanceCounterCategory.Delete(category);
}
CounterCreationData ccd = null;
counter = "EnqueueCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "DequeueCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "QueueLengthCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "DequeueProcessedCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "DequeueThreadStartCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "DequeueThreadEndCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "DequeueThreadsCountCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "QueueRunningThreadStartCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "QueueRunningThreadEndCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
counter = "QueueRunningThreadsCountCounter";
ccd = PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64);
ccdc.Add(PerformanceCounterHelper.GetCounterCreationData(counter, PerformanceCounterType.NumberOfItems64));
PerformanceCounterCategory.Create
(
category,
string.Format("{0} Category Help.", category),
PerformanceCounterCategoryType.MultiInstance,
ccdc
);
counter = "EnqueueCounter";
_enqueuePerformanceCounter = new PerformanceCounter();
_enqueuePerformanceCounter.CategoryName = category;
_enqueuePerformanceCounter.CounterName = counter;
_enqueuePerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_enqueuePerformanceCounter.InstanceName = instanceName;
_enqueuePerformanceCounter.ReadOnly = false;
_enqueuePerformanceCounter.RawValue = 0;
counter = "DequeueCounter";
_dequeuePerformanceCounter = new PerformanceCounter();
_dequeuePerformanceCounter.CategoryName = category;
_dequeuePerformanceCounter.CounterName = counter;
_dequeuePerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_dequeuePerformanceCounter.InstanceName = instanceName;
_dequeuePerformanceCounter.ReadOnly = false;
_dequeuePerformanceCounter.RawValue = 0;
counter = "DequeueProcessedCounter";
_dequeueProcessedPerformanceCounter = new PerformanceCounter();
_dequeueProcessedPerformanceCounter.CategoryName = category;
_dequeueProcessedPerformanceCounter.CounterName = counter;
_dequeueProcessedPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_dequeueProcessedPerformanceCounter.InstanceName = instanceName;
_dequeueProcessedPerformanceCounter.ReadOnly = false;
_dequeueProcessedPerformanceCounter.RawValue = 0;
counter = "QueueLengthCounter";
_queueLengthPerformanceCounter = new PerformanceCounter();
_queueLengthPerformanceCounter.CategoryName = category;
_queueLengthPerformanceCounter.CounterName = counter;
_queueLengthPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_queueLengthPerformanceCounter.InstanceName = instanceName;
_queueLengthPerformanceCounter.ReadOnly = false;
_queueLengthPerformanceCounter.RawValue = 0;
counter = "DequeueThreadStartCounter";
_dequeueThreadStartPerformanceCounter = new PerformanceCounter();
_dequeueThreadStartPerformanceCounter.CategoryName = category;
_dequeueThreadStartPerformanceCounter.CounterName = counter;
_dequeueThreadStartPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_dequeueThreadStartPerformanceCounter.InstanceName = instanceName;
_dequeueThreadStartPerformanceCounter.ReadOnly = false;
_dequeueThreadStartPerformanceCounter.RawValue = 0;
counter = "DequeueThreadEndCounter";
_dequeueThreadEndPerformanceCounter = new PerformanceCounter();
_dequeueThreadEndPerformanceCounter.CategoryName = category;
_dequeueThreadEndPerformanceCounter.CounterName = counter;
_dequeueThreadEndPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_dequeueThreadEndPerformanceCounter.InstanceName = instanceName;
_dequeueThreadEndPerformanceCounter.ReadOnly = false;
_dequeueThreadEndPerformanceCounter.RawValue = 0;
counter = "DequeueThreadsCountCounter";
_dequeueThreadsCountPerformanceCounter = new PerformanceCounter();
_dequeueThreadsCountPerformanceCounter.CategoryName = category;
_dequeueThreadsCountPerformanceCounter.CounterName = counter;
_dequeueThreadsCountPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_dequeueThreadsCountPerformanceCounter.InstanceName = instanceName;
_dequeueThreadsCountPerformanceCounter.ReadOnly = false;
_dequeueThreadsCountPerformanceCounter.RawValue = 0;
counter = "QueueRunningThreadStartCounter";
_queueRunningThreadStartPerformanceCounter = new PerformanceCounter();
_queueRunningThreadStartPerformanceCounter.CategoryName = category;
_queueRunningThreadStartPerformanceCounter.CounterName = counter;
_queueRunningThreadStartPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_queueRunningThreadStartPerformanceCounter.InstanceName = instanceName;
_queueRunningThreadStartPerformanceCounter.ReadOnly = false;
_queueRunningThreadStartPerformanceCounter.RawValue = 0;
counter = "QueueRunningThreadEndCounter";
_queueRunningThreadEndPerformanceCounter = new PerformanceCounter();
_queueRunningThreadEndPerformanceCounter.CategoryName = category;
_queueRunningThreadEndPerformanceCounter.CounterName = counter;
_queueRunningThreadEndPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_queueRunningThreadEndPerformanceCounter.InstanceName = instanceName;
_queueRunningThreadEndPerformanceCounter.ReadOnly = false;
_queueRunningThreadEndPerformanceCounter.RawValue = 0;
counter = "QueueRunningThreadsCountCounter";
_queueRunningThreadsCountPerformanceCounter = new PerformanceCounter();
_queueRunningThreadsCountPerformanceCounter.CategoryName = category;
_queueRunningThreadsCountPerformanceCounter.CounterName = counter;
_queueRunningThreadsCountPerformanceCounter.InstanceLifetime = PerformanceCounterInstanceLifetime.Process;
_queueRunningThreadsCountPerformanceCounter.InstanceName = instanceName;
_queueRunningThreadsCountPerformanceCounter.ReadOnly = false;
_queueRunningThreadsCountPerformanceCounter.RawValue = 0;
_isAttachedPerformanceCounters = true;
}
private int _maxConcurrentThreadsCount = 1; //Microshaoft 允许并发出列处理线程数为 1
public int MaxConcurrentThreadsCount
{
set
{
_maxConcurrentThreadsCount = value;
}
get
{
return _maxConcurrentThreadsCount;
}
}
//Microshaoft 服务启动后可立即开启新的线程调用此方法(死循环)
private void QueueRun() //Microshaoft ThreadStart
{
if (Interlocked.Read(ref _concurrentDequeueThreadsCount) < _maxConcurrentThreadsCount)
{
if (Interlocked.CompareExchange(ref _isQueueRunning, 0, 1) == 0)
{
ThreadStart ts = new ThreadStart(QueueRunThreadProcess);
Thread t = new Thread(ts);
t.Name = "QueueRunningThreadProcess";
t.Start();
}
}
}
public int Count
{
get
{
return _queue.Count;
}
}
public long ConcurrentThreadsCount
{
get
{
return _concurrentDequeueThreadsCount;
}
}
private void QueueRunThreadProcess()
{
if (_isAttachedPerformanceCounters)
{
_queueRunningThreadStartPerformanceCounter.Increment();
_queueRunningThreadsCountPerformanceCounter.Increment();
}
if (OnQueueRunningThreadStart != null)
{
OnQueueRunningThreadStart
(
string.Format
(
"{0} Threads Count {1},Queue Count {2},Current Thread: {3}({4}) at {5}"
, "Queue Running Start ..."
, _concurrentDequeueThreadsCount
, _queue.Count
, Thread.CurrentThread.Name
, Thread.CurrentThread.ManagedThreadId
, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fffff")
)
);
}
#if c2
while ((_queue.Count > 0)) //Microshaoft 死循环
#elif c4
while (!_queue.IsEmpty) //Microshaoft 死循环
#endif
{
int threadID = -1;
{
int r = (int) Interlocked.Read(ref _concurrentDequeueThreadsCount);
if (r < _maxConcurrentThreadsCount)
{
//if (_queue.Count > 0)
{
r = (int) Interlocked.Increment(ref _concurrentDequeueThreadsCount);
threadID = (int) _concurrentDequeueThreadsCount;
//ThreadProcessState tps = new ThreadProcessState();
//tps.element = element;
//tps.Sender = this;
Thread t = new Thread(new ThreadStart(DequeueThreadProcess));
t.Name = string.Format("ConcurrentDequeueProcessThread[{0}]", threadID);
t.Start();
}
/// else
/// {
/// break;
/// }
}
else
{
break;
}
}
}
//Interlocked.CompareExchange(ref _queueRuning, 0, 1);
if (OnQueueRunningThreadEnd != null)
{
int r = (int) Interlocked.Read(ref _concurrentDequeueThreadsCount);
OnQueueRunningThreadEnd
(
string.Format
(
"{0} Threads Count {1}, Queue Count {2}, Current Thread: {3}({4}) at {5}"
, "Queue Running Stop ..."
, r
, _queue.Count
, Thread.CurrentThread.Name
, Thread.CurrentThread.ManagedThreadId
, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fffff")
)
);
}
if (_isAttachedPerformanceCounters)
{
_queueRunningThreadEndPerformanceCounter.Increment();
_queueRunningThreadsCountPerformanceCounter.Decrement();
}
Interlocked.Exchange(ref _isQueueRunning, 0);
}
public void Enqueue(T element)
{
try
{
#if c2
lock (_syncQueueLockObject) //还算并发吗?
#endif
{
_queue.Enqueue(element);
}
if (_isAttachedPerformanceCounters)
{
_enqueuePerformanceCounter.Increment();
_queueLengthPerformanceCounter.Increment();
}
}
catch (Exception e)
{
if (OnException != null)
{
OnException(e);
}
}
//int r = Interlocked.CompareExchange(ref _queueRuning, 1, 0))
//if (r == 1)
//{
QueueRun();
//}
}
private void DequeueThreadProcess()
{
if (_isAttachedPerformanceCounters)
{
_dequeueThreadStartPerformanceCounter.Increment();
_dequeueThreadsCountPerformanceCounter.Increment();
}
if (OnDequeueThreadStart != null)
{
int r = (int) Interlocked.Read(ref _concurrentDequeueThreadsCount);
OnDequeueThreadStart
(
string.Format
(
"{0} Threads Count {1},Queue Count {2},Current Thread: {3} at {4}"
, "Threads ++ !"
, r
, _queue.Count
, Thread.CurrentThread.Name
, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fffff")
)
);
}
bool queueWasNotEmpty = false;
try
{
#if c2
while (true)
#elif c4
while (!_queue.IsEmpty)
#endif
{
T element = null;
#if c2
lock (_syncQueueLockObject)
{
if (_queue.Count > 0)
{
element = _queue.Dequeue();
}
else
{
//避免QueueRun 死循环
break;
}
}
#elif c4
if (_queue.TryDequeue(out element))
{
#elif c2
if (element != null)
{
#endif
if (!queueWasNotEmpty)
{
queueWasNotEmpty = true;
}
if (_isAttachedPerformanceCounters)
{
_dequeuePerformanceCounter.Increment();
_queueLengthPerformanceCounter.Decrement();
}
if (OnDequeue != null)
{
OnDequeue(element);
}
if (_isAttachedPerformanceCounters)
{
_dequeueProcessedPerformanceCounter.Increment();
}
#if c2
}
#elif c4
}
}
#endif
}
catch (Exception e)
{
if (OnException != null)
{
OnException(e);
}
}
finally
{
int r = (int) Interlocked.Decrement(ref _concurrentDequeueThreadsCount);
if (OnDequeueThreadEnd != null)
{
OnDequeueThreadEnd
(
string.Format
(
"{0} Threads Count {1},Queue Count {2},Current Thread: {3} at {4}"
, "Threads--"
, r
, _queue.Count
, Thread.CurrentThread.Name
, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fffff")
)
);
}
if (r == 0)
{
if (OnDequeueAllThreadsEnd != null)
{
OnDequeueAllThreadsEnd
(
string.Format
(
"{0} Threads Count {1},Queue Count {2},Current Thread: {3} at {4}"
, "All Threads End"
, r
, _queue.Count
, Thread.CurrentThread.Name
, DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fffff")
)
);
}
}
if (_isAttachedPerformanceCounters)
{
_dequeueThreadEndPerformanceCounter.Increment();
_dequeueThreadsCountPerformanceCounter.Decrement();
}
if (queueWasNotEmpty)
{
QueueRun(); //死循环???
}
}
}
}
}
namespace Microshaoft
{
using System;
using System.Diagnostics;
public static class PerformanceCounterHelper
{
public static CounterCreationData GetCounterCreationData(string counterName, PerformanceCounterType performanceCounterType)
{
CounterCreationData ccd = new CounterCreationData();
ccd.CounterName = counterName;
ccd.CounterHelp = string.Format("{0} Help", counterName);
ccd.CounterType = performanceCounterType;
return ccd;
}
}
}
//Server.cs
namespace Microshaoft
{
using System;
using System.Threading;
using System.Data;
using System.Data.SqlClient;
using Microshaoft.RemotingObjects.Share;
public class AsyncQueueProcessor
{
private AsyncQueue<Item> _queue;
public AsyncQueue<Item> Queue
{
get
{
return _queue;
}
}
public AsyncQueueProcessor(AsyncQueue<Item> queue)
{
_queue = queue;
_queue.OnDequeue += new AsyncQueue<Item>.QueueEventHandler(_queue_OnDequeue);
//_queue.OnDequeueThreadStart += new AsyncQueue<Item>.QueueLogEventHandler(_queue_OnQueueLog);
_queue.OnDequeueAllThreadsEnd += new AsyncQueue<Item>.QueueLogEventHandler(_queue_OnQueueLog);
_queue.OnDequeueThreadEnd += new AsyncQueue<Item>.QueueLogEventHandler(_queue_OnQueueLog);
//_queue.OnQueueRunningThreadStart += new AsyncQueue<Item>.QueueLogEventHandler(_queue_OnQueueLog);
//_queue.OnQueueRunningThreadEnd += new AsyncQueue<Item>.QueueLogEventHandler(_queue_OnQueueLog);
_queue.OnException += new AsyncQueue<Item>.ExceptionEventHandler(_queue_OnException);
_queue.MaxConcurrentThreadsCount = 64;
}
public void Enqueue(Item item)
{
_queue.Enqueue(item);
}
void _queue_OnQueueLog(string logMessage)
{
Console.WriteLine(logMessage);
}
void _queue_OnDequeue(Item item)
{
DateTime DequeueBeginTime = DateTime.Now;
/// SqlConnection connection = null;
/// try
/// {
/// connection = new SqlConnection(item.ConnectionString);
/// SqlCommand command = new SqlCommand(item.SqlCommandText, connection);
/// command.CommandType = CommandType.Text;
/// connection.Open();
/// command.ExecuteNonQuery();
/// }
/// catch (Exception e)
/// {
/// Console.WriteLine("Exception on Dequeue Process:{0}{1}", "\r\n", e.ToString());
/// }
/// finally
/// {
/// connection.Close();
/// connection.Dispose();
/// connection = null;
/// }
DateTime DequeueEndTime = DateTime.Now;
Console.WriteLine
(
"QueueRemainCount {0}, Enqueue {1}, Dequeue {2},[{3}], End {4},[{5}],[{6}]"
, _queue.Count
, item.EnqueueTime
, DequeueBeginTime
, (DequeueBeginTime.Ticks - item.EnqueueTime.Ticks) / 10000 / 1000
, DequeueEndTime
, (DequeueEndTime.Ticks - DequeueBeginTime.Ticks) / 10000 / 1000
, _queue.ConcurrentThreadsCount
);
}
void _queue_OnException(Exception e)
{
Console.WriteLine(e.ToString());
}
}
}
namespace Microshaoft.RemotingObjects.Server
{
using System;
using System.Threading;
using System.Collections;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Channels;
using System.Runtime.Remoting.Channels.Tcp;
using System.Runtime.Serialization.Formatters;
using System.ServiceProcess;
using System.ComponentModel;
using System.Configuration.Install;
using System.Security.Principal;
using Microshaoft.RemotingObjects;
using Microshaoft.RemotingObjects.Share;
using Microshaoft.Win32;
using Microshaoft;
//using eHome.UserProfile;
public class ServiceHost : ServiceBase
{
///// <summary>
/// 应用程序的主入口点。
/// </summary>
//[STAThread]
public static readonly string serviceName = "RemotingAsyncConcurrentQueueService";
private static AsyncQueueProcessor _asyncQueueProcessor;
public static AsyncQueueProcessor AsyncQueueProcessor
{
get
{
return _asyncQueueProcessor;
}
}
static void Main(string[] args)
{
//Microshaoft
//Microshaoft TODO: 在此处添加代码以启动应用程序
//Microshaoft
ServiceHost service = new ServiceHost();
int l = 0;
bool needFreeConsole = false;
if (args != null)
{
l = args.Length;
}
if (l > 0)
{
if (args[0].ToLower() == "/console")
{
needFreeConsole = true;
NativeMethods.AllocConsole();
Console.Title = "Server ...";
Console.WriteLine("Alloc Console ...");
Console.WriteLine("Current User Identity: {0}", WindowsIdentity.GetCurrent().Name);
Console.WriteLine(".Net Framework version: {0}", Environment.Version.ToString());
Console.Title = "Server"; //不能以服务运行
Console.WriteLine("Console");
service.OnStart(null);
Console.ReadLine();
return;
}
}
Console.WriteLine("Service");
ServiceBase.Run(service);
if (needFreeConsole)
{
Console.WriteLine("Free Console ...");
NativeMethods.FreeConsole();
}
}
public ServiceHost()
{
CanPauseAndContinue = true;
ServiceName = ServiceHost.serviceName;
}
protected override void OnStart(string[] args)
{
Console.WriteLine(Environment.Version.ToString());
AsyncQueue<Item> queue = new AsyncQueue<Item>();
queue.AttachPerformanceCounters("Q1");
_asyncQueueProcessor = new AsyncQueueProcessor(queue);
RemotingHelper.StartRemoting<RemotingAsyncQueue>
(
"queueurl"
, 8080
);
Console.WriteLine("Server . , Press Enter key to exit.");
}
}
[RunInstallerAttribute(true)]
public class ProjectInstaller : Installer
{
private ServiceInstaller serviceInstaller;
private ServiceProcessInstaller processInstaller;
public ProjectInstaller()
{
processInstaller = new ServiceProcessInstaller();
serviceInstaller = new ServiceInstaller();
// Service will run under system account
processInstaller.Account = ServiceAccount.LocalSystem;
// Service will have Start Type of Manual
serviceInstaller.StartType = ServiceStartMode.Manual;
serviceInstaller.ServiceName = ServiceHost.serviceName;
Installers.Add(serviceInstaller);
Installers.Add(processInstaller);
}
}
}
namespace Microshaoft.Win32
{
using System.Runtime.InteropServices;
public class NativeMethods
{
/// <summary>
/// 启动控制台
/// </summary>
/// <returns></returns>
[DllImport("kernel32.dll")]
public static extern bool AllocConsole();
/// <summary>
/// 释放控制台
/// </summary>
/// <returns></returns>
[DllImport("kernel32.dll")]
public static extern bool FreeConsole();
}
}
namespace Microshaoft.RemotingObjects
{
using System;
using System.IO;
using System.Net;
using System.Web;
using System.Text;
using System.Threading;
using System.Configuration;
using System.Collections.Generic;
using Microshaoft;
using Microshaoft.RemotingObjects.Share;
using Microshaoft.RemotingObjects.Server;
public class RemotingAsyncQueue : MarshalByRefObject
{
public void Enqueue(Item item)
{
//Microshaoft 队列的数据元素定义需自行实现 Item
ServiceHost.AsyncQueueProcessor.Enqueue(item);
}
}
}
//=============================================================
//===============================================
// Share.cs
//Server、Client 均需引用此 share.dll
//C:\WINDOWS\Microsoft.NET\Framework\v2.0.50727\csc.exe /t:library share.cs
//TO DO
//队列的数据元素定义需自行实现,示例如下:
namespace Microshaoft.RemotingObjects.Share
{
using System;
[Serializable]
public class Item
{
private DateTime _EnqueueTime;
public DateTime EnqueueTime
{
get
{
return _EnqueueTime;
}
set
{
_EnqueueTime = value;
}
}
private string _sql;
public string SqlCommandText
{
get
{
return _sql;
}
set
{
_sql = value;
}
}
private string _connectionString;
public string ConnectionString
{
get
{
return _connectionString;
}
set
{
_connectionString = value;
}
}
}
}
// remoting helper
//Share.cs
namespace Microshaoft
{
using System;
using System.Collections;
using System.Collections.Generic;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Channels;
using System.Runtime.Remoting.Channels.Tcp;
using System.Runtime.Serialization.Formatters;
using System.Text;
public static class RemotingHelper
{
public static void StartRemoting
(
Type RemotingType
, string Url
, int Port
)
{
BinaryServerFormatterSinkProvider provider = new BinaryServerFormatterSinkProvider();
provider.TypeFilterLevel = TypeFilterLevel.Full;
IDictionary ht = new Hashtable();
ht["port"] = Port;
TcpChannel tc = new TcpChannel(ht, null, provider);
ChannelServices.RegisterChannel(tc, false);
RemotingConfiguration.RegisterWellKnownServiceType(RemotingType, Url, WellKnownObjectMode.Singleton);
Console.WriteLine("Remoting Object Started ...");
}
public static void StartRemoting<T>
(
string Url
, int Port
)
{
StartRemoting(typeof(T), Url, Port);
}
public static T GetRemotingLocalClientProxyObject<T>
(
string Url
)
{
return (T) Activator.GetObject
(
typeof(T)
, Url
//, "tcp://127.0.0.1:8080/queueUrl"
);
}
}
}
//============================================================================
//============================================
// Client.cs
//C:\WINDOWS\Microsoft.NET\Framework\v1.1.4322\csc.exe client.cs /r:share.dll
namespace Microshaoft.RemotingObjects.Client
{
using System;
using System.Collections;
using System.Runtime.Remoting;
using System.Runtime.Remoting.Channels;
using System.Runtime.Remoting.Channels.Tcp;
using System.Runtime.Serialization.Formatters;
using System.Threading;
using System.Data;
using System.Data.SqlClient;
using Microshaoft.RemotingObjects;
using Microshaoft.RemotingObjects.Share;
public class Class1
{
static RemotingAsyncQueue _queue;
public static void Main()
{
Console.Title = "Client";
Console.WriteLine(Environment.Version.ToString());
Class1 a = new Class1();
a.Run();
}
public void Run()
{
_queue = RemotingHelper.GetRemotingLocalClientProxyObject<RemotingAsyncQueue>("tcp://127.0.0.1:8080/queueUrl");
//Microshaoft 以下是耗时的主程序
for (int i = 0; i < 20; i++)
{
Thread x = new Thread(new ThreadStart(ThreadProcess));
x.Start();
}
}
public void ThreadProcess()
{
for (int i = 0; i < 1000; i++)
{
Item x = new Item();
DateTime EnqueueTime = DateTime.Now;
x.EnqueueTime = EnqueueTime;
x.SqlCommandText = @"
--==========================
declare @ varchar(10)
set @ = 'aaa'
exec zsp_test @
--==========================
";
x.ConnectionString = "";
_queue.Enqueue(x);
Console.WriteLine
(
"Enqueue: {0},[{1}]"
, EnqueueTime
, (DateTime.Now.Ticks - EnqueueTime.Ticks)/10000
);
}
}
}
}
//Microshaoft =========================================
//Microshaoft Remoting Object Client Local Proxy
namespace Microshaoft.RemotingObjects
{
using System;
using Microshaoft.RemotingObjects.Share;
public interface RemotingAsyncQueue
{
void Enqueue(Item item);
}
}