C#——数据同步方法

傷城~ 2022-07-17 03:51 422阅读 0赞

## 序 ##

数据同步是并发编程不可避免的话题，今天我们就来谈谈C\#中的数据同步方法。

### 进程内数据同步 ###

数据主要通过同步原语来进行同步，先来看看C\#中的两种最简单的数据同步方式

#### （1）lock关键字 ####

lock关键字很简单，简单的解析就是一把锁，当有一个线程持有这把锁时，其他线程不能进入这个“锁”区域，C\#中表现这个区域为一个\{\}块，学术上称作临界区（critical  region）。

例子如下：

<pre name="code" class="csharp">using System;
    using System.Threading.Tasks;
    namespace Listing_06
    {
        class BankAccount
        {
            public int Balance
            {
                get;
                set;
            }
        }
    class Listing_06
    {
            static void Main(string[] args)
            {
                // create the bank account instance
                BankAccount account = new BankAccount();
                // create an array of tasks
                Task[] tasks = new Task[10];
                // create the lock object
                object lockObj = new object();
                for (int i = 0; i < 10; i++)
                {
                    // create a new task
                    tasks[i] = new Task(() => {
                        // enter a loop for 1000 balance updates
                        for (int j = 0; j < 1000; j++)
                        {
                            lock (lockObj)
                            {
                                // update the balance
                                account.Balance = account.Balance + 1;
                            }
                        }
                    });
                    // start the new task
                    tasks[i].Start();
                }
                // wait for all of the tasks to complete
                Task.WaitAll(tasks);
                // write out the counter value
                Console.WriteLine("Expected value {0}, Balance: {1}",
                10000, account.Balance);
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

lock后大括号\{\}内的区块是被锁定代码段，这个代码段在同一时刻只能被一个线程访问。有必要说明的是，\{\}中的代码与学术上的临界区概念一般是不等同的。

#### （2）关键字synchronization ####

这种方式的数据同步叫做声明式同步（Declarative Synchronization），用法如下：

using System;
    using System.Runtime.Remoting.Contexts;
    using System.Threading.Tasks;
    namespace Listing_14
    {
        [Synchronization]
        class BankAccount : ContextBoundObject
        {
            private int balance = 0;
            public void IncrementBalance()
            {
                balance++;
            }
            public int GetBalance()
            {
                return balance;
            }
        }
        class Listing_14
        {
            static void Main(string[] args)
            {
                // create the bank account instance
                BankAccount account = new BankAccount();
                // create an array of tasks
                Task[] tasks = new Task[10];
            for (int i = 0; i < 10; i++)
                {
                    // create a new task
                    tasks[i] = new Task(() => {
                        // enter a loop for 1000 balance updates
                        for (int j = 0; j < 1000; j++)
                        {
                            // update the balance
                            account.IncrementBalance();
                        }
                    });
                    // start the new task
                    tasks[i].Start();
                }
                // wait for all of the tasks to complete
                Task.WaitAll(tasks);
                // write out the counter value
                Console.WriteLine("Expected value {0}, Balance: {1}",
                10000, account.GetBalance());
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

怎么使用这种声明式同步呢？首先是要继承自ContextBoundObject类，然后引入命名空间using System.Runtime.Remoting.Contexts;，最后是对有需要进行数据同步的类使用\[Synchronization\]声明。

这种数据同步方式简单粗暴，简单是因为只需要做一个声明即可，粗暴是因为在使用该类中的所有数据成员和方法时，都会被锁定，要知道，有时候并不是类中所有的成员，所有的情形都需要进行数据同步的，这就可能是一个严重的性能问题了。

#### Interlocked类 ####

System.Threading.Interlocked类提供了一系列的静态方法来进行同步操作，它利用操作系统和硬件环境来提供高性能的数据同步。

![Center][]

Interlocked.Exchange()方法用于交换两个值，下面两个代码片段有相同的语义：

code snippet 1

int myval = 99;
    lock (lockObj) {
    // change value
    myval = 100;
    }

code snippet 2

// change value - synchronized
    Interlocked.Exchange(ref myval, 101);

除CompareExchange()方法外，其他方法见名知义，不再作特别的解析。下面来重点说一下CompareExchange()方法：

CompareExchange()的语义是检查一个值是否被改变，如果是，那么改变其变量的值。他的一个重要的用途是检测某个变量是否被另一线程所改变，如果是，改变其的值。用例如下：

using System;
    using System.Threading.Tasks;
    using System.Threading;
    
    namespace _InterLock
    {
        class Program
        {
            class BankAccount
            {
                public int Balance = 0;
            }
            static void Main(string[] args)
            {
                // create the bank account instance
                BankAccount account = new BankAccount();
                // create an array of tasks
                Task[] tasks = new Task[10];
                for (int i = 0; i < 10; i++)
                {
                    // create a new task
                    tasks[i] = new Task(() => {
                        // get a local copy of the shared data
                        int startBalance = account.Balance;
                        // create a local working copy of the shared data
                        int localBalance = startBalance;
                        // enter a loop for 1000 balance updates
                        for (int j = 0; j < 1000; j++)
                        {
                            // update the local balance
                            localBalance++;
                        }
                        // check to see if the shared data has changed since we started
                        // and if not, then update with our local value
                        int sharedData = Interlocked.CompareExchange(ref account.Balance, localBalance, startBalance);
                        if (sharedData == startBalance)
                        {
                            Console.WriteLine("Shared data updated OK");
                        }
                        else
                        {
                            Console.WriteLine("Shared data changed");
                      }
                    });
                    // start the new task
                    tasks[i].Start();
                }
                // wait for all of the tasks to complete
                Task.WaitAll(tasks);
                // write out the counter value
                Console.WriteLine("Expected value {0}, Balance: {1}", 10000, account.Balance);
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

上面的例子是一个多个Task将同一变量加到10000的过程，在每个Task内部，它首先将当前时刻的account.Balance保存两份，一份不变，一份用于循环加1直到加上1000。然后用那份不变的变量与此该的account.Balance进行比较，如果不相同，说明account.Balance已经被其他的Task更改了，这时交换account.Balance与localBalance的值，否则，保持原来的值。

#### Spinlock锁 ####

Spinlock中文名为旋转锁，旋转锁的特点是其他线程在等待期间并不投入等待，而是周期性地检测是否可获得这个锁，所以当有其他线程在等待这个锁的时候，CPU的占用将飚升，但当一个线程持有这个锁极短的情况下，这个锁的性能是非常好的。

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_10 {
    class BankAccount {
    public int Balance {
    get;
    set;
    }
    }
    class Listing_10 {
    static void Main(string[] args) {
    // create the bank account instance
    BankAccount account = new BankAccount();
    // create the spinlock
    SpinLock spinlock = new SpinLock();
    // create an array of tasks
    Task[] tasks = new Task[10];
    for (int i = 0; i < 10; i++) {
    // create a new task
    tasks[i] = new Task(() => {
    // enter a loop for 1000 balance updates
    for (int j = 0; j < 1000; j++) {
    bool lockAcquired = false;
    try {
    spinlock.Enter(ref lockAcquired);
    // update the balance
    account.Balance = account.Balance + 1;
    } finally {
    if (lockAcquired) spinlock.Exit();
    }
    }
    });
    // start the new task
    tasks[i].Start();
    }
    // wait for all of the tasks to complete
    Task.WaitAll(tasks);
    // write out the counter value
    Console.WriteLine("Expected value {0}, Balance: {1}",
    10000, account.Balance);
    // wait for input before exiting
    Console.WriteLine("Press enter to finish");
    Console.ReadLine();
    }
    }
    }

需要注意的是Spinlock不是循环锁，不要尝试在没有释放就重复获取这个锁， 这将导致死锁的产生或者运行时抛出异常，这取决于你构造Spinlock时的传入参数。

#### 多功能锁——Mutex ####

Mutex称作互斥锁，它可以等待一处的代码同步，也可以等待多处代码同步。

（1）等待一处的代码同步——Mutex::WaitOne()和Mutex::ReleaseMutex()

using System;
    using System.Threading.Tasks;
    using System.Threading;
    
    namespace _Mutex
    {
        class BankAccount
        {
            public int Balance
            {
                get;
                set;
            }
            class Program
            {
                static void Main(string[] args)
                {
                    // create the bank account instance
                    BankAccount account = new BankAccount();
                    // create the mutex
                    Mutex mutex = new Mutex();
                    // create an array of tasks
                    Task[] tasks = new Task[10];
                    for (int i = 0; i < 10; i++)
                    {
                        // create a new task
                        tasks[i] = new Task(() =>
                        {
                            // enter a loop for 1000 balance updates
                            for (int j = 0; j < 1000; j++)
                            {
                                // acquire the mutex
                                bool lockAcquired = mutex.WaitOne();
                                try
                                {
                                    // update the balance
                                    account.Balance = account.Balance + 1;
                                }
                                finally
                                {
                                    // release the mutext
                                    if (lockAcquired) mutex.ReleaseMutex();
                                }
                            }
                        });
                        // start the new task
                        tasks[i].Start();
                    }
                    // wait for all of the tasks to complete
                    Task.WaitAll(tasks);
                    // write out the counter value
                    Console.WriteLine("Expected value {0}, Balance: {1}", 10000, account.Balance);
                    // wait for input before exiting
                    Console.WriteLine("Press enter to finish");
                    Console.ReadLine();
                }
            }
        }
    }

等待多处代码同步——Mutex::WaitAll()

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_12
    {
        class BankAccount
        {
            public int Balance
            {
                get;
                set;
            }
        }
        class Listing_12
        {
            static void Main(string[] args)
            {
                // create the bank account instances
                BankAccount account1 = new BankAccount();
                BankAccount account2 = new BankAccount();
                // create the mutexes
                Mutex mutex1 = new Mutex();
                Mutex mutex2 = new Mutex();
                // create a new task to update the first account
                Task task1 = new Task(() => {
                    // enter a loop for 1000 balance updates
                    for (int j = 0; j < 1000; j++)
                    {
                        // acquire the lock for the account
                        bool lockAcquired = mutex1.WaitOne(); ;
                        try
                        {
                            // update the balance
                            account1.Balance++;
                        }
                        finally
                        {
                            if (lockAcquired) mutex1.ReleaseMutex();
                        }
                    }
                });
                // create a new task to update the first account
                Task task2 = new Task(() => {
                    // enter a loop for 1000 balance updates
                    for (int j = 0; j < 1000; j++)
                    {
                        // acquire the lock for the account
                        bool lockAcquired = mutex2.WaitOne();
                        try
                        {
                            // update the balance
                            account2.Balance += 2;
                        }
                        finally
                        {
                            if (lockAcquired) mutex2.ReleaseMutex();
                        }
                    }
                });
                // create a new task to update the first account
                Task task3 = new Task(() => {
                    // enter a loop for 1000 balance updates
                    for (int j = 0; j < 1000; j++)
                    {
                        // acquire the locks for both accounts
                        bool lockAcquired = Mutex.WaitAll(new WaitHandle[] { mutex1, mutex2 });
                        try
                        {
                            // simulate a transfer between accounts
                            account1.Balance++;
                            account2.Balance--;
                        }
                        finally
                        {
                            if (lockAcquired)
                            {
                                mutex1.ReleaseMutex();
                                mutex2.ReleaseMutex();
                            }
                        }
                    }
                });
                // start the tasks
                task1.Start();
                task2.Start();
                task3.Start();
                // wait for the tasks to complete
                Task.WaitAll(task1, task2, task3);
                // write out the counter value
                Console.WriteLine("Account1 balance {0}, Account2 balance: {1}",
                account1.Balance, account2.Balance);
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

#### 读写锁 ####

读写锁的应用场景非常明确，因为变量的读并不会破坏或者更改数据，这时候应该允许其他的线程对该变量进行读操作；但写操作就不一定了，因为一个线程在写操作其间读取一个变量的值时，读取到的可能是脏值。读写锁就是基于这一思想——允许多个进行对同一变量进行读操作，但只有一个线程能够对变量进行写操作。换句话说：读是共享的，写是排他的。

C\#中有两种读写锁，分别是ReaderWriterLockSlim和ReaderWriter。它们均在System.Threading命名空间下。ReaderWriterLockSlim锁是对ReaderWriter的改进。现在MS已经不再推荐使用ReaderWriter，原因是ReaderWriter会消耗更多的CPU资源和更容易陷入死锁。

锁方法：EnterReadLock(), ExitReadLock(), EnterWriteLock(), ExitWriteLock()，这些方法见名知义，不作多解析。例程如下：

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_15
    {
        class Listing_15
        {
            static void Main(string[] args)
            {
                // create the reader-writer lock
                ReaderWriterLockSlim rwlock = new ReaderWriterLockSlim();
                // create a cancellation token source
                CancellationTokenSource tokenSource = new CancellationTokenSource();
                // create an array of tasks
                Task[] tasks = new Task[5];
                for (int i = 0; i < 5; i++)
                {
                    // create a new task
                    tasks[i] = new Task(() => {
                        while (true)
                        {
                            // acqure the read lock
                            rwlock.EnterReadLock();
                            // we now have the lock
                            Console.WriteLine("Read lock acquired - count: {0}",
                            rwlock.CurrentReadCount);
                            // wait - this simulates a read operation
                            tokenSource.Token.WaitHandle.WaitOne(1000);
                            // release the read lock
                            rwlock.ExitReadLock();
                            Console.WriteLine("Read lock released - count {0}",
                            rwlock.CurrentReadCount);
                            // check for cancellation
                            tokenSource.Token.ThrowIfCancellationRequested();
                        }
                    }, tokenSource.Token);
                    // start the new task
                    tasks[i].Start();
                }
                // prompt the user
                Console.WriteLine("Press enter to acquire write lock");
                // wait for the user to press enter
                Console.ReadLine();
                // acquire the write lock
                Console.WriteLine("Requesting write lock");
                rwlock.EnterWriteLock();
                Console.WriteLine("Write lock acquired");
                Console.WriteLine("Press enter to release write lock");
                // wait for the user to press enter
                Console.ReadLine();
                // release the write lock
                rwlock.ExitWriteLock();
                // wait for 2 seconds and then cancel the tasks
                tokenSource.Token.WaitHandle.WaitOne(2000);
                tokenSource.Cancel();
                try
                {
                    // wait for the tasks to complete
                    Task.WaitAll(tasks);
                }
                catch (AggregateException)
                {
                    // do nothing
                }
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

使用读写锁的另一种情况是：在读锁定后，释放读锁定前，我们希望获得其写锁，这时就不能用EnterReadLock()方法进行锁定了。正确的用法应用是使用EnterUpgradeableReadLock()方法，就其名称而言，这个也是一个读锁，但这个读锁却不是共享锁，因此多个线程不能共享这个锁。例程如下：

using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using System.Threading;
    
    namespace _ReaderWriter
    {
        class Program
        {
            static void Main(string[] args)
            {
                // create the reader-writer lock
                ReaderWriterLockSlim rwlock = new ReaderWriterLockSlim();
                // create a cancellation token source
                CancellationTokenSource tokenSource = new CancellationTokenSource();
                // create some shared data
                int sharedData = 0;
                // create an array of tasks
                Task[] readerTasks = new Task[5];
                for (int i = 0; i < readerTasks.Length; i++)
                {
                    // create a new task
                    readerTasks[i] = new Task(() =>
                    {
                        while (true)
                        {
                            // acqure the read lock
                            rwlock.EnterReadLock();
                            // we now have the lock
                            Console.WriteLine("Read lock acquired - count: {0}",
                            rwlock.CurrentReadCount);
                            // read the shared data
                            Console.WriteLine("Shared data value {0}", sharedData);
                            // wait - slow things down to make the example clear
                            tokenSource.Token.WaitHandle.WaitOne(1000);
                            // release the read lock
                            rwlock.ExitReadLock();
                            Console.WriteLine("Read lock released - count {0}",
                            rwlock.CurrentReadCount);
                            // check for cancellation
                            tokenSource.Token.ThrowIfCancellationRequested();
                        }
                    }, tokenSource.Token);
                    // start the new task
                    readerTasks[i].Start();
                }
                Task[] writerTasks = new Task[2];
                for (int i = 0; i < writerTasks.Length; i++)
                {
                    writerTasks[i] = new Task(() =>
                    {
                        while (true)
                        {
                            // acquire the upgradeable lock
                            rwlock.EnterUpgradeableReadLock();
                            // simulate a branch that will require a write
                            if (true)
                            {
                                // acquire the write lock
                                rwlock.EnterWriteLock();
                                // print out a message with the details of the lock
                                Console.WriteLine("Write Lock acquired - waiting readers {0},writers {1}, upgraders {2}",
                            rwlock.WaitingReadCount, rwlock.WaitingWriteCount, rwlock.WaitingUpgradeCount);
                                // modify the shared data
                                sharedData++;
                                // wait - slow down the example to make things clear
                                tokenSource.Token.WaitHandle.WaitOne(1000);
                                // release the write lock
                                rwlock.ExitWriteLock();
                            }
                            // release the upgradable lock
                            rwlock.ExitUpgradeableReadLock();
                            // check for cancellation
                            tokenSource.Token.ThrowIfCancellationRequested();
                        }
                    }, tokenSource.Token);
                    // start the new task
                    writerTasks[i].Start();
                }
                // prompt the user
                Console.WriteLine("Press enter to cancel tasks");
                // wait for the user to press enter
                Console.ReadLine();
                // cancel the tasks
                tokenSource.Cancel();
                try
                {
                    // wait for the tasks to complete
                    Task.WaitAll(readerTasks);
                }
                catch (AggregateException agex)
                {
                    agex.Handle(ex => true);
                }
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

### 进程间数据同步 ###

#### Mutex ####

Mutex类可用于进程间的数据同步，与进程内数据同步不同的是，Mutex在用于进行间数据同步时，必须要有一个标识来标识同一个Mutex，这通过new一个Mutex时传入字符串来区分。

怎么判断一个Mutex已经被创建，这时可以通过字符串来进行查看，因为不同进行的Mutex是通过字符串来标识的。检测方法是通过Mutex类的静态方法OpenExist(string)来获得一个由string标识的Mutex,当这个Mutex不存在时，这个方法将返回System.Threading.WaitHandleCannotBeOpenedException异常，这时就知道该Mutex还不存在，应该new一个出来。例程如下：

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_13
    {
        class Listing_13
        {
            static void Main(string[] args)
            {
                // declare the name we will use for the mutex
                string mutexName = "myApressMutex";
                // declare the mutext
                Mutex namedMutext;
                try
                {
                    // test to see if the named mutex already exists
                    namedMutext = Mutex.OpenExisting(mutexName);
                }
                catch (WaitHandleCannotBeOpenedException)
                {
                    // the mutext does not exist - we must create it
                    namedMutext = new Mutex(false, mutexName);
                }
                // create the task
                Task task = new Task(() => {
                    while (true)
                    {
                        // acquire the mutex
                        Console.WriteLine("Waiting to acquire Mutex");
                        namedMutext.WaitOne();
                        Console.WriteLine("Acquired Mutex - press enter to release");
                        Console.ReadLine();
                        namedMutext.ReleaseMutex();
                        Console.WriteLine("Released Mutex");
                    }
                });
                // start the task
                task.Start();
                // wait for the task to complete
                task.Wait();
            }
        }
    }

[Center]: /images/20220717/40747c5923d64bde9f310c3d4d9ef688.png