【aspnetcore】用ConcurrentQueue實現一個簡單的隊列系統

第一步：定義隊列服務接口

public interface ISimpleQueueServer
{
    /// <summary>
    /// 添加隊列消息
    /// </summary>
    /// <param name="message">消息</param>
    /// <param name="clientName">客戶端名稱</param>
    /// <returns></returns>
    string Add(string message, string clientName);
}

第二步：添加隊列服務接口的實現

public class SimpleQueueServer : ISimpleQueueServer
{
    /// <summary>
    /// 隊列
    /// </summary>
    private static ConcurrentQueue<string> _queue = new ConcurrentQueue<string>();

    /// <summary>
    /// 日志
    /// </summary>
    private static ILogger _log;

    /// <summary>
    /// 后台任務
    /// </summary>
    private static Task _task;

    /// <summary>
    /// 連續獲取隊列為空的次數
    /// </summary>
    private int EmptyRepeatCount = 0;

    /// <summary>
    /// 屬性，后台任務
    /// </summary>
    private Task MyTask
    {
        get
        {
            if (_task == null)
            {
                _task = new Task(MessageHandler);
            }
            return _task;
        }
    }

    /// <summary>
    /// 構造函數
    /// </summary>
    /// <param name="factory"></param>
    public SimpleQueueServer(ILoggerFactory factory)
    {
        if (_log == null)
        {
            _log = factory.CreateLogger("SimpleQueueServer");
        }

        MyTask.Start();
    }

    /// <summary>
    /// 添加消息到隊列
    /// </summary>
    /// <param name="message">消息</param>
    /// <param name="clientName">發送的客戶端名稱</param>
    /// <returns></returns>
    public string Add(string message, string clientName = "")
    {
        try
        {
            string prefix = string.IsNullOrWhiteSpace(clientName) ? "" : $"【{clientName}】";
            _queue.Enqueue($"{prefix}{message}");
            return "OK";
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "向隊列添加信息失敗");
            return ex.Message;
        }
    }

    /// <summary>
    /// 隊列中要實現的任務
    /// </summary>
    /// <param name="threadName">線程名稱，如果多</param>
    /// <returns></returns>
    private Action MessageHandler => () => 
    {
        while (true)
        {
            try
            {
                if (_queue.IsEmpty)
                {
                    _log.LogDebug($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}    隊列為空");
                    Thread.Sleep(3000);
                }
                else
                {
                    if (_queue.TryDequeue(out string result))
                    {
                        _log.LogDebug($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}    獲取到數據：{result}");
                    }
                    else
                    {
                        _log.LogDebug($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}    嘗試從隊列獲取消息失敗");
                    }
                    Thread.Sleep(500);
                }
            }
            catch (Exception ex)
            {
                _log.LogError(ex, "系統錯誤");
            }
        }
    };
}

第三步：在startup中注冊服務，這里稍稍裝個x，定義一個IServiceCollection擴展，讓代碼看起來x格稍微高點

public static class ServiceCollectionExtension
{
    public static void AddSimpleQueueServer(this IServiceCollection services)
    {
        services.AddSingleton<ISimpleQueueServer, SimpleQueueServer>((provider) =>
        {
            return new SimpleQueueServer(provider.GetService<ILoggerFactory>());
        });
    }
}

第四步：在startup的ConfigureServices中添加服務

services.AddSimpleQueueServer();

第五步：修改appsettings.Development.json文件

{
  "Logging": {
    "LogLevel": {
      "Default": "Debug",
      "System": "Warning",
      "Microsoft": "Warning"
    }
  }
}

修改System和Microsoft的日志級別，防止調試時隊列顯示的消息淹沒在無窮無盡的info中。

=======無聊的分割線======

新建一個QueueController

namespace AspnetCoreMvcStudy.Controllers
{
    public class QueueController : Controller
    {
        private ISimpleQueueServer _server;

        public QueueController(ISimpleQueueServer server)
        {
            _server = server;
        }

        public IActionResult Index()
        {
            return View();
        }

        [HttpPost]
        public JsonResult Send(string msg, string client)
        {
            for (int i = 0; i < 100; i++)
            {
                _server.Add($"{msg}-{i}", client);
            }
            return Json(new { Code = 200 });
        }
    }
}

創建視圖 Index.cshtml

@{
    ViewData["Title"] = "Index";
}

<h2>隊列測試</h2>

<form id="form1" onsubmit="return false;">
    <div class="form-group">
        <label for="message">客戶端名稱</label>
        <input type="text" id="client" value="" />
        <label for="message">發送內容</label>
        <input type="text" id="message" value="" />
        <hr />
        <button id="btnSubmit" class="btn btn-success">發送</button>
        <span class="text-danger" id="info"></span>
    </div>
</form>

@section scripts 
{
    <script>
        $('#btnSubmit').on('click', function () {
            var that = $(this);
            that.attr('disabled', 'disabled');
            $.post('/Queue/Send', { msg: $('#message').val(), client: $('#client').val() }, function (response) {
                if (response.code == 200) {
                    $('#info').text('發送成功');
                } else {
                    $('#info').text(response.message);
                }
                that.removeAttr('disabled');
            });
        });
    </script>
}

運行程序，開整