带有序列ID的螺纹安全固定尺寸的圆形缓冲液

Question

我需要一个具有这些功能的队列：

• 固定尺寸（IE圆形缓冲区）
• 队列项目具有ID（例如主键），它们是顺序的
• 线程安全（来自多个ASP.NET核心请求），

以避免锁定，i a concurrentqueue ，但找到了种族条件。因此，我正在尝试一种自定义方法。

public interface IQueueItem
{
    long Id { get; set; }
}

public class CircularBuffer<T> : LinkedList<T> where T : class, IQueueItem
{
    public CircularBuffer(int capacity) => _capacity = capacity;
    private readonly int _capacity;

    private long _counter = 0;
    private readonly object _lock = new();

    public void Enqueue(T item)
    {
        lock (_lock) {         // works but feels "heavy"
            _counter++;
            item.Id = _counter;
            if (Count == _capacity) RemoveFirst();
            AddLast(item);
        }
    }
}

并测试：

public class Item : IQueueItem
{
    public long Id { get; set; }
    //...
}

public class Program
{
    public static void Main()
    {
        var q = new CircularBuffer<Item>(10);
        Parallel.For(0, 15, i => q.Enqueue(new Item()));
        Console.WriteLine(string.Join(", ", q.Select(x => x.Id)));
    }
}

它提供了正确的输出（即使竞争线程逐渐订购，并且具有固定尺寸，并且最古老的物品均列出）：

6、7、8、9、10、11、12、13、14、15

实际上，我有读（即枚举）排队的网络请求。

问题：如果一个线程在另一个线程添加时枚举队列，我将有错误。 （我可以在读取之前使用 tolist（），但是对于大型队列会吸收所有服务器的内存，因为这可以通过多个请求每秒进行多次完成）。我该如何处理这种情况？ 我使用了一个链接列表，但是我可以灵活地使用任何结构。

（另外，这似乎是一个很重的锁定部分；是否有更性能的方法？）

更新
如下所述：我希望队列从几百到几千个项目中有几个项目，但是项目本身很小（只有几种原始数据类型）。我希望每秒都有一个。从Web请求读取的频率较低，假设每分钟几次（但可以同时发生到服务器编写到队列）。

Answer 1

根据您在问题中提供的指标，您有很多选择。预期的 cindularBuffer＆lt; t＆gt; 并不是很重。包装 lock -pretected 队列＆lt; t＆gt; 应该很好地工作。将队列内容复制成每个枚举的阵列的成本（每分钟几次复制10,000个元素）是不可能的。现代机器可以眨眼间做这些事情。您必须列举每秒数百次的收藏，才能开始（稍微）成为一个问题。

在我的原始答案中（修订版3 ）我建议使用 immutablequeue＆lt; t＆gt; 作为基础存储。经过仔细检查，我注意到这堂课并不是完全付费的。首次枚举它时，它调用内部向后转换 property（锁定 - 保护 queue＆lt; t＆gt; 更糟糕的解决方案。关于CPU时间和分配，A/72809621/11178549“> lonix的答案。

以下是类似想法的低级实现，它利用了这样一个事实，即我们只需要 Immutablequeue＆lt; t＆gt; 类的功能的一个子集。这些项目存储在单一链接的列表结构中，可以列举无需额外费用：

public class ConcurrentCircularBuffer<T> : IEnumerable<T>
{
    private readonly object _locker = new();
    private readonly int _capacity;
    private Node _head;
    private Node _tail;
    private int _count = 0;

    private class Node
    {
        public readonly T Item;
        public Node Next;
        public Node(T item) => Item = item;
    }

    public ConcurrentCircularBuffer(int capacity)
    {
        if (capacity < 1) throw new ArgumentOutOfRangeException(nameof(capacity));
        _capacity = capacity;
    }

    public int Count => Volatile.Read(ref _count);

    public void Enqueue(T item)
    {
        Node node = new(item);
        lock (_locker)
        {
            if (_head is null) _head = node;
            if (_tail is not null) _tail.Next = node;
            _tail = node;
            if (_count < _capacity) _count++; else _head = _head.Next;
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        Node node; int count;
        lock (_locker) { node = _head; count = _count; }
        for (int i = 0; i < count && node is not null; i++, node = node.Next)
            yield return node.Item;
    }

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}

这种方法的主要优势比 lock pretected queue＆lt; t＆gt; is它最小化了争论。握住锁时完成的工作很小。

可以在类的替代实现。 72802942/5“>第五修订此答案。

Answer 2

这是使用锁定的 queue＆lt; t＆gt; 的另一个实现。

public interface IQueueItem
{
    long Id { get; set; }
}

public class CircularBuffer<T> : IEnumerable<T> where T : class, IQueueItem
{
    private readonly int _capacity;
    private readonly Queue<T> _queue;
    private long _lastId = 0;
    private readonly object _lock = new();

    public CircularBuffer(int capacity) {
        _capacity = capacity;
        _queue = new Queue<T>(capacity);
    }

    public void Enqueue(T item)
    {
        lock (_lock) {
            if (_capacity < _queue.Count)
                _queue.Dequeue();
            item.Id = ++_lastId;
            _queue.Enqueue(item);
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        lock (_lock) {
            var copy = _queue.ToArray();
            return ((IEnumerable<T>)copy).GetEnumerator();
        }
    }

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();

}

和测试：

public class Item : IQueueItem
{
    private long _id;

    public long Id
    {
        get => Volatile.Read(ref _id);
        set => Volatile.Write(ref _id, value);
    }
}

public class Program
{
    public static void Main()
    {
        var q = new CircularBuffer<Item>(10);
        Parallel.For(0, 15, i => q.Enqueue(new Item()));
        Console.WriteLine(string.Join(", ", q.Select(x => x.Id)));
    }
}

结果：

6、7、8、9、10、11、12、13、14、15

Answer 3

由于此问题中有同时性的情况，因此您可以尝试固定的数组。

IQueueItem[] items = new IQueueItem[SIZE];
long id = 0;

入口很简单。

void Enqueue(IQueueItem item)
{
    long id2 = Interlocked.Increment(ref id);
    item.Id = id2 - 1;
    items[id2 % SIZE] = item;
}

要输出数据，您只需要将数组复制到新数组，然后对其进行排序。 （当然，可以在这里进行优化）

var arr = new IQueueItem[SIZE];
Array.Copy(items, arr, SIZE);
return arr.Where(a => a != null).OrderBy(a => a.Id);

由于并发插入，该数组中可能存在一些差距，您可以采取序列直到找到差距。

var e = arr.Where(a => a != null).OrderBy(a => a.Id);
var firstId = e.First().Id;
return e.TakeWhile((a, index) => a.Id - index == firstId);