高并发场景之一般解决方案

今天我们来了解一下一些高并发的业务场景如何做到数据一致性的。

一、场景：

1、有数据表：ConCurrency，

CREATE TABLE [dbo].[ConCurrency](
[ID] [int] NOT NULL,
[Total] [int] NULL
)

2、初始值：ID=1,Total = 0

3、现要求每一次客户端请求Total + 1

二、单线程

static void Main(string[] args)
{
...
new Thread(Run).Start();
...
}

public static void Run()
{
for (int i = 1; i <= 100; i++)
{
var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString();
var value = int.Parse(total) + 1;

DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null);
Thread.Sleep(1);
}
}

2.1 按要求，正常情况下应该输出：100

2.2 运行结果



貌似没有问题。

三、多线程并发

3.1 Main改一下

static void Main(string[] args)
{
...
new Thread(Run).Start();
new Thread(Run).Start();
...
}

3.2 我们预期应该是要输出200

3.3 运行结果



很遗憾，却是150，造成这个结果的原因是这样的：T1、T2获取Total（假设此时值为10），T1更新一次或多次后，T2才更新（Total:10）

这就造成之前T1提交的被覆盖了

3.4 如何避免呢？一般做法加锁就可以了，如Run改成如下

public static void Run()
{
for (int i = 1; i <= 100; i++)
{
lock (resource)
{
var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString();
var value = int.Parse(total) + 1;

DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null);
}

Thread.Sleep(1);
}
}

3.5 再次运行

四、用队列实现

4.1、定义队列

static ConcurrentQueue<int> queue = new ConcurrentQueue<int>();

/// <summary>生产者</summary>
public static void Produce()
{
for (int i = 1; i <= 100; i++)
{
queue.Enqueue(i);
}
}

/// <summary>消费者</summary>
public static void Consume()
{
int times;
while (queue.TryDequeue(out times))
{
var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString();
var value = int.Parse(total) + 1;

DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null);
Thread.Sleep(1);
}
}

4.2 Main改一下

static void Main(string[] args)
{
...
new Thread(Produce).Start();
new Thread(Produce).Start();
Consume();
...
}

4.3 预期输出200，看运行结果



4.4 集群环境下测试，2台机器





有问题！最后运行的那台机器居然是379，数据库也是379。

这超出了我们的预期结果，看来即便加锁，对于高并发场景也是不能解决所有问题的

五、分布式队列

5.1 解决上边问题可以用分布式队列，这里用的是redis队列

/// <summary>生产者</summary>
public static void ProduceToRedis()
{
using (var client = RedisManager.GetClient())
{
for (int i = 1; i <= 100; i++)
{
client.EnqueueItemOnList("EnqueueName", i.ToString());
}
}
}

/// <summary>消费者</summary>
public static void ConsumeFromRedis()
{
using (var client = RedisManager.GetClient())
{
while (client.GetListCount("EnqueueName") > 0)
{
if (client.SetValueIfNotExists("lock", "lock"))
{
var item = client.DequeueItemFromList("EnqueueName");
var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString();
var value = int.Parse(total) + 1;

DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null);

client.Remove("lock");
}

Thread.Sleep(5);
}
}
}

5.2 Main也要改改

static void Main(string[] args)
{
...
new Thread(ProduceToRedis).Start();
new Thread(ProduceToRedis).Start();
Thread.Sleep(1000 * 10);

ConsumeFromRedis();
...
}

5.3 在集群里再试试，2个都是400，没有错（因为每个站点开了2个线程）





可以看到数据完全正确！