test_demo/MX-PD-盘古 - new/PanGu.DieBonderApp/MainShell/Filewritable/FileWriteQueue.cs

using MainShell.Log;
using MaxwellFramework.Core.Attributes;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace MXJM.FileWritable
{
    public class FileWriteQueue : IDisposable
    {
        private readonly Channel<IFileWritable> _fileChannel;
        private readonly int _capacity;
        private bool _isClosed = false;

        public FileWriteQueue(int capacity = 50)
        {
            _capacity = capacity;
            var options = new BoundedChannelOptions(capacity)
            {
                FullMode = BoundedChannelFullMode.Wait
            };
            _fileChannel = Channel.CreateBounded<IFileWritable>(options);
        }

        /// <summary>
        /// 当前队列长度
        /// </summary>
        public int Count => _fileChannel.Reader.Count;

        /// <summary>
        /// 队列容量
        /// </summary>
        public int Capacity => _capacity;

        /// <summary>
        /// 入队（同步，立即返回是否成功）
        /// </summary>
        public bool Enqueue(IFileWritable fileWritable)
        {
            if (_isClosed) return false;
            return _fileChannel.Writer.TryWrite(fileWritable);
        }

        /// <summary>
        /// 入队（异步，支持等待）
        /// </summary>
        public async Task<bool> EnqueueAsync(IFileWritable fileWritable, CancellationToken cancellationToken = default)
        {
            if (_isClosed) return false;
            try
            {
                await _fileChannel.Writer.WriteAsync(fileWritable, cancellationToken);
                return true;
            }
            catch (OperationCanceledException)
            {
                return false;
            }
            catch (Exception ex)
            {
                LogException(ex, "EnqueueAsync", fileWritable.Description);
                return false;
            }
        }

        /// <summary>
        /// 处理队列
        /// </summary>
        public async Task ProcessQueueAsync(CancellationToken cancellationToken = default, bool throwOnError = false)
        {
            try
            {
                while (await _fileChannel.Reader.WaitToReadAsync(cancellationToken))
                {
                    while (_fileChannel.Reader.TryRead(out IFileWritable fileWritable))
                    {
                        try
                        {
                            await Task.Run(() => fileWritable.Save(), cancellationToken);
                        }
                        catch (Exception ex)
                        {
                            LogException(ex, "ProcessQueueAsync", fileWritable.Description);
                            if (throwOnError)
                            {
                                throw;
                            }
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                if (throwOnError)
                {
                    throw;
                }
            }
        }

        /// <summary>
        /// 关闭队列，不再接受新任务
        /// </summary>
        public void Close()
        {
            if (!_isClosed)
            {
                _isClosed = true;
                _fileChannel.Writer.TryComplete();
            }
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            Close();
        }

        private void LogException(Exception ex, string method,string description)
        {
            LogManager.LogSysError($"{description}:[FileWriteQueue][{method}] 异常: {ex}");
        }
    }
}