Как читать / писать из глобальной переменной в основном потоке

Question

У меня есть приложение C # Windows IoT Background, которое я создал.Это приложение имеет несколько потоков в ThreadPool, которые работают бесконечно.

Эти потоки должны иметь возможность читать / записывать глобальные переменные в основном потоке, но я не уверен, как этого добиться.Вот пример того, что я пытаюсь сделать:

// main task
public sealed class StartupTask : IBackgroundTask
{
    private static BackgroundTaskDeferral _Deferral = null;

    private static MyThreadClass1 thread1 = null;
    private static MyThreadClass2 thread2 = null;
    private static MyThreadClass3 thread3 = null;

    List<Object> MyDevices = null;

    public async void Run(IBackgroundTaskInstance taskInstance)
    {
        _Deferral = taskInstance.GetDeferral();

        MyDevices = GetDeviceList();

        thread1 = new MyThreadClass1();
        await ThreadPool.RunAsync(workItem =>
        {
            thread1.Start();
        });

        thread2 = new MyThreadClass2();
        await ThreadPool.RunAsync(workItem =>
        {
            thread2.Start();
        });

        thread3 = new MyThreadClass3();
        await ThreadPool.RunAsync(workItem =>
        {
            thread3.Start();
        });
    }
}

internal class MyThreadClass1
{
    public async void Start()
    { }
}

internal class MyThreadClass2
{
    public async void Start()
    { }
}

internal class MyThreadClass3
{
    public async void Start()
    { }
}

В любом из трех работающих потоков мне нужно иметь возможность читать и писать в List<Object> MyDevices.

Все потоки имеют разные функции, но все они взаимодействуют с "MyDevices", поэтому, если один поток вносит изменения в этот список, другие потоки должны знать об этом изменении сразу.

Что лучшеспособ сделать это?

Спасибо!

Answer 1

Эти потоки должны иметь возможность чтения / записи в глобальные переменные в основном потоке

Самый простой способ справиться с этим требованием - удалить его.Можно ли кодировать решение так, чтобы каждый поток имел устройство?Или возможно переосмыслить обязанности потока, чтобы он общался путем передачи сообщений вместо обновления общих данных?Обычно эти альтернативные подходы приводят к гораздо более чистому и менее глючному коду.Но не всегда.

Вам понадобятся блокировки для защиты общих данных.Самый простой способ сделать это с помощью оператора lock, например:

object _mutex = new object();
List<Object> MyDevices = null;

...

var device = ...;
lock (_mutex)
{
  MyDevices.Add(device);
}

Как правило, вы хотите минимизировать код в операторе lock.Кроме того, вы можете захотеть иметь одну блокировку для List<Object> и отдельную блокировку для каждого элемента в списке, в зависимости от того, как ваш поток использует эти устройства.

Answer 2

другие потоки должны знать об изменениях сразу же

Если вы хотите получать уведомления с низкой задержкой, потоки должны тратить большую часть времени на сон.Например, выполнение Dispatcher.Run(), которое будет спать в ожидании обработки сообщений / задач.

Если это ваш случай, вы можете использовать ObservableCollection вместо List и написать обработчик CollectionChanged, который пересылает уведомления для ваших 3 потоков,Или, если вы этого хотите, перенаправьте уведомления в 2 других потока, исключая текущий, если вы не хотите, чтобы поток, инициировавший изменение, обрабатывал измененное событие.

Я не уверен, что *Класс 1012 * доступен на платформе Windows IoT.Определенно не относится к ядру .NET.Даже если нет, доступны строительные блоки высокого уровня для их создания.Вот пример реализации, которая также реализует контекст синхронизации, очень простой, потому что использует универсальные классы высокого уровня ConcurrentQueue и BlockingCollection.

using kvp = KeyValuePair<SendOrPostCallback, object>;

enum eShutdownReason : byte
{
    Completed,
    Failed,
    Unexpected,
}

class Dispatcher : IDisposable
{
    const int maxQueueLength = 100;

    readonly ConcurrentQueue<kvp> m_queue;
    readonly BlockingCollection<kvp> m_block;

    public Dispatcher()
    {
        m_queue = new ConcurrentQueue<kvp>();
        m_block = new BlockingCollection<kvp>( m_queue, maxQueueLength );
        createdThreadId = Thread.CurrentThread.ManagedThreadId;
        prevContext = SynchronizationContext.Current;
        SynchronizationContext.SetSynchronizationContext( new SyncContext( this ) );
    }

    readonly SynchronizationContext prevContext;
    readonly int createdThreadId;

    class SyncContext : SynchronizationContext
    {
        readonly Dispatcher dispatcher;

        public SyncContext( Dispatcher dispatcher )
        {
            this.dispatcher = dispatcher;
        }

        // https://blogs.msdn.microsoft.com/pfxteam/2012/01/20/await-synchronizationcontext-and-console-apps/
        public override void Post( SendOrPostCallback cb, object state )
        {
            dispatcher.Post( cb, state );
        }
    }

    /// <summary>Run the dispatcher. Must be called on the same thread that constructed the object.</summary>
    public eShutdownReason Run()
    {
        Debug.Assert( Thread.CurrentThread.ManagedThreadId == createdThreadId );

        while( true )
        {
            kvp h;
            try
            {
                h = m_block.Take();
            }
            catch( Exception ex )
            {
                ex.logError( "Dispatcher crashed" );
                return eShutdownReason.Unexpected;
            }
            if( null == h.Key )
                return (eShutdownReason)h.Value;

            try
            {
                h.Key( h.Value );
            }
            catch( Exception ex )
            {
                ex.logError( "Exception in Dispatcher.Run" );
            }
        }
    }

    /// <summary>Signal dispatcher to shut down. Can be called from any thread.</summary>
    public void Stop( eShutdownReason why )
    {
        Logger.Info( "Shutting down, because {0}", why );
        Post( null, why );
    }

    /// <summary>Post a callback to the queue. Can be called from any thread.</summary>
    public void Post( SendOrPostCallback cb, object state = null )
    {
        if( !m_block.TryAdd( new kvp( cb, state ) ) )
            throw new ApplicationException( "Unable to post a callback to the dispatcher: the dispatcher queue is full" );
    }

    void IDisposable.Dispose()
    {
        Debug.Assert( Thread.CurrentThread.ManagedThreadId == createdThreadId );
        SynchronizationContext.SetSynchronizationContext( prevContext );
    }
}

Независимо от того, будете ли вы использовать встроенный Dispatcher или мой пользовательский, всепотоки должны вызывать метод Run, а затем использовать асинхронные опубликованные задачи или асинхронные методы для запуска кода в диспетчере.

Answer 3

Одна вещь, которую вы, возможно, захотите рассмотреть, это ObservableCollection.Этот класс реализует интерфейс INotifyPropertyChanged, который уведомляет всех слушателей об изменениях в базовой коллекции.

Далее вы захотите реализовать обработчик событий для PropertyChanged в ваших Thread классах, например, так (Я рекомендую создать интерфейс или базовый класс, который обрабатывает это, поскольку вы, похоже, используете разные классы для каждого Thread):

public sealed class MyThreadBase
{
    private ObservableCollection<object> MyDevices;

    public MyThreadBase(ObservableCollection<object> deviceList)
    {
        MyDevices = deviceList;
        MyDevices.PropertyChanged += MyDevices_PropertyChanged; // Register listener
    }

    private void MyDevices_PropertyChanged(object sender, PropertyChangedEventArgs e)
    {
        lock (MyDevices)
        {
            // Do something with the data...
        }
    }
}

Оператор lock используется так, что поток блокируется, когдадругой поток читает или пишет в MyDevices.Это обычно важно для синхронизации и известно как проблема чтения-записи .Я бы посоветовал прочитать об этом и о возможных решениях.

Если вы намереваетесь, однако, для каждого потока перебрать устройства и что-то сделать с каждым, то вы столкнетесь с проблемами, как итерацияизменение коллекции не является хорошей идеей (и при использовании цикла foreach фактически вызовет исключение), так что имейте это в виду.