修复一个请求 timeout 导致后续全部 timeout 的问题

Author: zlzforever

package.props

@@ -1,11 +1,13 @@
 <Project>
 
   <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <LangVersion>latest</LangVersion>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
     <PackageRequireLicenseAcceptance>true</PackageRequireLicenseAcceptance>
     <PackageLicenseFile>LICENSE.txt</PackageLicenseFile>
-    <Version>5.0.1-beta6</Version>
-    <FileVersion>5.0.1.6</FileVersion>
+    <Version>5.0.1-beta7</Version>
+    <FileVersion>5.0.1.7</FileVersion>
     <Authors>[email protected];</Authors>
     <Copyright>Copyright 2018 Lewis Zou</Copyright>
     <Description>DotnetSpider, a .NET Standard web crawling library. It is lightweight, efficient and fast high-level web crawling &amp; scraping framework</Description>
@@ -16,7 +18,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <None Include="..\..\LICENSE.txt" Pack="true" PackagePath="LICENSE.txt" />
+    <None Include="..\..\LICENSE.txt" Pack="true" PackagePath="LICENSE.txt"/>
   </ItemGroup>
 
 </Project>

src/DotnetSpider.Portal/Startup.cs

@@ -120,8 +120,8 @@ public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
 			{
 				endpoints.MapHealthChecks("/health");
 				endpoints.MapControllerRoute(
-					name: "default",
-					pattern: "{controller=Home}/{action=Index}/{id?}");
+					"default",
+					"{controller=Home}/{action=Index}/{id?}");
 			});
 
 			SeedData.InitializeAsync(new PortalOptions(Configuration), app.ApplicationServices).GetAwaiter()

src/DotnetSpider.Portal/Views/Shared/_Layout.cshtml

@@ -71,6 +71,6 @@
 </environment>
 @await Html.PartialAsync("Pagination")
 @await Html.PartialAsync("Select2")
-@RenderSection("Scripts", required: false)
+@RenderSection("Scripts", false)
 </body>
 </html>

src/DotnetSpider.RabbitMQ/RabbitMQMessageQueue.cs

@@ -38,7 +38,7 @@ public RabbitMQMessageQueue(IOptions<RabbitMQOptions> options, ILoggerFactory lo
 			_logger.LogTrace("Creating RabbitMQ publish channel");
 
 			_publishChannel = _connection.CreateModel();
-			_publishChannel.ExchangeDeclare(exchange: _options.Exchange, type: "direct", durable: true);
+			_publishChannel.ExchangeDeclare(_options.Exchange, "direct", true);
 		}
 
 		private IConnectionFactory CreateConnectionFactory()
@@ -122,12 +122,12 @@ public Task ConsumeAsync(AsyncMessageConsumer<byte[]> consumer,
 
 			var channel = _connection.CreateModel();
 			var basicConsumer = new AsyncEventingBasicConsumer(channel);
-			channel.QueueDeclare(queue: consumer.Queue,
-				durable: true,
-				exclusive: false,
-				autoDelete: true,
-				arguments: null);
-			channel.QueueBind(queue: consumer.Queue, _options.Exchange, routingKey: consumer.Queue);
+			channel.QueueDeclare(consumer.Queue,
+				true,
+				false,
+				true,
+				null);
+			channel.QueueBind(consumer.Queue, _options.Exchange, consumer.Queue);
 			basicConsumer.Received += async (model, ea) =>
 			{
 				try
@@ -136,15 +136,15 @@ public Task ConsumeAsync(AsyncMessageConsumer<byte[]> consumer,
 				}
 				finally
 				{
-					channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
+					channel.BasicAck(ea.DeliveryTag, false);
 				}
 			};
 			consumer.OnClosing += x =>
 			{
 				channel.Close();
 			};
 			//7. 启动消费者
-			channel.BasicConsume(queue: consumer.Queue, autoAck: false, consumer: basicConsumer);
+			channel.BasicConsume(consumer.Queue, false, basicConsumer);
 
 			return Task.CompletedTask;
 		}

src/DotnetSpider/Infrastructure/RequestedQueue.cs

@@ -1,6 +1,6 @@
 using System;
-using System.Collections.Generic;
-using System.Runtime.CompilerServices;
+using System.Collections.Concurrent;
+using System.Threading;
 using System.Threading.Tasks;
 using DotnetSpider.Http;
 using HWT;
@@ -9,66 +9,55 @@ namespace DotnetSpider.Infrastructure
 {
 	public class RequestedQueue : IDisposable
 	{
-		private readonly Dictionary<string, Request> _dict;
-
+		private readonly ConcurrentDictionary<string, Request> _dict;
 		private readonly HashedWheelTimer _timer;
-
-		private readonly List<Request> _queue;
+		private ConcurrentBag<Request> _queue;
 
 		public RequestedQueue()
 		{
-			_dict = new Dictionary<string, Request>();
-			_queue = new List<Request>();
+			_dict = new ConcurrentDictionary<string, Request>();
+			_queue = new ConcurrentBag<Request>();
 			_timer = new HashedWheelTimer(TimeSpan.FromSeconds(1)
-				, ticksPerWheel: 100000
-				, maxPendingTimeouts: 0);
+				, 100000);
 		}
 
 		public int Count => _dict.Count;
 
-		[MethodImpl(MethodImplOptions.Synchronized)]
 		public bool Enqueue(Request request)
 		{
 			if (request.Timeout <= 2000)
 			{
 				throw new SpiderException("Timeout should not less than 2000 milliseconds");
 			}
 
-			if (!_dict.ContainsKey(request.Hash))
+			if (!_dict.TryAdd(request.Hash, request))
 			{
-				_dict.Add(request.Hash, request);
-				_timer.NewTimeout(new TimeoutTask(this, request.Hash),
-					TimeSpan.FromMilliseconds(request.Timeout));
-				return true;
+				return false;
 			}
 
-			return false;
+			_timer.NewTimeout(new TimeoutTask(this, request.Hash),
+				TimeSpan.FromMilliseconds(request.Timeout));
+			return true;
 		}
 
-		[MethodImpl(MethodImplOptions.Synchronized)]
+
 		public Request Dequeue(string hash)
 		{
-			var request = _dict[hash];
-			_dict.Remove(hash);
-			return request;
+			return _dict.TryRemove(hash, out var request) ? request : null;
 		}
 
-		[MethodImpl(MethodImplOptions.Synchronized)]
 		public Request[] GetAllTimeoutList()
 		{
 			var data = _queue.ToArray();
-			_queue.Clear();
+			Interlocked.Exchange(ref _queue, new ConcurrentBag<Request>());
 			return data;
 		}
 
-		[MethodImpl(MethodImplOptions.Synchronized)]
 		private void Timeout(string hash)
 		{
-			if (_dict.ContainsKey(hash))
+			if (_dict.TryRemove(hash, out var request))
 			{
-				var request = _dict[hash];
 				_queue.Add(request);
-				_dict.Remove(hash);
 			}
 		}
 
@@ -93,7 +82,7 @@ public Task RunAsync(ITimeout timeout)
 		public void Dispose()
 		{
 			_dict.Clear();
-			_queue.Clear();
+			Interlocked.Exchange(ref _queue, new ConcurrentBag<Request>());
 			_timer.Dispose();
 		}
 	}

src/DotnetSpider/Proxy/ProxyService.cs

@@ -30,8 +30,8 @@ public ProxyEntry(Uri uri)
 		private readonly IProxyValidator _proxyValidator;
 
 		private readonly HashedWheelTimer _timer = new HashedWheelTimer(TimeSpan.FromSeconds(1)
-			, ticksPerWheel: 100000
-			, maxPendingTimeouts: 0);
+			, 100000
+			, 0);
 
 		public ProxyService(IProxyValidator proxyValidator)
 		{