TJC.Cyclops.Threading 2025.12.15.1

Cyclops.Threading

项目概述

Cyclops.Threading是Cyclops.Framework框架中的线程同步与并发控制组件，提供高级线程管理、同步原语封装、并发集合和异步编程工具。该组件简化了.NET应用程序中的并发编程模型，提供了线程安全的对象池、读写锁、信号量、屏障等同步机制，以及更高级的异步协调原语，帮助开发者构建高性能、线程安全的应用程序。

核心功能模块

线程池管理

• 自定义线程池实现
• 工作线程监控
• 线程优先级控制
• 任务队列管理

同步原语

• 读写锁封装
• 信号量与互斥体
• 屏障与倒计时事件
• 自旋锁与轻量级同步

并发集合

• 线程安全字典
• 线程安全队列
• 线程安全栈
• 并发集合扩展方法

异步编程工具

• 异步锁
• 异步信号量
• 异步事件
• 异步协调器

对象池

• 通用对象池实现
• 资源池化管理
• 池化策略配置
• 内存优化工具

高级功能

• 超时控制机制
• 取消令牌扩展
• 异常传播处理
• 并发模式模式实现

技术栈

• .NET 8.0
• System.Threading.Tasks.Extensions
• System.Collections.Concurrent
• Cyclops.Common

环境依赖

• .NET 8.0 SDK

安装配置

NuGet安装

Install-Package Cyclops.Threading

基本配置

在应用程序启动时进行配置：

// 在Program.cs或Startup.cs中
using Cyclops.Threading;

var builder = WebApplication.CreateBuilder(args);

// 添加Cyclops.Threading服务
builder.Services.AddCyclopsThreading(options => {
    // 线程池配置
    options.ThreadPoolOptions = new ThreadPoolOptions {
        MinThreads = 4,
        MaxThreads = 20,
        IdleTimeout = TimeSpan.FromMinutes(5)
    };
    
    // 对象池配置
    options.ObjectPoolOptions = new ObjectPoolOptions {
        DefaultPoolSize = 100,
        MaxPoolSize = 500,
        EnableMetrics = true
    };
    
    // 并发控制配置
    options.ConcurrencyOptions = new ConcurrencyOptions {
        DefaultTimeout = TimeSpan.FromSeconds(30),
        EnableDeadlockDetection = true,
        DeadlockTimeout = TimeSpan.FromSeconds(10)
    };
});

// ...

代码示例

线程同步示例

using Cyclops.Threading.Sync;
using Microsoft.Extensions.DependencyInjection;

// 使用读写锁
public class ReadWriteLockExample
{
    private readonly IReadWriteLock _readWriteLock;
    private readonly Dictionary<string, string> _sharedData = new();
    
    public ReadWriteLockExample(IReadWriteLock readWriteLock)
    {
        _readWriteLock = readWriteLock;
    }
    
    public void UpdateData(string key, string value)
    {
        // 获取写锁
        using (_readWriteLock.AcquireWriteLock())
        {
            _sharedData[key] = value;
            Console.WriteLine($"已更新数据: {key} = {value}");
        }
    }
    
    public string GetData(string key)
    {
        // 获取读锁
        using (_readWriteLock.AcquireReadLock())
        {
            if (_sharedData.TryGetValue(key, out var value))
            {
                Console.WriteLine($"读取数据: {key} = {value}");
                return value;
            }
            return null;
        }
    }
    
    // 带超时的锁获取
    public bool TryUpdateData(string key, string value, TimeSpan timeout)
    {
        try
        {
            // 尝试在指定时间内获取写锁
            using (_readWriteLock.TryAcquireWriteLock(timeout))
            {
                _sharedData[key] = value;
                Console.WriteLine($"已更新数据: {key} = {value}");
                return true;
            }
        }
        catch (TimeoutException)
        {
            Console.WriteLine($"获取锁超时: {key}");
            return false;
        }
    }
}

// 使用信号量
public class SemaphoreExample
{
    private readonly ISemaphore _semaphore;
    
    public SemaphoreExample(ISemaphoreFactory semaphoreFactory)
    {
        // 创建最多允许3个并发访问的信号量
        _semaphore = semaphoreFactory.Create("LimitedResourceSemaphore", 3, 3);
    }
    
    public async Task AccessLimitedResourceAsync(string resourceId)
    {
        Console.WriteLine($"等待访问资源: {resourceId}");
        
        // 获取信号量
        await _semaphore.WaitAsync();
        
        try
        {
            Console.WriteLine($"开始访问资源: {resourceId}");
            // 模拟资源访问
            await Task.Delay(2000);
            Console.WriteLine($"完成访问资源: {resourceId}");
        }
        finally
        {
            // 释放信号量
            _semaphore.Release();
        }
    }
    
    // 模拟并发访问
    public async Task SimulateConcurrentAccessAsync()
    {
        var tasks = new List<Task>();
        
        for (int i = 1; i <= 10; i++)
        {
            var resourceId = $"Resource-{i}";
            tasks.Add(AccessLimitedResourceAsync(resourceId));
        }
        
        await Task.WhenAll(tasks);
    }
}

并发集合示例

using Cyclops.Threading.Collections;
using Microsoft.Extensions.DependencyInjection;

// 使用线程安全集合
public class ConcurrentCollectionExample
{
    // 线程安全字典示例
    private readonly IConcurrentDictionary<string, Product> _products;
    
    // 线程安全队列示例
    private readonly IConcurrentQueue<Order> _orders;
    
    public ConcurrentCollectionExample(
        IConcurrentDictionaryFactory dictionaryFactory,
        IConcurrentQueueFactory queueFactory)
    {
        _products = dictionaryFactory.Create<string, Product>("ProductsDictionary");
        _orders = queueFactory.Create<Order>("OrdersQueue");
    }
    
    // 使用线程安全字典
    public void ManageProducts()
    {
        // 添加产品
        _products.AddOrUpdate("P001", new Product { Id = "P001", Name = "产品A", Price = 99.99 },
            (key, existingValue) => {
                existingValue.Price = 99.99;
                return existingValue;
            });
        
        // 获取产品
        if (_products.TryGetValue("P001", out var product))
        {
            Console.WriteLine($"找到产品: {product.Name}, 价格: {product.Price}");
        }
        
        // 条件更新产品
        _products.AddOrUpdate("P001", 
            key => throw new InvalidOperationException("产品不存在"),
            (key, existingValue) => {
                if (existingValue.Price < 100)
                {
                    existingValue.Price = 109.99;
                    return existingValue;
                }
                return existingValue;
            });
        
        // 删除产品
        _products.TryRemove("P001", out _);
        Console.WriteLine($"产品是否存在: {_products.ContainsKey("P001")}");
    }
    
    // 使用线程安全队列
    public async Task ProcessOrdersAsync()
    {
        // 入队订单
        for (int i = 1; i <= 5; i++)
        {
            _orders.Enqueue(new Order { OrderId = $"O{1000 + i}", CustomerId = $"C{100 + i}", Amount = i * 100 });
        }
        
        Console.WriteLine($"队列中的订单数量: {_orders.Count}");
        
        // 并行处理订单
        var processingTasks = new List<Task>();
        
        for (int i = 0; i < 3; i++)
        {
            processingTasks.Add(Task.Run(async () => {
                while (_orders.TryDequeue(out var order))
                {
                    Console.WriteLine($"处理订单: {order.OrderId}, 客户: {order.CustomerId}, 金额: {order.Amount}");
                    // 模拟订单处理
                    await Task.Delay(500);
                }
            }));
        }
        
        await Task.WhenAll(processingTasks);
        Console.WriteLine($"处理后队列中的订单数量: {_orders.Count}");
    }
}

public class Product
{
    public string Id { get; set; }
    public string Name { get; set; }
    public double Price { get; set; }
}

public class Order
{
    public string OrderId { get; set; }
    public string CustomerId { get; set; }
    public double Amount { get; set; }
}

异步同步原语示例

using Cyclops.Threading.Async;
using Microsoft.Extensions.DependencyInjection;
using System.Threading.Tasks;

// 使用异步同步原语
public class AsyncSynchronizationExample
{
    private readonly IAsyncLock _asyncLock;
    private readonly IAsyncSemaphore _asyncSemaphore;
    private readonly IAsyncEvent _asyncEvent;
    
    private int _counter = 0;
    private bool _dataReady = false;
    
    public AsyncSynchronizationExample(
        IAsyncLockFactory asyncLockFactory,
        IAsyncSemaphoreFactory asyncSemaphoreFactory,
        IAsyncEventFactory asyncEventFactory)
    {
        _asyncLock = asyncLockFactory.Create("CounterLock");
        _asyncSemaphore = asyncSemaphoreFactory.Create("LimitedOperationSemaphore", 5);
        _asyncEvent = asyncEventFactory.CreateAutoResetEvent("DataReadyEvent");
    }
    
    // 使用异步锁保护共享资源
    public async Task<int> IncrementCounterAsync()
    {
        using (await _asyncLock.LockAsync())
        {
            Console.WriteLine($"线程 {Environment.CurrentManagedThreadId} 获取了异步锁");
            // 模拟工作
            await Task.Delay(100);
            _counter++;
            return _counter;
        }
        // 锁在这里自动释放
    }
    
    // 使用带超时的异步锁
    public async Task<int?> TryIncrementCounterWithTimeoutAsync(TimeSpan timeout)
    {
        try
        {
            using (await _asyncLock.LockAsync(timeout))
            {
                _counter++;
                return _counter;
            }
        }
        catch (TimeoutException)
        {
            Console.WriteLine("获取异步锁超时");
            return null;
        }
    }
    
    // 使用异步信号量控制并发
    public async Task PerformLimitedOperationAsync(string operationId)
    {
        await _asyncSemaphore.WaitAsync();
        
        try
        {
            Console.WriteLine($"执行受限操作: {operationId}, 可用资源: {_asyncSemaphore.CurrentCount}");
            // 模拟资源密集型操作
            await Task.Delay(2000);
        }
        finally
        {
            _asyncSemaphore.Release();
        }
    }
    
    // 使用异步事件进行线程间通信
    public async Task ProducerConsumerExampleAsync()
    {
        // 消费者任务
        var consumerTask = Task.Run(async () => {
            Console.WriteLine("消费者等待数据准备就绪...");
            await _asyncEvent.WaitAsync();
            Console.WriteLine("消费者检测到数据已准备就绪，开始处理");
            _dataReady = false;
            // 处理数据...
        });
        
        // 生产者准备数据
        await Task.Delay(3000);
        Console.WriteLine("生产者准备数据完成");
        _dataReady = true;
        
        // 通知消费者
        _asyncEvent.Set();
        
        await consumerTask;
    }
    
    // 模拟并发场景
    public async Task SimulateConcurrencyAsync()
    {
        // 1. 测试异步锁
        var counterTasks = new List<Task>();
        for (int i = 0; i < 10; i++)
        {
            counterTasks.Add(IncrementCounterAsync());
        }
        await Task.WhenAll(counterTasks);
        Console.WriteLine($"最终计数器值: {_counter}");
        
        // 2. 测试异步信号量
        var operationTasks = new List<Task>();
        for (int i = 0; i < 15; i++)
        {
            operationTasks.Add(PerformLimitedOperationAsync($"Op-{i}"));
        }
        await Task.WhenAll(operationTasks);
        
        // 3. 测试异步事件
        await ProducerConsumerExampleAsync();
    }
}

对象池示例

using Cyclops.Threading.Pooling;
using Microsoft.Extensions.DependencyInjection;
using System.Text;

// 使用对象池
public class ObjectPoolExample
{
    private readonly IObjectPool<StringBuilder> _stringBuilderPool;
    private readonly IObjectPool<DbConnection> _dbConnectionPool;
    
    public ObjectPoolExample(
        IObjectPoolFactory objectPoolFactory)
    {
        // 创建StringBuilder对象池
        _stringBuilderPool = objectPoolFactory.CreatePool<StringBuilder>(new StringBuilderPooledObjectPolicy {
            InitialCapacity = 1024,
            MaximumRetainedCapacity = 4096
        });
        
        // 创建数据库连接池
        _dbConnectionPool = objectPoolFactory.CreatePool<DbConnection>(new DbConnectionPooledObjectPolicy {
            ConnectionString = "Server=localhost;Database=TestDb;User Id=sa;Password=yourpassword;"
        });
    }
    
    // 使用StringBuilder对象池
    public string BuildLargeString(IEnumerable<string> items)
    {
        // 从池中获取StringBuilder
        var sb = _stringBuilderPool.Get();
        
        try
        {
            sb.Clear(); // 确保对象状态干净
            
            foreach (var item in items)
            {
                sb.AppendLine(item);
            }
            
            return sb.ToString();
        }
        finally
        {
            // 归还到池中
            _stringBuilderPool.Return(sb);
        }
    }
    
    // 使用数据库连接池
    public async Task<int> ExecuteDatabaseQueryAsync(string sql, params object[] parameters)
    {
        // 从池中获取连接
        var connection = _dbConnectionPool.Get();
        
        try
        {
            // 确保连接已打开
            if (connection.State != ConnectionState.Open)
            {
                await connection.OpenAsync();
            }
            
            // 执行查询
            using (var command = connection.CreateCommand())
            {
                command.CommandText = sql;
                
                // 添加参数
                for (int i = 0; i < parameters.Length; i++)
                {
                    var parameter = command.CreateParameter();
                    parameter.ParameterName = $"@p{i}";
                    parameter.Value = parameters[i] ?? DBNull.Value;
                    command.Parameters.Add(parameter);
                }
                
                return await command.ExecuteNonQueryAsync();
            }
        }
        catch (Exception ex)
        {
            // 连接出错，不归还到池中
            Console.WriteLine($"数据库操作异常: {ex.Message}");
            connection.Dispose();
            throw;
        }
        finally
        {
            // 归还连接到池中
            if (connection.State == ConnectionState.Open)
            {
                try
                {
                    connection.Close();
                }
                catch { }
            }
            
            _dbConnectionPool.Return(connection);
        }
    }
    
    // 对象池使用示例
    public async Task DemonstrateObjectPoolsAsync()
    {
        // 测试StringBuilder对象池
        var largeData = Enumerable.Range(1, 1000).Select(i => $"Item {i}: This is a test string that would be appended to the StringBuilder");
        
        var sw = Stopwatch.StartNew();
        for (int i = 0; i < 100; i++)
        {
            var result = BuildLargeString(largeData);
            // 仅使用前100个字符进行验证
            Console.WriteLine($"构建的字符串长度: {result.Length}");
        }
        sw.Stop();
        Console.WriteLine($"使用对象池构建字符串耗时: {sw.ElapsedMilliseconds}ms");
        
        // 测试数据库连接池
        var query = "INSERT INTO TestTable (Name, Value) VALUES (@p0, @p1)";
        var batchSize = 100;
        
        sw.Restart();
        for (int i = 0; i < batchSize; i++)
        {
            await ExecuteDatabaseQueryAsync(query, $"Test-{i}", i * 10);
        }
        sw.Stop();
        Console.WriteLine($"使用连接池执行{batchSize}条SQL语句耗时: {sw.ElapsedMilliseconds}ms");
    }
}

// 自定义池化对象策略
public class StringBuilderPooledObjectPolicy : IPooledObjectPolicy<StringBuilder>
{
    public int InitialCapacity { get; set; } = 100;
    public int MaximumRetainedCapacity { get; set; } = 1024;
    
    public StringBuilder Create()
    {
        return new StringBuilder(InitialCapacity);
    }
    
    public bool Return(StringBuilder obj)
    {
        // 重置对象状态
        obj.Clear();
        
        // 超过最大容量的对象不再归还池中
        return obj.Capacity <= MaximumRetainedCapacity;
    }
}

public class DbConnectionPooledObjectPolicy : IPooledObjectPolicy<DbConnection>
{
    public string ConnectionString { get; set; }
    
    public DbConnection Create()
    {
        // 根据实际使用的数据库返回对应的连接对象
        // 这里以SQL Server为例
        return new SqlConnection(ConnectionString);
    }
    
    public bool Return(DbConnection obj)
    {
        // 检查连接是否有效
        if (obj.State != ConnectionState.Closed)
        {
            try
            {
                obj.Close();
            }
            catch
            {
                // 连接已损坏，不归还
                return false;
            }
        }
        
        // 连接有效，可以归还池中
        return true;
    }
}

// 用于示例的DbConnection抽象
public abstract class DbConnection : IDisposable
{
    public abstract ConnectionState State { get; }
    public abstract void Close();
    public abstract Task OpenAsync();
    public abstract DbCommand CreateCommand();
    public abstract void Dispose();
}

public enum ConnectionState
{
    Closed,
    Open
}

public abstract class DbCommand
{
    public abstract string CommandText { get; set; }
    public abstract IDataParameterCollection Parameters { get; }
    public abstract Task<int> ExecuteNonQueryAsync();
    public abstract DbParameter CreateParameter();
}

public abstract class DbParameter : IDataParameter
{
    public abstract string ParameterName { get; set; }
    public abstract object Value { get; set; }
    // 其他接口成员...
}

public interface IDataParameterCollection : IList, ICollection, IEnumerable
{}

public interface IDataParameter
{
    string ParameterName { get; set; }
    object Value { get; set; }
    // 其他接口成员...
}

public class SqlConnection : DbConnection
{
    private ConnectionState _state = ConnectionState.Closed;
    private readonly string _connectionString;
    
    public SqlConnection(string connectionString)
    {
        _connectionString = connectionString;
    }
    
    public override ConnectionState State => _state;
    
    public override void Close()
    {
        _state = ConnectionState.Closed;
        Console.WriteLine("SQL连接已关闭");
    }
    
    public override async Task OpenAsync()
    {
        // 模拟连接打开
        await Task.Delay(100);
        _state = ConnectionState.Open;
        Console.WriteLine("SQL连接已打开");
    }
    
    public override DbCommand CreateCommand()
    {
        return new SqlCommand();
    }
    
    public override void Dispose()
    {
        if (_state == ConnectionState.Open)
        {
            Close();
        }
        Console.WriteLine("SQL连接已释放");
    }
}

public class SqlCommand : DbCommand
{
    public override string CommandText { get; set; }
    public override IDataParameterCollection Parameters { get; } = new SqlParameterCollection();
    
    public override async Task<int> ExecuteNonQueryAsync()
    {   
        // 模拟执行SQL
        await Task.Delay(50);
        Console.WriteLine($"执行SQL: {CommandText}");
        return 1;
    }
    
    public override DbParameter CreateParameter()
    {
        return new SqlParameter();
    }
}

public class SqlParameter : DbParameter
{
    public override string ParameterName { get; set; }
    public override object Value { get; set; }
}

public class SqlParameterCollection : IDataParameterCollection
{
    // 简化实现，实际应完整实现接口
    // ...
}

高级并发模式示例

using Cyclops.Threading.Modes;
using Microsoft.Extensions.DependencyInjection;

// 高级并发模式实现
public class ConcurrentPatternExample
{
    private readonly IReaderWriterLockSlim _rwLock;
    private readonly IAsyncCoordinator _asyncCoordinator;
    private readonly ITimeoutPolicy _timeoutPolicy;
    
    private Dictionary<string, string> _sharedData = new();
    
    public ConcurrentPatternExample(
        IReaderWriterLockFactory readerWriterLockFactory,
        IAsyncCoordinatorFactory asyncCoordinatorFactory,
        ITimeoutPolicyFactory timeoutPolicyFactory)
    {
        // 创建读写锁
        _rwLock = readerWriterLockFactory.Create();
        
        // 创建异步协调器
        _asyncCoordinator = asyncCoordinatorFactory.Create("DataProcessingCoordinator");
        
        // 创建超时策略
        _timeoutPolicy = timeoutPolicyFactory.Create(TimeSpan.FromSeconds(30), 
            TimeSpan.FromSeconds(5)); // 操作超时30秒，心跳间隔5秒
    }
    
    // 读写锁模式
    public void ReaderWriterPattern()
    {
        // 读取操作 - 获取读锁
        _rwLock.EnterReadLock();
        try
        {
            if (_sharedData.TryGetValue("config", out var configValue))
            {
                Console.WriteLine($"读取配置: {configValue}");
            }
        }
        finally
        {
            _rwLock.ExitReadLock();
        }
        
        // 写入操作 - 获取写锁
        _rwLock.EnterWriteLock();
        try
        {
            _sharedData["config"] = DateTime.Now.ToString();
            Console.WriteLine("更新配置完成");
        }
        finally
        {
            _rwLock.ExitWriteLock();
        }
        
        // 升级锁模式
        _rwLock.EnterUpgradeableReadLock();
        try
        {
            if (!_sharedData.ContainsKey("featureFlag"))
            {
                // 需要时升级为写锁
                _rwLock.EnterWriteLock();
                try
                {
                    // 双重检查锁定模式
                    if (!_sharedData.ContainsKey("featureFlag"))
                    {
                        _sharedData["featureFlag"] = "true";
                        Console.WriteLine("添加功能标志");
                    }
                }
                finally
                {
                    _rwLock.ExitWriteLock();
                }
            }
            
            // 继续读取
            Console.WriteLine($"功能标志: {_sharedData["featureFlag"]}");
        }
        finally
        {
            _rwLock.ExitUpgradeableReadLock();
        }
    }
    
    // 异步协调模式
    public async Task AsyncCoordinationPatternAsync()
    {
        // 注册多个并行操作
        var operation1 = _asyncCoordinator.RegisterOperation(async (token) => {
            Console.WriteLine("开始操作1");
            await Task.Delay(3000, token);
            Console.WriteLine("完成操作1");
            return "结果1";
        });
        
        var operation2 = _asyncCoordinator.RegisterOperation(async (token) => {
            Console.WriteLine("开始操作2");
            await Task.Delay(2000, token);
            Console.WriteLine("完成操作2");
            return "结果2";
        });
        
        var operation3 = _asyncCoordinator.RegisterOperation(async (token) => {
            Console.WriteLine("开始操作3");
            await Task.Delay(4000, token);
            Console.WriteLine("完成操作3");
            return "结果3";
        });
        
        // 设置进度回调
        _asyncCoordinator.ProgressCallback = (progress) => {
            Console.WriteLine($"操作进度: {progress.CompletedOperations}/{progress.TotalOperations}, " +
                            $"完成百分比: {progress.CompletionPercentage:P2}");
        };
        
        // 等待所有操作完成
        var results = await _asyncCoordinator.WaitForAllOperationsAsync();
        
        Console.WriteLine("所有操作完成");
        foreach (var result in results)
        {
            Console.WriteLine($"操作结果: {result}");
        }
    }
    
    // 超时策略模式
    public async Task TimeoutPatternAsync()
    {
        try
        {
            // 使用超时策略执行操作
            var result = await _timeoutPolicy.ExecuteWithTimeoutAsync(async () => {
                // 模拟可能超时的操作
                await Task.Delay(25000); // 25秒，小于30秒超时
                return "操作完成";
            });
            
            Console.WriteLine($"操作结果: {result}");
        }
        catch (TimeoutException)
        {
            Console.WriteLine("操作执行超时");
        }
        
        // 带心跳的长时间操作
        try
        {
            var longRunningResult = await _timeoutPolicy.ExecuteWithHeartbeatAsync(
                async (heartbeat) => {
                    for (int i = 0; i < 10; i++)
                    {
                        // 定期发送心跳
                        heartbeat();
                        Console.WriteLine($"执行步骤 {i + 1}/10");
                        await Task.Delay(2000);
                    }
                    return "长时间操作完成";
                });
            
            Console.WriteLine($"长时间操作结果: {longRunningResult}");
        }
        catch (TimeoutException)
        {
            Console.WriteLine("带心跳的操作执行超时");
        }
    }
    
    // 演示所有模式
    public async Task DemonstrateAllPatternsAsync()
    {
        // 演示读写锁模式
        ReaderWriterPattern();
        
        // 演示异步协调模式
        await AsyncCoordinationPatternAsync();
        
        // 演示超时策略模式
        await TimeoutPatternAsync();
    }
}

贡献者

版本信息

• 当前版本
• 作者：yswenli
• 描述：企服版框架中线程同步与并发控制组件

贡献者

• yswenli

许可证

保留所有权利