1 namespace MikeWare.Core.Components.CrawlerFramework.Policies
 2 {
 3     using System;
 4 
 5     public abstract class AConcurrentPolicy
 6     {
 7         public virtual bool WaitOne(TimeSpan timeout) => throw new NotImplementedException();
 8 
 9         public virtual void ReleaseOne() => throw new NotImplementedException();
10     }
11 }

 1 namespace MikeWare.Core.Components.CrawlerFramework.Policies
 2 {
 3     using System;
 4     using System.Threading;
 5 
 6     public class SemaphoreConcurrentPolicy : AConcurrentPolicy
 7     {
 8         private Semaphore semaphore = null;
 9 
10         public SemaphoreConcurrentPolicy(int init, int max)
11         {
12             semaphore = new Semaphore(init, max);
13         }
14 
15         public override bool WaitOne(TimeSpan timeout)
16         {
17             return semaphore.WaitOne(timeout);
18         }
19 
20         public override void ReleaseOne()
21         {
22             semaphore.Release(1);
23         }
24     }
25 }

namespace System.Threading
{
    //
    // Summary:
    //     Limits the number of threads that can access a resource or pool of resources
    //     concurrently.
    public sealed class Semaphore : WaitHandle
    {
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be granted concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be granted concurrently.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        public Semaphore(int initialCount, int maximumCount);
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries, and
        //     optionally specifying the name of a system semaphore object.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be granted concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be granted concurrently.
        //
        //   name:
        //     The name of a named system semaphore object.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists and has access control security, and the user does
        //     not have System.Security.AccessControl.SemaphoreRights.FullControl.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore cannot be created, perhaps because a wait handle of a different
        //     type has the same name.
        public Semaphore(int initialCount, int maximumCount, string name);
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries, optionally
        //     specifying the name of a system semaphore object, and specifying a variable that
        //     receives a value indicating whether a new system semaphore was created.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be satisfied concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be satisfied concurrently.
        //
        //   name:
        //     The name of a named system semaphore object.
        //
        //   createdNew:
        //     When this method returns, contains true if a local semaphore was created (that
        //     is, if name is null or an empty string) or if the specified named system semaphore
        //     was created; false if the specified named system semaphore already existed. This
        //     parameter is passed uninitialized.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists and has access control security, and the user does
        //     not have System.Security.AccessControl.SemaphoreRights.FullControl.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore cannot be created, perhaps because a wait handle of a different
        //     type has the same name.
        public Semaphore(int initialCount, int maximumCount, string name, out bool createdNew);

        //
        // Summary:
        //     Opens the specified named semaphore, if it already exists.
        //
        // Parameters:
        //   name:
        //     The name of the system semaphore to open.
        //
        // Returns:
        //     An object that represents the named system semaphore.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     name is an empty string. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentNullException:
        //     name is null.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore does not exist.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists, but the user does not have the security access required
        //     to use it.
        public static Semaphore OpenExisting(string name);
        //
        // Summary:
        //     Opens the specified named semaphore, if it already exists, and returns a value
        //     that indicates whether the operation succeeded.
        //
        // Parameters:
        //   name:
        //     The name of the system semaphore to open.
        //
        //   result:
        //     When this method returns, contains a System.Threading.Semaphore object that represents
        //     the named semaphore if the call succeeded, or null if the call failed. This parameter
        //     is treated as uninitialized.
        //
        // Returns:
        //     true if the named semaphore was opened successfully; otherwise, false.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     name is an empty string. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentNullException:
        //     name is null.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists, but the user does not have the security access required
        //     to use it.
        public static bool TryOpenExisting(string name, out Semaphore result);
        //
        // Summary:
        //     Exits the semaphore and returns the previous count.
        //
        // Returns:
        //     The count on the semaphore before the System.Threading.Semaphore.Release* method
        //     was called.
        //
        // Exceptions:
        //   T:System.Threading.SemaphoreFullException:
        //     The semaphore count is already at the maximum value.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred with a named semaphore.
        //
        //   T:System.UnauthorizedAccessException:
        //     The current semaphore represents a named system semaphore, but the user does
        //     not have System.Security.AccessControl.SemaphoreRights.Modify. -or- The current
        //     semaphore represents a named system semaphore, but it was not opened with System.Security.AccessControl.SemaphoreRights.Modify.
        public int Release();
        //
        // Summary:
        //     Exits the semaphore a specified number of times and returns the previous count.
        //
        // Parameters:
        //   releaseCount:
        //     The number of times to exit the semaphore.
        //
        // Returns:
        //     The count on the semaphore before the System.Threading.Semaphore.Release* method
        //     was called.
        //
        // Exceptions:
        //   T:System.ArgumentOutOfRangeException:
        //     releaseCount is less than 1.
        //
        //   T:System.Threading.SemaphoreFullException:
        //     The semaphore count is already at the maximum value.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred with a named semaphore.
        //
        //   T:System.UnauthorizedAccessException:
        //     The current semaphore represents a named system semaphore, but the user does
        //     not have System.Security.AccessControl.SemaphoreRights.Modify rights. -or- The
        //     current semaphore represents a named system semaphore, but it was not opened
        //     with System.Security.AccessControl.SemaphoreRights.Modify rights.
        public int Release(int releaseCount);
    }
}

 1 namespace MikeWare.Core.Components.CrawlerFramework
 2 {
 3     using MikeWare.Core.Components.CrawlerFramework.Policies;
 4     using System;
 5     using System.Collections.Concurrent;
 6     using System.Threading;
 7     using System.Threading.Tasks;
 8 
 9     public class LeaderAnt : Ant
10     {
11         private ConcurrentQueue<JobContext> Queue;
12         private ManualResetEvent mre = new ManualResetEvent(false);
13         public AConcurrentPolicy EnqueuePolicy { get; set; }
14 
15         ……
16 
17         public void Enqueue(JobContext context)
18         {
19             if (null != EnqueuePolicy)
20             {
21                 while (!EnqueuePolicy.WaitOne(TimeSpan.FromMilliseconds(3)) && !mre.WaitOne(1))
22                     continue;
23             }
24 
25             Queue.Enqueue(context);
26         }
27 
28         ……
29 }

 1 namespace MikeWare.Crawlers.EBooks.Bizs
 2 {
 3     using MikeWare.Core.Components.CrawlerFramework;
 4     using MikeWare.Core.Components.CrawlerFramework.Policies;
 5     using MikeWare.Crawlers.EBooks.Entities;
 6     using System;
 7     using System.Collections.Generic;
 8     using System.Net;
 9 
10     public class EBooksCrawler
11     {
12         public static void Start(int pageIndex, DateTime lastUpdateTime)
13         {
14             var leader = new LeaderAnt()
15             {
16                 EnqueuePolicy = new SemaphoreConcurrentPolicy(100, 100)
17                 //EnqueuePolicy = new PeriodEnqueuePolicy(TimeSpan.FromMilliseconds(150))
18             };
19 
20             var newContext = new JobContext
21             {
22                 JobName = $"奇书网-最新电子书-列表-第{pageIndex.ToString("00000")}页",
23                 Uri = $"http://www.xqishuta.com/s/new/index_{pageIndex}.html",
24                 Method = WebRequestMethods.Http.Get,
25                 InParams = new Dictionary<string, object>(),
26                 Analizer = new BooksListAnalizer(),
27             };
28             newContext.InParams.Add(Consts.PAGE_INDEX, 1);
29             newContext.InParams.Add(Consts.LAST_UPDATE_TIME, DateTime.MinValue);
30 
31             leader.Enqueue(newContext);
32 
33             leader.Work();
34         }
35     }
36 }