首先在windows上安装好Redis，RabbitMQ
Redis-cli使用示例
ModelContext.cs代码:
public class ModelContext : DbContext
{
//您的上下文已配置为从您的应用程序的配置文件(App.config 或 Web.config)
//连接字符串。
public ModelContext()
: base("name=default")
{
}
public virtual DbSet<Person> Person { get; set; }
}
public class Person
{
public int Id { get; set; }
public string Id2 { get; set; }
public string Name { get; set; }
}在 Package Manager Console 下运行命令 Enable-Migrations
这个命令将在项目下创建文件夹 Migrations
The Configuration class 这个类允许你去配置如何迁移，对于本文将使用默认的配置（在本文中因为只有一个 Context, Enable-Migrations 将自动对 context type 作出适配）;
An InitialCreate migration （本文为201702220232375_20170222.cs）这个迁移之所以存在是因为我们之前用 Code First 创建了数据库, 在启用迁移前，scaffolded migration 里的代码表示在数据库中已经创建的对象，本文中即为表 Person（列 Id 和 Name）. 文件名包含一个 timestamp 以便排序（如果之前数据库没有被创建，那么 InitialCreate migration 将不会被创建，相反，当我们第一次调用 Add-Migration 的时候所有表都将归集到一个新的 migration 中）
多个实体锁定同一数据库
当使用 EF6 之前的版本时，只会有一个 Code First Model 被用来生成/管理数据库的 Schema, 这将导致每个数据库只会有一张 __MigrationsHistory 表，从而无法辨别实体与模型的对应关系。
从 EF6 开始，Configuration 类将会包含一个 ContextKey 属性，它将作为每一个 Code First Model 的唯一标识符， __MigrationsHistory 表中一个相应地的列允许来自多个模型（multiple models）的实体共享表（entries），默认情况下这个属性被设置成 context 的完全限定名。
定制化迁移
在 Package Manager Console 中运行命令 Add-Migration XXXXXXXXX
生成的迁移如下
public partial class _20170222 : DbMigration
{
public override void Up()
{
CreateTable(
"dbo.People",
c => new
{
Id = c.Int(nullable: false, identity: true),
Id2 = c.String(),
Name = c.String(),
})
.PrimaryKey(t => t.Id);
}
public override void Down()
{
DropTable("dbo.People");
}
}Configuration.cs代码:
internal sealed class Configuration : DbMigrationsConfiguration<EF.ModelContext>
{
public Configuration()
{
AutomaticMigrationsEnabled = false;
}
protected override void Seed(EF.ModelContext context)
{
}
}我们对迁移做些更改：
以下是本项目无关的其他示例:
namespace MigrationsDemo.Migrations
{
using System;
using System.Data.Entity.Migrations;
public partial class AddPostClass : DbMigration
{
public override void Up()
{
CreateTable(
"dbo.Posts",
c => new
{
PostId = c.Int(nullable: false, identity: true),
Title = c.String(maxLength: 200),
Content = c.String(),
BlogId = c.Int(nullable: false),
})
.PrimaryKey(t => t.PostId)
.ForeignKey("dbo.Blogs", t => t.BlogId, cascadeDelete: true)
.Index(t => t.BlogId)
.Index(p => p.Title, unique: true);
AddColumn("dbo.Blogs", "Rating", c => c.Int(nullable: false, defaultValue: 3));
}
public override void Down()
{
DropIndex("dbo.Posts", new[] { "Title" });
DropForeignKey("dbo.Posts", "BlogId", "dbo.Blogs");
DropIndex("dbo.Posts", new[] { "BlogId" });
DropColumn("dbo.Blogs", "Rating");
DropTable("dbo.Posts");
}
}
}在 Package Manager Console 中运行命令 Update-Database –Verbose
消费者端，用来把消息队列里的数据写入数据库
MqHelper.cs代码:
public class MqHelper
{
private static IConnection _connection;
/// <summary>
/// 获取连接对象
/// </summary>
/// <returns></returns>
public static IConnection GetConnection()
{
if (_connection != null) return _connection;
_connection = GetNewConnection();
return _connection;
}
public static IConnection GetNewConnection()
{
//从工厂中拿到实例 本地host、用户admin
var factory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "guest",
Password = "guest",
};
_connection = factory.CreateConnection();
return _connection;
}
}Program.cs代码:
internal class Program
{
private static void Main(string[] args)
{
using (var channel = MqHelper.GetConnection().CreateModel())
{
//参数有 queue名字 是否持久化 独占的queue（仅供此连接） 不使用时是否自动删除 其他参数
channel.QueueDeclare("NET", true, false, false, null);
//我们要告诉服务器从队列里推送消息，因为消息是异步发送的，所以我们需要提供一个回调事件EventingBasicConsumer，用于处理接收到的消息。
//这就是 EventingBasicConsumer.Received 事件处理程序做的事。
var consumber = new EventingBasicConsumer(channel);
//QoS = quality-of-service， 顾名思义，服务的质量。
//代码第一个参数是可接收消息的大小的，但是似乎在客户端2.8.6版本中它必须为0，即使：不受限制。
//如果不输0，程序会在运行到这一行的时候报错，说还没有实现不为0的情况。
//第二个参数是处理消息最大的数量。举个例子，如果输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息，消息只会在队列中阻塞。
//如果输入3，那么可以最多有3个消息不应答，如果到达了3个，则发送端发给这个接收方得消息只会在队列中，而接收方不会有接收到消息的事件产生。
//总结说，就是在下一次发送应答消息前，客户端可以收到的消息最大数量。
//第三个参数则设置了是不是针对整个Connection的，因为一个Connection可以有多个Channel，如果是false则说明只是针对于这个Channel的。
//Fair dispatch 公平分发
//通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。 设置方法如下：
channel.BasicQos(0, 1, false);
consumber.Received += (sender, e) =>
{
try
{
var user = JsonConvert.DeserializeObject<User>(Encoding.UTF8.GetString(e.Body));
//Redis INCR命令用于将键的整数值递增1。如果键不存在，则在执行操作之前将其设置为0。 如果键包含错误类型的值或包含无法表示为整数的字符串，则会返回错误。此操作限于64位有符号整数。
var flag = RedisHelper.GetRedisClient().Incr(user.Id.ToString());
if (flag == 1)
{
//用户的第一次请求,为有效请求
//下面开始入库,这里使用List做为模拟
Console.WriteLine(string.Format("{0}标识为{1} {2}", user.Id, flag, user.Name));
var dbContext = new ModelContext();
dbContext.Person.Add(new Person() {Id2 = user.Id.ToString(), Name = user.Name});
Task ts = dbContext.SaveChangesAsync();
ts.Wait();
//添加入库标识
RedisHelper.GetRedisClient().Incr(string.Format("{0}入库", user.Id.ToString()));
Console.WriteLine("入库成功");
}
//用户的N次请求,为无效请求
channel.BasicAck(e.DeliveryTag, false); // 回发ACK 参数 tag 是否多个 //对message进行确认
}
catch (Exception ex)
{
File.AppendAllText(string.Format("{0}/bin/log.txt", System.AppDomain.CurrentDomain.BaseDirectory), ex.Message);
}
};
Console.WriteLine("开始工作");
// 如果 channel.BasicConsume 中参数 noAck 设置为 false，必须加上消息确认语句
// Message acknowledgment（消息确认机制作用）
//打开应答机制
//no_ack 的用途：确保 message 被 consumer “成功”处理了。
//这里“成功”的意思是，（在设置了 no_ack=false 的情况下）只要 consumer 手动应答了 Basic.Ack ，就算其“成功”处理了。
//情况一：no_ack=true （此时为自动应答）
//在这种情况下，consumer 会在接收到 Basic.Deliver + Content-Header + Content-Body 之后，立即回复 Ack 。
//而这个 Ack 是 TCP 协议中的 Ack 。此 Ack 的回复不关心 consumer 是否对接收到的数据进行了处理，当然也不关心处理数据所需要的耗时。
//情况二：no_ack=false （此时为手动应答）
//在这种情况下，要求 consumer 在处理完接收到的 Basic.Deliver + Content-Header + Content-Body 之后才回复 Ack 。
//而这个 Ack 是 AMQP 协议中的 Basic.Ack 。此 Ack 的回复是和业务处理相关的，所以具体的回复时间应该要取决于业务处理的耗时。
channel.BasicConsume("NET", false, consumber);
Console.ReadKey();
}
}
}模拟并发的MVC网站，写入队列
HomeController.cs代码
public class HomeController : Controller
{
// GET: Controller
public ActionResult Index()
{
return View();
}
/// <summary>
/// 抢单接口
/// </summary>
/// <param name="user"></param>
/// <returns></returns>
[HttpPost]
public ActionResult GrabSingle(User user)
{
//使用后台任务
//BackgroundJob.Enqueue(() => MqPublish.AddQueue(user));
MqPublish.AddQueue(user);
//MqPublish.AddQueue(user);
return Json(new { Status = "OK" });
}
/// <summary>
/// 获取数量
/// </summary>
/// <returns></returns>
[HttpGet]
public async Task<ActionResult> GetCount()
{
using (var dbcontext = new ModelContext())
{
return Json(new { Count = await dbcontext.Person.CountAsync() }, JsonRequestBehavior.AllowGet);
}
}
}MqPublish.cs类
/// <summary>
/// 发布者
/// </summary>
public class MqPublish
{
public const string QueueName = "NET";
public static IList<User> UserList = new List<User>();
/// <summary>
/// 添加到队列
/// </summary>
public static void AddQueue(User user)
{
//创建一个channel
using (var channel = MqHelper.GetNewConnection().CreateModel())
{
//json序列化
var bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(user));
//channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
//参数说明:
//要发出的交换机名字
//路由关键字
//是否强制（设置为true时，找不到收的人时可以通过returnListener返回）
//是否立即（其实rabbitmq不支持）
//其他属性
//消息主体
channel.BasicPublish(String.Empty, QueueName, null, bytes);
}
}
}RedisHelper.cs代码:
/// <summary>
/// Redis帮助类
/// </summary>
public class RedisHelper
{
public static RedisClient GetRedisClient()
{
return new RedisClient("127.0.0.1", 6379,"123456");
}
}Index.cshtml内容:
@{
Layout = null;
}
<!DOCTYPE html>
<html>
<head>
<meta name="viewport" content="width=device-width" />
<title>Index</title>
<link href="//cdn.bootcss.com/bootstrap/4.0.0-alpha.6/css/bootstrap.min.css" rel="stylesheet">
<script src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>
<script>
var uarray = [
{ Id: 1, Name: "小明" }, { Id: 2, Name: "张三" }, { Id: 3, Name: "李四" }, { Id: 4, Name: "王五" },
{ Id: 5, Name: "赵六" }, { Id: 6, Name: "钱八" }, { Id: 7, Name: "小红" }, { Id: 8, Name: "小紫" },
{ Id: 9, Name: "小蓝" }, { Id: 10, Name: "老王" }
];
function btnClick() {
//构造并发模拟
var flag;
for (var i = 0; i < 1000; i++) {
flag = parseInt(Math.random() * (9 - 0 + 1) + 0);
$.ajax({
type: "post",
url: "/Home/GrabSingle",
data: uarray[flag]
});
}
}
function GetCount() {
$.ajax({
type: "get",
url: "/Home/GetCount",
dataType: "json",
success: function (data) {
alert(data.Count + "个");
}
});
}
</script>
</head>
<body>
<h2>.Net高并发解决思路</h2>
<hr />
<button class="btn btn-primary" onclick="btnClick()">点击抢单</button>
<button class="btn btn-primary" onclick="GetCount()">查看入库数量</button>
</body>
</html>Startup.cs代码:
using System;
using System.Threading.Tasks;
using Hangfire;
using Hangfire.MemoryStorage;
using Microsoft.Owin;
using NetHigh.RabbitMq;
using Owin;
[assembly: OwinStartup(typeof(NetHigh.Startup))]
namespace NetHigh
{
public partial class Startup
{
public void Configuration(IAppBuilder app)
{
GlobalConfiguration.Configuration.UseMemoryStorage();
app.UseHangfireServer();
app.UseHangfireDashboard("/hangfire");
//添加三个后台任务也就是三个consumer
//BackgroundJob.Enqueue(() => MqConsumber.ConsumeQueue());
//BackgroundJob.Enqueue(() => MqConsumber.ConsumeQueue());
//BackgroundJob.Enqueue(() => MqConsumber.ConsumeQueue());
}
}
}Web.config
<?xml version="1.0" encoding="utf-8"?>
<configuration>
<configSections>
<section name="entityFramework" type="System.Data.Entity.Internal.ConfigFile.EntityFrameworkSection, EntityFramework, Version=6.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089" requirePermission="false" />
</configSections>
<system.web>
<compilation debug="true" targetFramework="4.5" />
<httpRuntime targetFramework="4.5" />
</system.web>
<entityFramework>
<defaultConnectionFactory type="System.Data.Entity.Infrastructure.LocalDbConnectionFactory, EntityFramework">
<parameters>
<parameter value="mssqllocaldb" />
</parameters>
</defaultConnectionFactory>
<providers>
<provider invariantName="System.Data.SqlClient" type="System.Data.Entity.SqlServer.SqlProviderServices, EntityFramework.SqlServer" />
</providers>
</entityFramework>
<connectionStrings>
<add name="default" connectionString="Server=.;User Id=sa;PassWord=123456;DataBase=Person" providerName="System.Data.SqlClient" />
</connectionStrings>
</configuration>运行结果如图:
运行消费者端(控制台运用程序)