Я хочу реализовать List<T> как свойство, которое можно без проблем использовать без проблем.
Что-то вроде этого:
private List<T> _list;
private List<T> MyT
{
get { // return a copy of _list; }
set { _list = value; }
}
Мне кажется, мне нужно вернуть копию (клонированную) из коллекции, поэтому, если где-то мы итерируем коллекцию и в то же время собираем коллекцию, то исключение не возникает.
Как реализовать свойство потокобезопасной коллекции?
Если вы нацеливаетесь на .Net 4, в System.Collections.Concurrent Пространство имен
В этом случае вы можете использовать ConcurrentBag<T> вместо List<T>
Даже если у него получилось больше всего голосов, обычно нельзя взять System.Collections.Concurrent.ConcurrentBag<T> как замену потокобезопасности для System.Collections.Generic.List<T>, как он есть (Radek Stromský уже указал на это), не упорядоченный.
Но есть класс под названием System.Collections.Generic.SynchronizedCollection<T>, который уже с .NET Framework является частью фреймворка, но он хорошо скрыт в месте, где никто не ожидает, что он мало известен и, вероятно, никогда не был наткнулся на него (по крайней мере, я этого не делал).
SynchronizedCollection<T> скомпилирован в сборку System.ServiceModel.dll (которая является частью профиля клиента, но не переносимой библиотеки классов).
Надеюсь, что это поможет.
Я бы подумал, что создать образец класса ThreadSafeList будет легко:
public class ThreadSafeList<T> : IList<T>
{
protected List<T> _interalList = new List<T>();
// Other Elements of IList implementation
public IEnumerator<T> GetEnumerator()
{
return Clone().GetEnumerator();
}
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
{
return Clone().GetEnumerator();
}
protected static object _lock = new object();
public List<T> Clone()
{
List<T> newList = new List<T>();
lock (_lock)
{
_interalList.ForEach(x => newList.Add(x));
}
return newList;
}
}
Вы просто клонируете список перед запросом перечислителя, и, таким образом, всякое перечисление завершает работу с копией, которая не может быть изменена во время работы.
Даже принятый ответ - ConcurrentBag, я не думаю, что это реальная замена списка во всех случаях, поскольку комментарий Radek к ответу говорит: "ConcurrentBag - неупорядоченная коллекция, поэтому в отличие от List она не гарантирует упорядочение. Также вы не можете получить доступ к элементам по индексу ".
Поэтому, если вы используете .NET 4.0 или выше, можно обойти это путем использования ConcurrentDictionary с целочисленным TKey в качестве индекса массива и TValue в качестве значения массива. Это рекомендуемый способ замены списка в курсе " Параллельные коллекции Pluralsight С#". ConcurrentDictionary решает обе проблемы, упомянутые выше: доступ к индексу и упорядочение (мы не можем полагаться на упорядочение, поскольку это хеш-таблица под капотом, но текущая реализация .NET сохраняет порядок добавления элементов).
Класс С# ArrayList имеет метод Synchronized.
var threadSafeArrayList = ArrayList.Synchronized(new ArrayList());
Это возвращает потокобезопасную оболочку вокруг любого экземпляра IList. Все операции должны выполняться через обертку для обеспечения безопасности резьбы.
Если вы посмотрите на исходный код для списка T (https://referencesource.microsoft.com/#mscorlib/system/collections/generic/list.cs,c66df6f36c131877), вы заметите, что там есть класс (который, конечно, внутренняя - почему, Microsoft, почему?!?!) называется SynchronizedList of T. Я скопирую код здесь:
[Serializable()]
internal class SynchronizedList : IList<T> {
private List<T> _list;
private Object _root;
internal SynchronizedList(List<T> list) {
_list = list;
_root = ((System.Collections.ICollection)list).SyncRoot;
}
public int Count {
get {
lock (_root) {
return _list.Count;
}
}
}
public bool IsReadOnly {
get {
return ((ICollection<T>)_list).IsReadOnly;
}
}
public void Add(T item) {
lock (_root) {
_list.Add(item);
}
}
public void Clear() {
lock (_root) {
_list.Clear();
}
}
public bool Contains(T item) {
lock (_root) {
return _list.Contains(item);
}
}
public void CopyTo(T[] array, int arrayIndex) {
lock (_root) {
_list.CopyTo(array, arrayIndex);
}
}
public bool Remove(T item) {
lock (_root) {
return _list.Remove(item);
}
}
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() {
lock (_root) {
return _list.GetEnumerator();
}
}
IEnumerator<T> IEnumerable<T>.GetEnumerator() {
lock (_root) {
return ((IEnumerable<T>)_list).GetEnumerator();
}
}
public T this[int index] {
get {
lock(_root) {
return _list[index];
}
}
set {
lock(_root) {
_list[index] = value;
}
}
}
public int IndexOf(T item) {
lock (_root) {
return _list.IndexOf(item);
}
}
public void Insert(int index, T item) {
lock (_root) {
_list.Insert(index, item);
}
}
public void RemoveAt(int index) {
lock (_root) {
_list.RemoveAt(index);
}
}
}
Лично я думаю, что они знали, что лучшая реализация с использованием SemaphoreSlim может быть создана, но не достигла этого.
Вы также можете использовать более примитивный
Monitor.Enter(lock);
Monitor.Exit(lock);
который использует блокировка (см. этот пост С# Блокировка объекта, переназначенного в блоке блокировки).
Если вы ожидаете исключений в коде, это небезопасно, но вы можете сделать что-то вроде следующего:
using System;
using System.Collections.Generic;
using System.Threading;
using System.Linq;
public class Something
{
private readonly object _lock;
private readonly List<string> _contents;
public Something()
{
_lock = new object();
_contents = new List<string>();
}
public Modifier StartModifying()
{
return new Modifier(this);
}
public class Modifier : IDisposable
{
private readonly Something _thing;
public Modifier(Something thing)
{
_thing = thing;
Monitor.Enter(Lock);
}
public void OneOfLotsOfDifferentOperations(string input)
{
DoSomethingWith(input);
}
private void DoSomethingWith(string input)
{
Contents.Add(input);
}
private List<string> Contents
{
get { return _thing._contents; }
}
private object Lock
{
get { return _thing._lock; }
}
public void Dispose()
{
Monitor.Exit(Lock);
}
}
}
public class Caller
{
public void Use(Something thing)
{
using (var modifier = thing.StartModifying())
{
modifier.OneOfLotsOfDifferentOperations("A");
modifier.OneOfLotsOfDifferentOperations("B");
modifier.OneOfLotsOfDifferentOperations("A");
modifier.OneOfLotsOfDifferentOperations("A");
modifier.OneOfLotsOfDifferentOperations("A");
}
}
}
Одна из приятных вещей об этом - вы получите блокировку для продолжительности серии операций (а не запирания в каждой операции). Это означает, что вывод должен появляться в правильных фрагментах (мое использование этого получило некоторый вывод на экран из внешнего процесса)
Мне действительно нравится простота + прозрачность ThreadSafeList +, которая делает важный бит при остановке сбоев
Я верю, что _list.ToList() сделает вам копию. Вы также можете запросить его, если вам нужно:
_list.Select("query here").ToList();
В любом случае, msdn говорит, что это действительно копия, а не просто ссылка. О, и да, вам нужно будет заблокировать установленный метод, как указали другие.
Похоже, что многие из тех, кто находит это, хотят получить индексируемую динамически измеряемую коллекцию с безопасным потоком. Самая близкая и простая вещь, о которой я знаю, была бы.
System.Collections.Concurrent.ConcurrentDictionary<int, YourDataType>
Это потребует от вас убедиться, что ваш ключ правильно инкриминируется, если вы хотите нормального поведения при индексировании. Если вы будете осторожны, вам может быть достаточно .count в качестве ключа для любых новых пар ключ-значение, которые вы добавляете.
Я бы предложил всем, кто имеет дело со List<T> в многопоточных сценариях, взглянуть на неизменяемые коллекции, в частности на ImmutableArray.
Я нашел это очень полезным, когда у вас есть:
Также может быть полезно, когда вам нужно реализовать какое-либо поведение, подобное транзакции (т.е. отменить операцию вставки/обновления/удаления в случае сбоя)
Вот класс, который вы просили:
namespace AI.Collections {
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
/// <summary>
/// Just a simple thread safe collection.
/// </summary>
/// <typeparam name="T"></typeparam>
/// <value>Version 1.5</value>
/// <remarks>TODO replace locks with AsyncLocks</remarks>
[DataContract( IsReference = true )]
public class ThreadSafeList<T> : IList<T> {
/// <summary>
/// TODO replace the locks with a ReaderWriterLockSlim
/// </summary>
[DataMember]
private readonly List<T> _items = new List<T>();
public ThreadSafeList( IEnumerable<T> items = null ) { this.Add( items ); }
public long LongCount {
get {
lock ( this._items ) {
return this._items.LongCount();
}
}
}
public IEnumerator<T> GetEnumerator() { return this.Clone().GetEnumerator(); }
IEnumerator IEnumerable.GetEnumerator() { return this.GetEnumerator(); }
public void Add( T item ) {
if ( Equals( default( T ), item ) ) {
return;
}
lock ( this._items ) {
this._items.Add( item );
}
}
public Boolean TryAdd( T item ) {
try {
if ( Equals( default( T ), item ) ) {
return false;
}
lock ( this._items ) {
this._items.Add( item );
return true;
}
}
catch ( NullReferenceException ) { }
catch ( ObjectDisposedException ) { }
catch ( ArgumentNullException ) { }
catch ( ArgumentOutOfRangeException ) { }
catch ( ArgumentException ) { }
return false;
}
public void Clear() {
lock ( this._items ) {
this._items.Clear();
}
}
public bool Contains( T item ) {
lock ( this._items ) {
return this._items.Contains( item );
}
}
public void CopyTo( T[] array, int arrayIndex ) {
lock ( this._items ) {
this._items.CopyTo( array, arrayIndex );
}
}
public bool Remove( T item ) {
lock ( this._items ) {
return this._items.Remove( item );
}
}
public int Count {
get {
lock ( this._items ) {
return this._items.Count;
}
}
}
public bool IsReadOnly { get { return false; } }
public int IndexOf( T item ) {
lock ( this._items ) {
return this._items.IndexOf( item );
}
}
public void Insert( int index, T item ) {
lock ( this._items ) {
this._items.Insert( index, item );
}
}
public void RemoveAt( int index ) {
lock ( this._items ) {
this._items.RemoveAt( index );
}
}
public T this[ int index ] {
get {
lock ( this._items ) {
return this._items[ index ];
}
}
set {
lock ( this._items ) {
this._items[ index ] = value;
}
}
}
/// <summary>
/// Add in an enumerable of items.
/// </summary>
/// <param name="collection"></param>
/// <param name="asParallel"></param>
public void Add( IEnumerable<T> collection, Boolean asParallel = true ) {
if ( collection == null ) {
return;
}
lock ( this._items ) {
this._items.AddRange( asParallel
? collection.AsParallel().Where( arg => !Equals( default( T ), arg ) )
: collection.Where( arg => !Equals( default( T ), arg ) ) );
}
}
public Task AddAsync( T item ) {
return Task.Factory.StartNew( () => { this.TryAdd( item ); } );
}
/// <summary>
/// Add in an enumerable of items.
/// </summary>
/// <param name="collection"></param>
public Task AddAsync( IEnumerable<T> collection ) {
if ( collection == null ) {
throw new ArgumentNullException( "collection" );
}
var produce = new TransformBlock<T, T>( item => item, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount } );
var consume = new ActionBlock<T>( action: async obj => await this.AddAsync( obj ), dataflowBlockOptions: new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount } );
produce.LinkTo( consume );
return Task.Factory.StartNew( async () => {
collection.AsParallel().ForAll( item => produce.SendAsync( item ) );
produce.Complete();
await consume.Completion;
} );
}
/// <summary>
/// Returns a new copy of all items in the <see cref="List{T}" />.
/// </summary>
/// <returns></returns>
public List<T> Clone( Boolean asParallel = true ) {
lock ( this._items ) {
return asParallel
? new List<T>( this._items.AsParallel() )
: new List<T>( this._items );
}
}
/// <summary>
/// Perform the <paramref name="action" /> on each item in the list.
/// </summary>
/// <param name="action">
/// <paramref name="action" /> to perform on each item.
/// </param>
/// <param name="performActionOnClones">
/// If true, the <paramref name="action" /> will be performed on a <see cref="Clone" /> of the items.
/// </param>
/// <param name="asParallel">
/// Use the <see cref="ParallelQuery{TSource}" /> method.
/// </param>
/// <param name="inParallel">
/// Use the
/// <see
/// cref="Parallel.ForEach{TSource}(System.Collections.Generic.IEnumerable{TSource},System.Action{TSource})" />
/// method.
/// </param>
public void ForEach( Action<T> action, Boolean performActionOnClones = true, Boolean asParallel = true, Boolean inParallel = false ) {
if ( action == null ) {
throw new ArgumentNullException( "action" );
}
var wrapper = new Action<T>( obj => {
try {
action( obj );
}
catch ( ArgumentNullException ) {
//if a null gets into the list then swallow an ArgumentNullException so we can continue adding
}
} );
if ( performActionOnClones ) {
var clones = this.Clone( asParallel: asParallel );
if ( asParallel ) {
clones.AsParallel().ForAll( wrapper );
}
else if ( inParallel ) {
Parallel.ForEach( clones, wrapper );
}
else {
clones.ForEach( wrapper );
}
}
else {
lock ( this._items ) {
if ( asParallel ) {
this._items.AsParallel().ForAll( wrapper );
}
else if ( inParallel ) {
Parallel.ForEach( this._items, wrapper );
}
else {
this._items.ForEach( wrapper );
}
}
}
}
/// <summary>
/// Perform the <paramref name="action" /> on each item in the list.
/// </summary>
/// <param name="action">
/// <paramref name="action" /> to perform on each item.
/// </param>
/// <param name="performActionOnClones">
/// If true, the <paramref name="action" /> will be performed on a <see cref="Clone" /> of the items.
/// </param>
/// <param name="asParallel">
/// Use the <see cref="ParallelQuery{TSource}" /> method.
/// </param>
/// <param name="inParallel">
/// Use the
/// <see
/// cref="Parallel.ForEach{TSource}(System.Collections.Generic.IEnumerable{TSource},System.Action{TSource})" />
/// method.
/// </param>
public void ForAll( Action<T> action, Boolean performActionOnClones = true, Boolean asParallel = true, Boolean inParallel = false ) {
if ( action == null ) {
throw new ArgumentNullException( "action" );
}
var wrapper = new Action<T>( obj => {
try {
action( obj );
}
catch ( ArgumentNullException ) {
//if a null gets into the list then swallow an ArgumentNullException so we can continue adding
}
} );
if ( performActionOnClones ) {
var clones = this.Clone( asParallel: asParallel );
if ( asParallel ) {
clones.AsParallel().ForAll( wrapper );
}
else if ( inParallel ) {
Parallel.ForEach( clones, wrapper );
}
else {
clones.ForEach( wrapper );
}
}
else {
lock ( this._items ) {
if ( asParallel ) {
this._items.AsParallel().ForAll( wrapper );
}
else if ( inParallel ) {
Parallel.ForEach( this._items, wrapper );
}
else {
this._items.ForEach( wrapper );
}
}
}
}
}
}
В принципе, если вы хотите безопасно перечислить, вам нужно использовать блокировку.
Пожалуйста, обратитесь к MSDN. http://msdn.microsoft.com/en-us/library/6sh2ey19.aspx
Вот часть MSDN, которая может вас заинтересовать:
Элементы public static (Shared in Visual Basic) этого типа являются потокобезопасными. Любые члены экземпляра не гарантируют безопасность потоков.
Список может поддерживать несколько считывателей одновременно, пока коллекция не будет изменена. Перечисление через коллекцию по существу не является потокобезопасной процедурой. В редком случае, когда перечисление связано с одним или несколькими образами записи, единственным способом обеспечения безопасности потоков является блокировка коллекции во время всего перечисления. Чтобы обеспечить доступ к коллекции несколькими потоками для чтения и записи, вы должны реализовать свою собственную синхронизацию.
Используйте оператор lock для этого. (Читайте здесь для получения дополнительной информации.)
private List<T> _list;
private List<T> MyT
{
get { return _list; }
set
{
//Lock so only one thread can change the value at any given time.
lock (_list)
{
_list = value;
}
}
}
FYI это, вероятно, не совсем то, что вы спрашиваете - скорее всего, вы захотите заблокировать дальше в своем коде, но я не могу этого допустить. Посмотрите ключевое слово lock и настройте его использование в конкретной ситуации.
Если вам нужно, вы могли бы lock в блоке get и set, используя переменную _list, которая сделала бы это так, чтобы чтение/запись не могли возникать одновременно.