Как бы вы кодировали эффективный кольцевой буфер в Java или C#
Я хочу простой класс, который реализует кольцевой буфер фиксированного размера. Это должно быть эффективно, легко для глаз, типично напечатано.
РЕДАКТИРОВАТЬ: Это не должно быть MT-совместимым, на данный момент. Я всегда могу добавить блокировку позже, в любом случае это не будет высокий параллелизм.
Методы должны быть: .Добавить и, я думаю,.List, где я получаю все записи. Во-вторых, поиск, я думаю, должен осуществляться через индексатор. В любой момент я захочу получить любой элемент в буфере по индексу. Но имейте в виду, что от одного момента к другому Элемент [n] может отличаться, так как Круговой буфер заполняется и переворачивается.
Это не стек, это круговой буфер. Относительно "переполнения": я ожидал бы, что внутри будет массив, содержащий элементы, и со временем голова и хвост буфера будут вращаться вокруг этого фиксированного массива. Но это должно быть невидимым для пользователя. Не должно быть обнаруживаемого извне события или поведения "переполнения".
Это не школьное задание - чаще всего оно будет использоваться для кэша MRU или транзакции фиксированного размера или журнала событий.
13 ответов
Я хотел бы использовать массив T, указатель головы и хвоста, а также добавить и получить методы.
Нравится: (Охота за ошибками оставлена пользователю)
// Hijack these for simplicity
import java.nio.BufferOverflowException;
import java.nio.BufferUnderflowException;
public class CircularBuffer<T> {
private T[] buffer;
private int tail;
private int head;
@SuppressWarnings("unchecked")
public CircularBuffer(int n) {
buffer = (T[]) new Object[n];
tail = 0;
head = 0;
}
public void add(T toAdd) {
if (head != (tail - 1)) {
buffer[head++] = toAdd;
} else {
throw new BufferOverflowException();
}
head = head % buffer.length;
}
public T get() {
T t = null;
int adjTail = tail > head ? tail - buffer.length : tail;
if (adjTail < head) {
t = (T) buffer[tail++];
tail = tail % buffer.length;
} else {
throw new BufferUnderflowException();
}
return t;
}
public String toString() {
return "CircularBuffer(size=" + buffer.length + ", head=" + head + ", tail=" + tail + ")";
}
public static void main(String[] args) {
CircularBuffer<String> b = new CircularBuffer<String>(3);
for (int i = 0; i < 10; i++) {
System.out.println("Start: " + b);
b.add("One");
System.out.println("One: " + b);
b.add("Two");
System.out.println("Two: " + b);
System.out.println("Got '" + b.get() + "', now " + b);
b.add("Three");
System.out.println("Three: " + b);
// Test Overflow
// b.add("Four");
// System.out.println("Four: " + b);
System.out.println("Got '" + b.get() + "', now " + b);
System.out.println("Got '" + b.get() + "', now " + b);
// Test Underflow
// System.out.println("Got '" + b.get() + "', now " + b);
// Back to start, let's shift on one
b.add("Foo");
b.get();
}
}
}
Вот как я бы (или сделал) написать эффективный кольцевой буфер в Java. Он поддерживается простым массивом. Для моего конкретного случая использования мне требовалась высокая параллельная пропускная способность, поэтому я использовал CAS для выделения индекса. Затем я создал механизмы для надежных копий, включая копию CAS всего буфера. Я использовал это в случае, который изложен более подробно в короткой статье.
import java.util.concurrent.atomic.AtomicLong;
import java.lang.reflect.Array;
/**
* A circular array buffer with a copy-and-swap cursor.
*
* <p>This class provides an list of T objects who's size is <em>unstable</em>.
* It's intended for capturing data where the frequency of sampling greatly
* outweighs the frequency of inspection (for instance, monitoring).</p>
*
* <p>This object keeps in memory a fixed size buffer which is used for
* capturing objects. It copies the objects to a snapshot array which may be
* worked with. The size of the snapshot array will vary based on the
* stability of the array during the copy operation.</p>
*
* <p>Adding buffer to the buffer is <em>O(1)</em>, and lockless. Taking a
* stable copy of the sample is <em>O(n)</em>.</p>
*/
public class ConcurrentCircularBuffer <T> {
private final AtomicLong cursor = new AtomicLong();
private final T[] buffer;
private final Class<T> type;
/**
* Create a new concurrent circular buffer.
*
* @param type The type of the array. This is captured for the same reason
* it's required by {@link java.util.List.toArray()}.
*
* @param bufferSize The size of the buffer.
*
* @throws IllegalArgumentException if the bufferSize is a non-positive
* value.
*/
public ConcurrentCircularBuffer (final Class <T> type,
final int bufferSize)
{
if (bufferSize < 1) {
throw new IllegalArgumentException(
"Buffer size must be a positive value"
);
}
this.type = type;
this.buffer = (T[]) new Object [ bufferSize ];
}
/**
* Add a new object to this buffer.
*
* <p>Add a new object to the cursor-point of the buffer.</p>
*
* @param sample The object to add.
*/
public void add (T sample) {
buffer[(int) (cursor.getAndIncrement() % buffer.length)] = sample;
}
/**
* Return a stable snapshot of the buffer.
*
* <p>Capture a stable snapshot of the buffer as an array. The snapshot
* may not be the same length as the buffer, any objects which were
* unstable during the copy will be factored out.</p>
*
* @return An array snapshot of the buffer.
*/
public T[] snapshot () {
T[] snapshots = (T[]) new Object [ buffer.length ];
/* Determine the size of the snapshot by the number of affected
* records. Trim the size of the snapshot by the number of records
* which are considered to be unstable during the copy (the amount the
* cursor may have moved while the copy took place).
*
* If the cursor eliminated the sample (if the sample size is so small
* compared to the rate of mutation that it did a full-wrap during the
* copy) then just treat the buffer as though the cursor is
* buffer.length - 1 and it was not changed during copy (this is
* unlikley, but it should typically provide fairly stable results).
*/
long before = cursor.get();
/* If the cursor hasn't yet moved, skip the copying and simply return a
* zero-length array.
*/
if (before == 0) {
return (T[]) Array.newInstance(type, 0);
}
System.arraycopy(buffer, 0, snapshots, 0, buffer.length);
long after = cursor.get();
int size = buffer.length - (int) (after - before);
long snapshotCursor = before - 1;
/* Highly unlikely, but the entire buffer was replaced while we
* waited...so just return a zero length array, since we can't get a
* stable snapshot...
*/
if (size <= 0) {
return (T[]) Array.newInstance(type, 0);
}
long start = snapshotCursor - (size - 1);
long end = snapshotCursor;
if (snapshotCursor < snapshots.length) {
size = (int) snapshotCursor + 1;
start = 0;
}
/* Copy the sample snapshot to a new array the size of our stable
* snapshot area.
*/
T[] result = (T[]) Array.newInstance(type, size);
int startOfCopy = (int) (start % snapshots.length);
int endOfCopy = (int) (end % snapshots.length);
/* If the buffer space wraps the physical end of the array, use two
* copies to construct the new array.
*/
if (startOfCopy > endOfCopy) {
System.arraycopy(snapshots, startOfCopy,
result, 0,
snapshots.length - startOfCopy);
System.arraycopy(snapshots, 0,
result, (snapshots.length - startOfCopy),
endOfCopy + 1);
}
else {
/* Otherwise it's a single continuous segment, copy the whole thing
* into the result.
*/
System.arraycopy(snapshots, startOfCopy,
result, 0, endOfCopy - startOfCopy + 1);
}
return (T[]) result;
}
/**
* Get a stable snapshot of the complete buffer.
*
* <p>This operation fetches a snapshot of the buffer using the algorithm
* defined in {@link snapshot()}. If there was concurrent modification of
* the buffer during the copy, however, it will retry until a full stable
* snapshot of the buffer was acquired.</p>
*
* <p><em>Note, for very busy buffers on large symmetric multiprocessing
* machines and supercomputers running data processing intensive
* applications, this operation has the potential of being fairly
* expensive. In practice on commodity hardware, dualcore processors and
* non-processing intensive systems (such as web services) it very rarely
* retries.</em></p>
*
* @return A full copy of the internal buffer.
*/
public T[] completeSnapshot () {
T[] snapshot = snapshot();
/* Try again until we get a snapshot that's the same size as the
* buffer... This is very often a single iteration, but it depends on
* how busy the system is.
*/
while (snapshot.length != buffer.length) {
snapshot = snapshot();
}
return snapshot;
}
/**
* The size of this buffer.
*/
public int size () {
return buffer.length;
}
}
Вот готовая к использованию реализация CircularArrayList для Java, которую я использую в производственном коде. Переопределяя AbstractList рекомендованным Java способом, он поддерживает все функциональные возможности, которые можно ожидать от стандартной реализации List в Java Collections Framework (универсальный тип элемента, subList, итерация и т. Д.).
Следующие вызовы завершены в O(1):
- добавить (элемент) - добавляет в конец списка
- удалить (0) - удаляет из начала списка
- get(i) - возвращает случайный элемент в списке
Я бы использовал ArrayBlockingQueue или одну из других готовых реализаций Queue, в зависимости от потребностей. Очень редко необходимо реализовать такую структуру данных самостоятельно (если это не школьное задание).
РЕДАКТИРОВАТЬ: Теперь, когда вы добавили требование "извлекать любой элемент в буфере по индексу", я предполагаю, что вам нужно реализовать свой собственный класс (если только в google-collection или какой - либо другой библиотеке его нет). Циклический буфер довольно легко реализовать, как показывает пример JeeBee. Вы также можете посмотреть на исходный код ArrayBlockingQueue - его код достаточно чистый, просто удалите блокирующие и ненужные методы и добавьте методы для доступа к нему по индексу.
Вот реализация, которую я написал для собственного использования, но это может быть полезно.
Буфер содержит максимально фиксированный набор элементов. Набор круговой, старые предметы автоматически удаляются. Вызывающая сторона может получить элементы хвоста по абсолютному инкрементному индексу (long), но элементы могут быть потеряны между вызовами, слишком удаленными во времени. Этот класс полностью потокобезопасен.
public sealed class ConcurrentCircularBuffer<T> : ICollection<T>
{
private T[] _items;
private int _index;
private bool _full;
public ConcurrentCircularBuffer(int capacity)
{
if (capacity <= 1) // need at least two items
throw new ArgumentException(null, "capacity");
Capacity = capacity;
_items = new T[capacity];
}
public int Capacity { get; private set; }
public long TotalCount { get; private set; }
public int Count
{
get
{
lock (SyncObject) // full & _index need to be in sync
{
return _full ? Capacity : _index;
}
}
}
public void AddRange(IEnumerable<T> items)
{
if (items == null)
return;
lock (SyncObject)
{
foreach (var item in items)
{
AddWithLock(item);
}
}
}
private void AddWithLock(T item)
{
_items[_index] = item;
_index++;
if (_index == Capacity)
{
_full = true;
_index = 0;
}
TotalCount++;
}
public void Add(T item)
{
lock (SyncObject)
{
AddWithLock(item);
}
}
public void Clear()
{
lock (SyncObject)
{
_items = new T[Capacity];
_index = 0;
_full = false;
TotalCount = 0;
}
}
// this gives raw access to the underlying buffer. not sure I should keep that
public T this[int index]
{
get
{
return _items[index];
}
}
public T[] GetTail(long startIndex)
{
long lostCount;
return GetTail(startIndex, out lostCount);
}
public T[] GetTail(long startIndex, out long lostCount)
{
if (startIndex < 0 || startIndex >= TotalCount)
throw new ArgumentOutOfRangeException("startIndex");
T[] array = ToArray();
lostCount = (TotalCount - Count) - startIndex;
if (lostCount >= 0)
return array;
lostCount = 0;
// this maybe could optimized to not allocate the initial array
// but in multi-threading environment, I suppose this is arguable (and more difficult).
T[] chunk = new T[TotalCount - startIndex];
Array.Copy(array, array.Length - (TotalCount - startIndex), chunk, 0, chunk.Length);
return chunk;
}
public T[] ToArray()
{
lock (SyncObject)
{
T[] items = new T[_full ? Capacity : _index];
if (_full)
{
if (_index == 0)
{
Array.Copy(_items, items, Capacity);
}
else
{
Array.Copy(_items, _index, items, 0, Capacity - _index);
Array.Copy(_items, 0, items, Capacity - _index, _index);
}
}
else if (_index > 0)
{
Array.Copy(_items, items, _index);
}
return items;
}
}
public IEnumerator<T> GetEnumerator()
{
return ToArray().AsEnumerable().GetEnumerator();
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
bool ICollection<T>.Contains(T item)
{
return _items.Contains(item);
}
void ICollection<T>.CopyTo(T[] array, int arrayIndex)
{
if (array == null)
throw new ArgumentNullException("array");
if (array.Rank != 1)
throw new ArgumentException(null, "array");
if (arrayIndex < 0)
throw new ArgumentOutOfRangeException("arrayIndex");
if ((array.Length - arrayIndex) < Count)
throw new ArgumentException(null, "array");
T[] thisArray = ToArray();
Array.Copy(thisArray, 0, array, arrayIndex, thisArray.Length);
}
bool ICollection<T>.IsReadOnly
{
get
{
return false;
}
}
bool ICollection<T>.Remove(T item)
{
return false;
}
private static object _syncObject;
private static object SyncObject
{
get
{
if (_syncObject == null)
{
object obj = new object();
Interlocked.CompareExchange(ref _syncObject, obj, null);
}
return _syncObject;
}
}
}
Просто используйте чужую реализацию:
Коллекция Power Collections <T> реализуется кольцевым буфером.
Библиотека мощных коллекций неоднородна, но Deque - вполне приемлемый расширяющийся круговой буфер.
Поскольку вы указываете, что не хотите расширения и вместо этого хотите перезаписать, вы можете довольно легко изменить код для перезаписи. Это будет просто включать снятие проверки для логически смежных указателей и просто писать в любом случае. В то же время приватный буфер может быть сделан только для чтения.
System.Collections.Generic.Queue - простой круглый буфер внутри (T[] с головой и хвостом, как в примере из JeeBee).
Я хочу ответить на этот вопрос в перспективе Java.
Чтобы реализовать циклический буфер с Java, вам, вероятно, понадобятся три вещи, включая: класс циклического буфера, универсальный и несколько операций над ним (чтобы узнать, какие операции вам нужны, и внутренний механизм этих операций, вам может понадобиться прочитать вики для круговой буфер вначале).
Во-вторых, к суждению о полном или пустом буфере следует относиться очень осторожно. Здесь я даю два инстинктивных решения для полного / пустого суждения. В первом решении вам нужно создать два целочисленных варианта для хранения как текущего размера вашего буфера, так и максимального размера вашего буфера. Очевидно, что если текущий размер равен максимальному размеру, буфер заполнен.
В другом решении мы устанавливаем последнее место хранения в режиме ожидания (для кольцевого буфера размером семь мы устанавливаем хранилище равным семи в режиме ожидания). Согласно этому, мы можем определить, что буфер заполнен, когда выражение (rp+1)%MAXSIZE == fp; доволен.
Для большей ясности, здесь приведены примеры реализации с использованием языка Java.
import java.nio.BufferOverflowException;
import java.nio.BufferUnderflowException;
public class CircularBuffer<T> {
private int front;
private int rear;
private int currentSize;
private int maxSize;
private T[] buffer;
public CircularBuffer(int n) {
buffer = (T[]) new Object[n];
front = 0;
rear = 0;
currentSize = 0;
maxSize = n;
}
public void push(T e) {
if (!isFull()) {
buffer[rear] = e;
currentSize++;
rear = (rear + 1) % maxSize;
} else throw new BufferOverflowException();
}
public T pop() {
if (!isEmpty()) {
T temp = buffer[front];
buffer[front] = null;
front = (front + 1) % maxSize;
currentSize--;
return temp;
} else throw new BufferUnderflowException();
}
public T peekFirst() {
if (!isEmpty()) {
return buffer[front];
} else return null;
}
public T peekLast() {
if (!isEmpty()) {
return buffer[rear - 1];
} else return null;
}
public int size() {
return currentSize;
}
public boolean isEmpty() {
if (currentSize == 0) {
return true;
} else return false;
}
public boolean isFull() {
if (currentSize == maxSize) {
return true;
} else return false;
}
public boolean clean() {
front = 0;
rear = 0;
while (rear != 0) {
buffer[rear] = null;
rear = (rear + 1) % maxSize;
}
return true;
}
public static void main(String[] args) {
CircularBuffer<Integer> buff = new CircularBuffer<>(7);
buff.push(0);
buff.push(1);
buff.push(2);
System.out.println(buff.size());
System.out.println("The head element is: " + buff.pop());
System.out.println("Size should be twoo: " + buff.size());
System.out.println("The last element is one: " + buff.peekLast());
System.out.println("Size should be two: " + buff.size());
buff.clean();
System.out.println("Size should be zero: " + buff.size());
}
}
В гуаве 15 мы ввели EvictingQueue, которая представляет собой неблокирующую ограниченную очередь, которая автоматически вытесняет (удаляет) элементы из заголовка очереди при попытке добавить элементы в полную очередь. Это отличается от обычных ограниченных очередей, которые блокируют или отклоняют новые элементы при заполнении.
Похоже, это должно соответствовать вашим потребностям, и имеет гораздо более простой интерфейс, чем использование ArrayDeque напрямую (он использует один под капотом, хотя!).
Более подробную информацию можно найти здесь.
Если кэш lru будет работать, рассмотрите возможность использования http://java.sun.com/javase/6/docs/api/java/util/LinkedHashMap.html, http://java.sun.com/javase/6/docs/api/java/util/LinkedHashMap.html
Вот еще одна реализация, которая использует BoundedFifoBuffer общей коллекции Apache. пожалуйста, используйте CircularFifoQueue, если вы используете последнюю версию JAR от Apache, так как приведенный ниже класс устарел
BoundedFifoBuffer apiCallHistory = new BoundedFifoBuffer(20);
for(int i =1 ; i < 25; i++){
if(apiCallHistory.isFull()){
System.out.println("removing :: "+apiCallHistory.remove());
}
apiCallHistory.add(i);
}
// The following is in C#
public class fqueue
{
// The following code implements a circular queue of objects
//private data:
private bool empty;
private bool full;
private int begin, end;
private object[] x;
//public data:
public fqueue()
{
empty = !(full = false);
begin = end = 0xA2;
x = new object[256];
return;
}
public fqueue(int size)
{
if (1 > size) throw new Exception("fqueue: Size cannot be zero or negative");
empty = !(full = false);
begin = end = 0xA2;
x = new object[size];
return;
}
public object write
{
set
{
if(full) throw new Exception("Write error: Queue is full");
end = empty ? end : (end + 1) % x.Length;
full = ((end + 1) % x.Length) == begin;
empty = false;
x[end] = value;
}
}
public object read
{
get
{
if(empty) throw new Exception("Read error: Queue is empty");
full = false;
object ret = x[begin];
begin = (empty=end==begin) ?
begin :
(begin + 1) % x.Length;
return ret;
}
}
public int maxSize
{
get
{
return x.Length;
}
}
public int queueSize
{
get
{
return end - begin + (empty ? 0 : 1 + ((end < begin) ? x.Length : 0));
}
}
public bool isEmpty
{
get
{
return empty;
}
}
public bool isFull
{
get
{
return full;
}
}
public int start
{
get
{
return begin;
}
}
public int finish
{
get
{
return end;
}
}
}