我的 win 服务实例需要一个贡献函数

Question

我有一张以Guid 作为主键的表。该表已经包含很多行。另外，我有一个win service，它对每一行执行一些操作（可能需要从另一个数据库读取和写入数据）。所以处理一行需要相当多的时间。 （平均约 100 秒）

我的 win 服务是这样工作的：

public class MyDto
{
     public Guid Id { get; set; }
}

while (true){
     if(time to start){
          List<MyDto> rows = LoadData();
          foreach(MyDto obj in rows){
               Process(obj);//it takes in average about 100 sec
          }
     }
}

我需要减少所有行的执行时间。由于某些原因，我决定增加我赢得服务的次数。所以我需要每个 win 服务都处理自己的行集。

我已经参数化了我的LoadData() fun：

public List<MyDto> LoadData(int winServInstanceNumber){

}

所以我需要一个贡献函数取决于总赢服务实例数和创建赢服务实例数。

你能提供比

更好的东西吗？

//on .net side
obj.Id.GetHashCode()%totalWinServiceInstancesCount

或

--on sql side
HASHBYTES('MD5', CAST(id as varbinary(16)))%totalWinServiceInstancesCount

Answer 1

看起来您只需要旋转更多线程来处理您的数据。但要做到这一点，您需要控制您处理的内容，以免重复处理同一件事。例如，要获得控件，您可以使用MSMQ 或System.Collections.Queue。您的服务应该负责查询数据库并将未处理的行加载到您的队列中。

然后，你可以调用一些静态方法ProcessBatch。它将进入队列并旋转一个线程，并将行的 id 传递给处理器/工人。 worker 只会处理一行。 Worker 可以是单独的 EXE 并用完进程。您的“ProcessBatch”应该控制已处理/未处理的内容。它应该控制当前正在运行的线程数。你不想旋转太多。

所以

Service           ProcessControl          Worker
  |                       |                |
  |---Load Queue          |                |
  |         |             |                |
  |<--------|             |                |
  |                       |                |
  |-----Call When Q ----->|---Queue        |
  |                       |       |        |
  |                       |<------|        |
  |                       |                |
  |---Load Queue          |----Start------>|
  |        |              |<---Success-----|
  |<-------|              |                |
  |                       |---Permanent    |
  |-----Call When Q ----->|    Dequeue     |
  |                       |       |        |
  |                       |<------|        |

这可能是典型的工作负载拆分，可以加速原本缓慢的进程

Answer 2

您应该采用异步生产者/消费者模式，而不是尝试运行同一服务的多个实例。使用 Task 对象启动生产者，然后创建许多消费者。如果您的数据需要按特定顺序处理，那么您将不得不安排消费者只处理分配的数据块。否则他们可以拿起他们的工作并开始处理。

以下示例假定工作可以按任何顺序继续。您可以根据系统资源微调优化消费者数量。使用 AppSetting 配置 MaxConsumer 并找到优化处理的理想数字。

从您的服务启动/停止连接启动/停止方法，以及任何必要的日志记录/异常处理。此处的示例很简单，并显示了该模式的基础知识。

public class MyService
{
    BlockingCollection<MyDto> sharedResource = new BlockingCollection<MyDto>();
    CancellationTokenSource cancellation;
    private Task producer;
    private List<Task> consumers;
    //Load/Set this from configuration
    private static readonly int MaxConsumer = 3;

    public void Start()
    {
        this.cancellation = new CancellationTokenSource();

        // Start the producer & Consumers, as long running task
        this.producer = Task.Factory.StartNew(() => this.Produce(), TaskCreationOptions.LongRunning);
        this.consumers = new List<Task>();
        for(int i=0; i<MaxConsumer; i++)
        {
            this.consumers.Add(Task.Factory.StartNew(() => this.Consume() , TaskCreationOptions.LongRunning));
        }

       // If you need primary service loop you can do 
       // something like the following
       // while(!this.cancellation.IsCancellationRequested)
       //{
       //        this.cancellation.Token.WaitHandle.WaitOne(1000);
       //}
    }

    public void Stop()
    {
        this.cancellation.Cancel();
        WaitOnTask(producer);
        foreach(var t in this.consumers)
        {
            WaitOnTask(t);
        }
        this.cancellation.Dispose();
    }

    private void WaitOnTask(Task task)
    {
        try
        {
            if (!task.IsCompleted)
            {
                //May want to use timeout
                //instead of blindly waiting
                task.Wait();
            }
        }
        catch(ObjectDisposedException oex)           
        {
            // Task might have been disposed/closed already
        }

    }

    public void Produce()
    {
        var token = this.cancellation.Token;
        while(!token.IsCancellationRequested)
        {
            //Code for your data loading
            if (time to start)
            {
                List<MyDto> rows = LoadData();
                foreach(var data in rows)
                {
                    this.sharedResource.Add(data, token);
                }
            }

            //Wait and repeat
            token.WaitHandle.WaitOne(1000);
        }
    }

    public void Consume()
    {
        var token = this.cancellation.Token;
        try
        {
            foreach (var data in this.sharedResource.GetConsumingEnumerable(token))
            {
                // Code for your data processing
                Process(data);
            }
        }
        catch(OperationCanceledException ex)
        {
            // service stop requested, can log here
            // or take action for saving state as needed
        }
    }
}