如何在执行共同任务时等待多个线程？答案

【问题标题】：How to wait multiple threads while a common task is being carried out?如何在执行共同任务时等待多个线程？
【发布时间】：2019-08-30 11:37:48
【问题描述】：

我有多个线程并行执行 API 调用。每当 jwt 令牌过期时，我希望我的所有线程等到刷新令牌 API 被调用并返回有效更新的 jwt 令牌。我有一个单例类，它有一个刷新令牌方法，它将进行调用并更新令牌。如何确保我的所有其他线程都等到令牌提取完成？

public class JWTTokenManager
    {
        private static JWTTokenManager _tokenManager;
        private string _token;
        private bool _refreshingToken;

        public static JWTTokenManager GetManager()
        {
            if (_tokenManager == null)
                _tokenManager = new JWTTokenManager();
            return _tokenManager;
        }

        public void UpdateToken(string token)
        {
            _token = token;
        }

        public string GetToken()
        {
            return _token;
        }

        public async Task<bool> ValidateRefreshTocken()
        {
            UserInfo userdata = JsonConvert.DeserializeObject<UserInfo>(GetUserInfo(_token), new Helper.DefaultJsonSetting());
            if (!string.IsNullOrWhiteSpace(userdata.Exp) && TokenExpired(long.Parse(userdata.Exp)))
            {
                _refreshingToken = true;
                JWTToken jwtToken = Database.DBService.GetDB().FetchJWTToken();
                RefreshToken requestRefresh = new RefreshToken
                {
                    ExpiredTocken = jwtToken.Token,
                    RefreshTocken = jwtToken.RefreshToken
                };
                HttpClient httpClient = CloudService.GetCloudService().GetHttpClient();
                HttpResponseMessage response = await httpClient.PostAsync($"account/v1/tokenRefresh", new StringContent(JsonConvert.SerializeObject(requestRefresh), Encoding.UTF8, "application/json"));
                bool responseStatus = await ParseTokenResponseAsync(response);
                _refreshingToken = false;
                return responseStatus;
            }
            else
            {
                return true;
            }
        }

        private string GetUserInfo(string key)
        {
            string[] base64Url = key.Split('.');
            if (base64Url.Length > 1)
            {
                string userinfo = base64Url[1];
                userinfo = userinfo.Replace(" ", "+");
                int mod4 = userinfo.Length % 4;
                if (mod4 > 0)
                {
                    userinfo += new string('=', 4 - mod4);
                }
                var base64EncodedBytes = System.Convert.FromBase64String(userinfo);
                return Encoding.UTF8.GetString(base64EncodedBytes);
            }
            else
            {
                return "";
            }
        }

        public bool TokenExpired(long unixTimeStamp)
        {
            DateTime tokenExpiryDateTime = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
            tokenExpiryDateTime = tokenExpiryDateTime.AddSeconds(unixTimeStamp).ToLocalTime();
            DateTime currentDateTime = DateTime.Now;
            return DateTime.Compare(tokenExpiryDateTime, currentDateTime) <= 0;
        }

        public async Task<bool> ParseTokenResponseAsync(HttpResponseMessage httpResponse)
        {
            if (httpResponse.IsSuccessStatusCode == true)
            {
                string responseString = await httpResponse.Content.ReadAsStringAsync();
                Newtonsoft.Json.Linq.JObject responsedataObject = Newtonsoft.Json.Linq.JObject.Parse(responseString);
                string token = responsedataObject["data"]["token"].ToString();
                string refreshToken = responsedataObject["data"]["refreshToken"].ToString();
                _token = token;
                JWTToken updatedToken = new JWTToken()
                {
                    Token = _token,
                    RefreshToken = refreshToken
                };
                Database.DBService.GetDB().InsertOrUpdateJWTToken(updatedToken);
                return true;
            }
            else
            {
                return false;
            }
        }
    }
public class CloudService
{
    private const int TIME_OUT = 50;
    private const int HTTP_GET = 0;
    private const int HTTP_PUT = 1;
    private const int HTTP_POST = 2;
    private const int HTTP_DELETE = 3;

    private static CloudService _serviceInstance;

    public static CloudService GetCloudService()
    {
        if (_serviceInstance == null)
            _serviceInstance = new CloudService();
        return _serviceInstance;
    }

    private async Task<HttpResponseMessage> ExecuteHttpTask(int taskType, string url, StringContent content = null)
    {
        HttpClient httpClient = GetHttpClient();
        switch (taskType)
        {
            case HTTP_GET:
                return await httpClient.GetAsync(url);
            case HTTP_PUT:
                return await httpClient.PutAsync(url, content);
            case HTTP_POST:
                return await httpClient.PostAsync(url, content);
            case HTTP_DELETE:
                return await httpClient.DeleteAsync(url);
            default:
                return null;
        }
    }

    public async Task<Response> HTTPTask(string url, int taskType, StringContent content = null, bool login = false)
    {
                bool refreshTocken = await JWTTokenManager.GetManager().ValidateRefreshTocken();

        Response httpResponse = new Response();
        try
        {
            HttpResponseMessage response = await ExecuteHttpTask(taskType, url, content);

            string responseString = await response.Content.ReadAsStringAsync();
            if (!response.IsSuccessStatusCode)
                httpResponse.status = "error";
            else
                httpResponse.status = "data";
            httpResponse.data = ParseResponseData(httpResponse.status, responseString);
        }
        catch (Exception e)
        {
            httpResponse = GenericErrorResponse(e.Message);
        }
        return httpResponse;
    }

    public async Task<Response> GetSectionAsync(string id)
    {
        string url = $"catalog/v2/homepageSections/{id}?order-by=name,asc";
        return await HTTPTask(url, Constants.HTTP_GET);
    }

    public async Task<Response> GetProductAsync(string id)
    {
        string url = $"catalog/v2/products/{id}";
        return await HTTPTask(url, Constants.HTTP_GET);
    }

    public async Task<Response> GetCourseDetailsAsync(string id)
    {
        string url = $"catalog/v2/products/{id}/courseDetails";
        return await HTTPTask(url, Constants.HTTP_GET);
    }

}

不同的线程会调用 ClouService 中的方法，这些方法依次并行调用不同的 API，所有这些都通过 HTTPTask 方法，在该方法中验证令牌，如果无效，则调用 API 以获取更新的令牌。如何让所有 API（线程）从令牌无效的那一刻起一直等待，直到 API 返回有效令牌？

基于 cmets，我更新了这两个类。请看一下。

public sealed class JWTTokenManager
    {
        private static readonly JWTTokenManager _tokenManager = new JWTTokenManager();
        private static SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
        private string _token;
        // Explicit static constructor to tell C# compiler not to mark type as beforefieldinit
        static JWTTokenManager()
        {
        }
        private JWTTokenManager()
        {
        }

        public static JWTTokenManager GetManager()
        {
            return _tokenManager;
        }

        public void UpdateToken(string token)
        {
            _token = token;
        }

        public string GetToken()
        {
            return _token;
        }

        public async Task<bool> ValidateRefreshTocken(bool forceRefresh = false)
        {
            bool validToken;
            await _semaphore.WaitAsync();
            try
            {
                UserInfo userdata = JsonConvert.DeserializeObject<UserInfo>(GetUserInfo(_token), new DefaultJsonSetting());
                if (forceRefresh || !string.IsNullOrWhiteSpace(userdata.Exp) && TokenExpired(long.Parse(userdata.Exp)))
                {
                    validToken = await RefreshToken();
                }
                else
                {
                    validToken = true;
                }
            }
            catch (Exception exception)
            {
                Console.WriteLine(exception);
                validToken = false;
            }
            finally
            {
                _semaphore.Release();
            }
            return validToken;
        }

        private async Task<bool> RefreshToken()
        {
            JWTToken jwtToken = Database.DBService.GetDB().FetchJWTToken();
            RefreshToken requestRefresh = new RefreshToken
            {
                ExpiredTocken = jwtToken.Token,
                RefreshTocken = jwtToken.RefreshToken
            };
            HttpClient httpClient = CloudService.GetCloudService().GetHttpClient();
            HttpResponseMessage response = await httpClient.PostAsync($"account/v1/tokenRefresh", new StringContent(JsonConvert.SerializeObject(requestRefresh), Encoding.UTF8, "application/json"));
            bool status = await ParseTokenResponseAsync(response);
            return status;
        }

        private string GetUserInfo(string key)
        {
            string[] base64Url = key.Split('.');
            if (base64Url.Length > 1)
            {
                string userinfo = base64Url[1];
                userinfo = userinfo.Replace(" ", "+");
                int mod4 = userinfo.Length % 4;
                if (mod4 > 0)
                {
                    userinfo += new string('=', 4 - mod4);
                }
                var base64EncodedBytes = System.Convert.FromBase64String(userinfo);
                return Encoding.UTF8.GetString(base64EncodedBytes);
            }
            else
            {
                return "";
            }
        }

        public bool TokenExpired(long unixTimeStamp)
        {
            DateTime tokenExpiryDateTime = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
            tokenExpiryDateTime = tokenExpiryDateTime.AddSeconds(unixTimeStamp).ToLocalTime();
            DateTime currentDateTime = DateTime.Now;
            return DateTime.Compare(tokenExpiryDateTime, currentDateTime) <= 0;
        }

        public async Task<bool> ParseTokenResponseAsync(HttpResponseMessage httpResponse)
        {
            if (httpResponse.IsSuccessStatusCode == true)
            {
                string responseString = await httpResponse.Content.ReadAsStringAsync();
                Newtonsoft.Json.Linq.JObject responsedataObject = Newtonsoft.Json.Linq.JObject.Parse(responseString);
                string token = responsedataObject["data"]["token"].ToString();
                string refreshToken = responsedataObject["data"]["refreshToken"].ToString();
                _token = token;
                JWTToken updatedToken = new JWTToken()
                {
                    Token = _token,
                    RefreshToken = refreshToken
                };
                Database.DBService.GetDB().InsertOrUpdateJWTToken(updatedToken);
                return true;
            }
            else
            {
                return false;
            }
        }
    }

public sealed class CloudService
    {
        private const int TIME_OUT = 50;
        private const int HTTP_GET = 0;
        private const int HTTP_PUT = 1;
        private const int HTTP_POST = 2;
        private const int HTTP_DELETE = 3;

        private static readonly CloudService _serviceInstance = new CloudService();
        static CloudService()
        {
        }
        private CloudService()
        {
        }

        public static CloudService GetCloudService()
        {
            return _serviceInstance;
        }

        public HttpClient GetHttpClient()
        {
            HttpClient httpClient = new HttpClient
            {
                Timeout = TimeSpan.FromSeconds(TIME_OUT),
                BaseAddress = new Uri($"{AppConst.ServerBaseURL}/"),
            };
            httpClient.DefaultRequestHeaders.Add("X-Jwt-Token", JWTTokenManager.GetManager().GetToken());
            httpClient.DefaultRequestHeaders.Add("tenantId", AppConst.TenanatID);
            return httpClient;
        }

        private async Task<HttpResponseMessage> ExecuteHttpTask(int taskType, string url, StringContent content = null)
        {
            HttpClient httpClient = GetHttpClient();
            switch (taskType)
            {
                case HTTP_GET:
                    return await httpClient.GetAsync(url);
                case HTTP_PUT:
                    return await httpClient.PutAsync(url, content);
                case HTTP_POST:
                    return await httpClient.PostAsync(url, content);
                case HTTP_DELETE:
                    return await httpClient.DeleteAsync(url);
                default:
                    return null;
            }
        }

        public async Task<Response> HTTPTask(string url, int taskType, StringContent content = null, bool login = false)
        {
            Response httpResponse = new Response();
            try
            {
                HttpResponseMessage response = await ExecuteHttpTask(taskType, url, content);
                if (response.StatusCode == System.Net.HttpStatusCode.Unauthorized && !login)
                {
                    bool refreshTocken = await JWTTokenManager.GetManager().ValidateRefreshTocken(true);
                    if (refreshTocken == true)
                    {
                        response = await ExecuteHttpTask(taskType, url, content);
                    }
                    else
                    {
                        httpResponse = GenericErrorResponse();
                        return httpResponse;
                    }
                }

                string responseString = await response.Content.ReadAsStringAsync();
                if (!response.IsSuccessStatusCode)
                    httpResponse.status = "error";
                else
                    httpResponse.status = "data";
                httpResponse.data = ParseResponseData(httpResponse.status, responseString);
            }
            catch (Exception e)
            {
                httpResponse = GenericErrorResponse(e.Message);
            }
            return httpResponse;
        }
    }

【问题讨论】：

寻找等待所有docs.microsoft.com/en-us/dotnet/api/…
无关：您的单例实现不是线程安全的。你确定你没有任何问题吗？
@Fildor 请您详细说明一下。
我的意思是理论上可能有不止一个 JWTTokenManager 实例。这可能会导致问题。尤其是在您尝试同步线程时。
您很可能忽略了隐式线程竞争错误。您无法确保令牌不会过期线程正忙于尝试使用它。所以这不是阻塞线程，你必须重新启动它们。

标签： c# multithreading asynchronous async-await thread-synchronization

【解决方案1】：

您可以使用ManualResetEvent。假设您有三个线程 A、B（并行调用 api 调用）和 C（执行令牌刷新）。

private static ManualResetEvent mre = new ManualResetEvent(true);  // event is always set except when refreshing the token

Code on thread A,B

mre.WaitOne()  // it blocks whenever the event is reset
...


Code on thread C

mre.Reset() // this blocks all the waiting threads
perform the token refresh
mre.Set()  // this frees all the waiting threads

【讨论】：

我真的不喜欢你说的mre.Reset()“阻塞所有等待的线程”。我更愿意说，“当一个线程等待事件时，它会阻塞直到事件被设置。”它需要等待线程的合作。您所说的方式在技术上是错误的，因为重置事件绝对不会阻塞已经在等待的线程。如果等待线程看到事件已设置，那么重置它以导致线程阻塞为时已晚。一个线程尤其不能等待最终导致它阻塞——我希望澄清一个重要方面。
> 一个线程必须不等待最终导致它阻塞 - 我希望明确的一个重要方面令人惊讶的是，它是等待调用使线程阻塞，如果它是不等待它不会阻塞（除非你在它上面调用 Thread.Suspend）
嗯，Event 是一个同步工具，所以所有线程都必须合作。阻塞线程不是用户程序功能（即使您可以使用 Thread.Suspend），而是操作系统功能（用户程序不能这样做）。我认为提出问题的人可能希望避免编写代码来检查令牌是否在每次函数调用后过期并将其更新委托给另一个线程，但你不应该这样做（你应该辞职写每个函数调用后的样板代码）。

【解决方案2】：

你应该让JWTTokenManager返回一个令牌而不是它需要一个令牌。

如果令牌有效，它将返回一个带有有效令牌的已完成任务，如果不是，它将返回一个在检索到令牌时将完成的任务。多个并发线程可以等待同一个任务。

public class JWTTokenManager
{
    private Task<string> tokenTask;
    private readonly object sync = new object();

    public Task<string> GetTokenAsync()
    {
        lock (sync)
        {
            if (tokenTask.IsCompleted && !IsTokenValid(tokenTask.Result))
            {
                tokenTask = GetNewTokenAsync();
            }

            return tokenTask;
        }
    }
}

无锁版：

public class JWTTokenManager
{
    private Task<string> tokenTask;
    private readonly object sync = new object();

    public Task<string> GetTokenAsync()
    {
        var currentTokenTask = Volatile.Read(ref tokenTask);

        if (currentTokenTask .IsCompleted && !IsTokenValid(currentTokenTask .Result))
        {
            currentTokenTask = GetNewTokenAsync();
            Volatile.Write(ref tokenTask, currentTokenTask);
        }

        return currentTokenTask ;
    }
}

【讨论】：

锁定的块可以包含异步方法吗？
假设 5 个线程需要并行获取令牌，如果现有令牌已过期，那么所有五个线程都将尝试获取更新的新令牌，因此我希望所有其他线程等到一个/第一个请求获取更新的令牌，其他人继续。可以吗？
一个块可以包含异步方法，或者准确地说，我得到的错误是“不能在锁语句体中等待”
@MidhunKumar，异步方法在哪里？
@MidhunKumar，如果并发真的发生并且是问题（无论出于何种原因），你可以有一个信号量。但这只有在并发频繁且获取令牌成本高昂时才会出现问题。