WEP_V2.1.3/SourceCode/We.Engine/WeEngine.Base/Message/ScenePusher.cs

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace WeEngine.Message
{
    public class ScenePusher
    {
        private volatile ConcurrentQueue<ScenePushMessage> _messages;
        private CancellationTokenSource _cts;

        private bool _switch;

        /// <summary>
        /// 发送消息事件
        /// </summary>
        public event SendMsgEventHandle SendMsg;

        public delegate void SendMsgEventHandle(ScenePushMessage msg);

        public ScenePusher()

        {
            _switch = false;
            _messages = new ConcurrentQueue<ScenePushMessage>();
            _cts = new CancellationTokenSource();
        }

        public bool NeedHandel => SendMsg == null;

        public void Push(ScenePushMessage message)
        {
            lock (_messages)
            {
                _messages.Enqueue(message);
            }
        }

        public void Stop()
        {
            lock (_messages)
            {
                _messages = new ConcurrentQueue<ScenePushMessage>();
            }
            _switch = false;
            _cts.Cancel();
        }

        public bool Start()
        {
            if (!_switch)
            {
                _switch = true;
                if (_cts.IsCancellationRequested)
                {
                    _cts = new CancellationTokenSource();
                }
                Listener(_cts.Token);
                return false;
            }
            return true;
        }

        /// <summary>
        /// 监听
        /// </summary>
        /// <param name="cancellationToken"></param>
        private void Listener(CancellationToken cancellationToken)
        {
            Task.Factory.StartNew(() =>
            {
                while (!cancellationToken.IsCancellationRequested)//如果没有取消线程，则一直监听执行发送消息
                {
                    // ReSharper disable  InconsistentlySynchronizedField
                    if (_messages.Count > 0)
                    {
                        var msg = DequeueS();

                        if (msg != null)
                        {
                            SendMsg?.Invoke(msg);
                        }
                    }
                    else
                    {
                        SpinWait.SpinUntil(() => _messages.Count >= 1, 5000);//自旋等待直到消息队列有>=1的记录或超时5S后再进入下一轮的判断
                    }
                }
            }, cancellationToken);
        }

        /// <summary>
        /// 取待发送消息
        /// </summary>
        /// <returns></returns>
        private ScenePushMessage DequeueS()
        {
            lock (_messages)
            {
                if (_messages.Count <= 0)
                {
                    return null;
                }
                _messages.TryDequeue(out var msg);
                return msg;
            }
        }
    }
}