//--------------------------------------------------------------------------
//
// Copyright (c) BUSHUOSX. All rights reserved.
//
// File: HttpClientHelper.cs
//
// Version:1.0.0.1
//
// Datetime:20170815
//
//--------------------------------------------------------------------------
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
namespace BUSHUOSX.HTTP
{
class HttpClientHelper
{
/// <summary>
/// http task state
/// </summary>
class TASKSTATE
{
public object userState;
public HttpClient httpClient;
public TASKSTATE(object state,HttpClient client)
{
userState = state;
httpClient = client;
}
}
/// <summary>
/// 不为Null时，任意http请求完成、取消、失败时回调
/// </summary>
Action<Task, object> _callbackOnAnyTaskComplited;
/// <summary>
/// 并发控制信号
/// </summary>
SemaphoreSlim _semaphore;
/// <summary>
/// 工作标记
/// </summary>
bool _isWorking;
/// <summary>
/// 共用httpclient。为Null时，为每个http请求new httpclient。
/// </summary>
HttpClient _httpClient;
/// <summary>
/// http request task 记录
/// </summary>
readonly ConcurrentDictionary<Task, TASKSTATE> _workingTasks = new ConcurrentDictionary<Task, TASKSTATE>();
/// <summary>
/// 启动的task计数。
/// </summary>
long _workingCount;//原子操作
/// <summary>
/// 任务取消标记
/// </summary>
CancellationTokenSource _cancelSource;
/// <summary>
/// 工作器监视线程
/// </summary>
Task _worker;
/// <summary>
/// 禁止自动启动任务
/// </summary>
bool _autoStartWorking = true;
/// <summary>
/// Http Request最大并发数
/// </summary>
public int MaxConcurrencyLevel { get; }
/// <summary>
/// Http Request超时设置，毫秒。小于等于0时，使用默认值。
/// </summary>
public int TimeoutMillisecond { get; }
/// <summary>
/// 可控制并发数的HttpClient
/// </summary>
/// <param name="concurrencyLevel">Http Request最大并发数</param>
/// <param name="callbackOnAnyTaskComplited">不为Null时，任意http请求完成、取消、失败时回调。注意：如果使用，请保证方法的线程安全。</param>
/// <param name="timeoutMillisecond">Http Request超时设置，毫秒。小于等于0时，使用默认值。</param>
/// <param name="httpClient">留空则每个任务使用一个新的HttpClient对象</param>
public HttpClientHelper(int concurrencyLevel, Action<Task, object> callbackOnAnyTaskComplited = null, int timeoutMillisecond = 0, HttpClient httpClient = null)
{
if (concurrencyLevel < 1)
{
throw new ArgumentOutOfRangeException("concurrencyLevel < 1");
}
MaxConcurrencyLevel = concurrencyLevel;
_callbackOnAnyTaskComplited = callbackOnAnyTaskComplited;
TimeoutMillisecond = timeoutMillisecond;
_httpClient = httpClient;
}
/// <summary>
/// 返回已发出的http请求数
/// </summary>
/// <returns></returns>
public int GetWorkingTasksCount()
{
return _workingTasks.Count;
}
//public KeyValuePair<Task, object>[] GetTasks()
//{
// return _workingTasks.ToArray();
//}
/// <summary>
/// 停止工作器
/// </summary>
/// <param name="cancelPendingRequests">取消已发起的所有http请求</param>
public void Stop(bool cancelPendingRequests = false)
{
_autoStartWorking = false;
notifyEndWorker();
_cancelSource.Cancel();
if (_httpClient != null)
{
_httpClient.CancelPendingRequests();
}
else
{
foreach (var item in _workingTasks)
{
if (item.Value.httpClient != null)
{
item.Value.httpClient.CancelPendingRequests();
}
}
}
_workingTasks.Clear();
}
/// <summary>
/// 重新启动工作器。stop之后使用。
/// </summary>
public void ReStart()
{
_autoStartWorking = true;
}
/// <summary>
/// 阻塞直到所有http任务完成。
/// </summary>
public void WaitWorkComplited()
{
if (_worker != null)
{
_worker.Wait();
}
}
/// <summary>
/// 通过 http client 获取 string。超过并发数时会阻塞，直到有http request完成。
/// </summary>
/// <param name="address">http url string</param>
/// <param name="state">用户状态。用于callbackOnAnyTaskComplited回调。</param>
/// <returns></returns>
public Task<string> GetString(string address, object state = null)
{
Interlocked.Increment(ref _workingCount);
notifyStartWorker();
if (!_isWorking)
{
Interlocked.Decrement(ref _workingCount);
return null;
}
_semaphore.Wait(_cancelSource.Token);
TASKSTATE tstate = new TASKSTATE(state, null);
HttpClient client;
try
{
if (_httpClient != null)
{
client = _httpClient;
}
else
{
client = new HttpClient();
if (TimeoutMillisecond > 0)
{
client.Timeout = TimeSpan.FromMilliseconds(TimeoutMillisecond);
}
tstate.httpClient = client;
}
var t = client.GetStringAsync(address);
_workingTasks[t] = tstate;
t.ContinueWith(anyTaskComplited,
_cancelSource.Token,
TaskContinuationOptions.OnlyOnRanToCompletion & TaskContinuationOptions.NotOnRanToCompletion,
TaskScheduler.Current);
return t;
}
catch (HttpRequestException e)
{
Interlocked.Decrement(ref _workingCount);
_semaphore.Release();
throw new HttpRequestException(e.Message, e.InnerException);
}
}
/// <summary>
/// 任意http task完成时回调
/// </summary>
/// <param name="task"></param>
private void anyTaskComplited(Task task)
{
TASKSTATE tstate;
_workingTasks.TryRemove(task, out tstate);
//Debug.Assert(tstate.userState == state);
_callbackOnAnyTaskComplited?.Invoke(task, tstate.userState);
_semaphore.Release();
Interlocked.Decrement(ref _workingCount);
}
/// <summary>
/// 工作器初始化
/// </summary>
private void notifyStartWorker()
{
if (_isWorking) return;
if (!_autoStartWorking) return;
//初始化
_isWorking = true;
Debug.WriteLine("httpClientWorker启动……");
_semaphore = new SemaphoreSlim(MaxConcurrencyLevel, MaxConcurrencyLevel);
_cancelSource = new CancellationTokenSource();
//_workingCount = 0;
if (_httpClient != null && TimeoutMillisecond > 0)
{
_httpClient.Timeout = TimeSpan.FromMilliseconds(TimeoutMillisecond);
}
_worker = Task.Run(new Action(workerMonitor), _cancelSource.Token);
_worker.ContinueWith(a=>{ notifyEndWorker(); },TaskContinuationOptions.NotOnRanToCompletion & TaskContinuationOptions.OnlyOnRanToCompletion);
}
/// <summary>
/// 工作器结束清理工作
/// </summary>
private void notifyEndWorker()
{
if (_isWorking)
{
_isWorking = false;
Debug.WriteLine("httpClientWorker结束……");
}
}
/// <summary>
/// 任务工作器监视器
/// </summary>
private void workerMonitor()
{
do
{
Thread.Sleep(1000);
} while (_workingTasks.Count > 0 || Interlocked.Read(ref _workingCount) > 0);
}
}
}
