[C#]使用RabbitMQ模拟抽奖系统的例子

　　背景：在实际的项目中，经常有客户需要做抽奖的活动，大部分的都是注册送产品、送红包这些需求。这都是有直接的利益效果，所以经常会遇见系统被盗刷的情况，每一次遇见这种项目的上线都是绷紧神经，客户又都喜欢在过节的时候上这种活动，有好多次放假前夕都是在解决这种事情，甚至有一次的活动短信接口直接被恶意刷爆了。在这种恶意请求下对系统并发性要求就很高，但是即使做多方面的完善，有一个问题始终得不到根本的解决，那就是奖品池数量的控制，总是会出现超兑，或者一个奖品被多个人兑走的问题。之后尝试了多种及方法，例如：限制IP，限制次数等等。后来最有效的解决方法就是使用Redis锁住奖品逻辑，但是这种实现有点复杂，也不是很友好，因此就想到了使用消息队列的优势来实现此功能。

　　做这个示例首先是为了学习，再者也是留下学习的笔记，不然后面又遗忘掉了

　　这个示例是一边学习RabbitMQ，一边实现自己的需求功能的。主要功能有【投放奖品】、【模拟多户请求】、【模拟用户抽奖】，并且在这些操作中及时的展示各个队列中数据的数量变化，先上一张效果图：

示例测试下来，始终能保证奖品的数量与实际的中奖人数是一致的，不会多出一个中奖人，也不会出现有多个人中同一个奖品的问题。

实现方式主要就是多线程模拟用户请求，结合RabbitMQ，其中还是用了RabbitMQ的在线API进行数据的监控展示。

实现思路：

1：先将奖品丢入奖品池;

#region 投放奖品
        /// <summary>
        /// 投放奖品
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn1_Click(object sender, EventArgs e)
        {
            try
            {
                SetSendfigModel(PrizeQueueName);  //设置队列信息(奖品池)
                new Thread(SetPrize) { IsBackground = true }.Start();
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "出错了", MessageBoxButtons.OK);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        private void SetPrize()
        {
            string value = string.Empty;
            for (int i = 1; i <= PrizeCount; i++)
            {
                PrizeInfo prize = new PrizeInfo
                {
                    Id = i,
                    Name = "我是奖品" + i,
                    Type = 1,
                    PrizeNo = DateTime.Now.ToString("hhmmssfff"),
                    Total = PrizeCount,
                    Balance = PrizeCount
                };
                value = JsonConvert.SerializeObject(prize);
                RabbitSend.Send(prize);
                ShowSysMessage($"我骄傲，我是奖品：{i}/{PrizeCount}");
            }
            ShowSysMessage("奖品投放完成");
        }
        #endregion

2:模拟多用户页面请求

　　利用多线程实现用户随机访问抽奖系统，这里将所有用户的信息来了就做插入到用户池当中，后续进行抽奖的时候再从用户池中顺序取出。

  #region 模拟多用户页面请求
        /// <summary>
        /// 模拟多用户页面请求
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn2_Click(object sender, EventArgs e)
        {
            try
            {
                SetSendfigModel(UserQueueName);  //设置队列信息(用户池)
                ShowSysMessage("开始模拟多用户页面请求...");
                new Thread(ThreadFunction) { IsBackground = true }.Start();
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "出错了", MessageBoxButtons.OK);
            }
        }

        private const int threadLength = 8;
        private static CancellationTokenSource cts = new CancellationTokenSource();

        /// <summary>
        /// 
        /// </summary>
        private void ThreadFunction()
        {
            cts = new CancellationTokenSource();
            TaskFactory taskFactory = new TaskFactory();
            Task[] tasks = new Task[threadLength];

            for (int i = 0; i < threadLength; i++)
            {
                Task t1 = Task.Factory.StartNew(delegate { ParallelFunction(cts.Token); });
                tasks.SetValue(t1, i);
            }
            taskFactory.ContinueWhenAll(tasks, TasksEnded, CancellationToken.None);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="tasks"></param>
        void TasksEnded(Task[] tasks)
        {
            ShowSysMessage("所有任务已完成/或已取消！");
        }

        /// <summary>
        /// 
        /// </summary>
        private void ParallelFunction(CancellationToken ct)
        {
            Parallel.For(0, 1000, item =>
            {
                if (!ct.IsCancellationRequested)
                {
                    string value = string.Empty;
                    UsersInfo user = new UsersInfo
                    {
                        Id = item,
                        Name = "我是：" + item
                    };
                    value = Newtonsoft.Json.JsonConvert.SerializeObject(user);
                    ShowSysMessage($"进来了一位用户：{value}");
                    RabbitSend.Send(user);
                }
            });
        }
        #endregion

3：模拟多用户抽奖

　　从用户池中顺序取出一个用户进行奖品的锁定，锁定之后生成用户与奖品的关系，插入中奖池中。

 #region 模拟多用户抽奖

        /// <summary>
        /// 模拟多用户抽奖
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn3_Click(object sender, EventArgs e)
        {
            //1:先去用户池中取出一个人 2 拿用户去抽一个奖品 3:将中奖人塞入中奖队列
            new Thread(() =>
            {
                for (int i = 0; i < 10000; i++)
                {
                    SetReceivefigModel(UserQueueName);//设置队列信息(用户池)  
                    RabbitReceive.BasicGet(LockUser);
                }

                //Parallel.For(0, 200000, item =>
                //{
                //    RabbitReceive.BasicGet(LockUser);
                //});
            })
            { IsBackground = true }.Start();
        }

        /// <summary>
        /// 先去用户池中取出一个人
        /// </summary>
        /// <param name="tp"></param>
        private void LockUser(ValueTuple<bool, string, Dictionary<string, object>> tp)
        {
            try
            {
                if (tp.Item1)
                {
                    ShowSysMessage($"锁定到一个用户：{tp.Item2}");
                    UsersInfo user = JsonConvert.DeserializeObject<UsersInfo>(tp.Item2);
                    if (null != user)
                    {
                        Thread.Sleep(50);
                        LockPrize(user);//拿用户去抽一个奖品
                    }
                }
                else
                {
                    ShowSysMessage(tp.Item2);
                }
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "出错了", MessageBoxButtons.OK);
            }
        }

        /// <summary>
        /// 拿用户去抽一个奖品
        /// </summary>
        /// <param name="user"></param>
        private void LockPrize(UsersInfo user)
        {
            SetReceivefigModel(PrizeQueueName);//设置队列信息(奖品池)
            Dictionary<string, object> data = new Dictionary<string, object> { { "User", user } };
            RabbitReceive.BasicGet(LockPrize, data);
        }

        /// <summary>
        /// 锁定奖品
        /// </summary>
        /// <param name="value"></param>
        private void LockPrize(ValueTuple<bool, string, Dictionary<string, object>> tp)
        {
            try
            {
                if (tp.Item1)
                {
                    UsersInfo user = tp.Item3["User"] as UsersInfo;
                    PrizeInfo prize = JsonConvert.DeserializeObject<PrizeInfo>(tp.Item2);
                    if (null != user && null != prize)
                    {
                        user.PrizeInfo = prize;
                        ShowSysMessage($"用户{user.Name}锁定到一个奖品：{tp.Item2}");
                        PrizeUser(user);// 将中奖人塞入中奖队列
                    }
                }
                else
                {
                    ShowSysMessage(tp.Item2);
                }
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "出错了", MessageBoxButtons.OK);
            }
        }

        /// <summary>
        /// 将中奖人塞入中奖队列
        /// </summary>
        /// <param name="user"></param>
        private void PrizeUser(UsersInfo user)
        {
            SetSendfigModel(PrizeUserQueueName);  //设置队列信息(中奖人)
            RabbitSend.Send(user);
            Thread.Sleep(50);
        }
        #endregion

4：使用RabbitMQ的在线API进行数据的监控展示

 #region 处理队列中数据

        /// <summary>
        /// 
        /// </summary>
        private void LoadData()
        {
            System.Timers.Timer t = new System.Timers.Timer(3000);   //实例化Timer类，设置间隔时间为10000毫秒；   
            t.Elapsed += new System.Timers.ElapsedEventHandler(InitRabbit); //到达时间的时候执行事件；   
            t.AutoReset = true;   //设置是执行一次（false）还是一直执行(true)；   
            t.Enabled = true;     //是否执行System.Timers.Timer.Elapsed事件；   
        }

        /// <summary>
        /// 初始化队列中已有的数据
        /// </summary>
        /// <param name="source"></param>
        /// <param name="e"></param>
        private void InitRabbit(object source, System.Timers.ElapsedEventArgs e)
        {
            if (this.IsHandleCreated)
            {
                Invoke(new Action(() =>
                {
                    ShowLbUserUserExchanges(RabbitSendConfig.ExchangesApi);
                    ShowLbQueues(RabbitSendConfig.QueuesApi);
                    ShowLbBindings(RabbitSendConfig.BingdingsApi);
                    ShowSysMessage($"[{DateTime.Now}]数据已更新....................");
                }));
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="apiUrl"></param>
        private async void ShowLbUserUserExchanges(string apiUrl)
        {
            userExchanges = await GetListModel<List<ExchangeEntity>>(apiUrl);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="apiUrl"></param>
        private async void ShowLbQueues(string apiUrl)
        {
            queues = await GetListModel<List<QueueEntity>>(apiUrl);
            if (queues != null && queues.Any())
            {
                lbQueues.Items.Clear();
                lbPrize.Text = "0";
                lbUser.Text = "0";
                lbPrizeUser.Text = "0";
                foreach (var queueEntity in queues)
                {
                    lbQueues.Items.Add(queueEntity.name);
                    if (queueEntity.name == PrizeQueueName)
                    {
                        lbPrize.Text = queueEntity.messages_ready.ToString();  //奖品剩余数量
                    }
                    if (queueEntity.name == UserQueueName)
                    {
                        lbUser.Text = queueEntity.messages_ready.ToString();  //用户数量
                    }
                    if (queueEntity.name == PrizeUserQueueName)
                    {
                        lbPrizeUser.Text = queueEntity.messages_ready.ToString();  //中奖人数
                    }
                }
            }
            else
            {
                lbQueues.Items.Clear();
                lbPrize.Text = "0";
                lbUser.Text = "0";
                lbPrizeUser.Text = "0";
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="apiUrl"></param>
        private async void ShowLbBindings(string apiUrl)
        {
            bindings = await GetListModel<List<BindingEntity>>(apiUrl);
            if (bindings != null)
            {
                lbBindings.Items.Clear();
                foreach (var bindingEntity in bindings)
                {
                    lbBindings.Items.Add(string.Format("交换机:{0}---队列:{1}---Key:{2}", string.IsNullOrWhiteSpace(bindingEntity.source) ? "默认" : bindingEntity.source, bindingEntity.destination, bindingEntity.routing_key));
                }
            }
            else
            {
                lbBindings.Items.Clear();
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="apiUrl"></param>
        /// <returns></returns>
        private async Task<T> GetListModel<T>(string apiUrl)
        {
            string jsonContent = await ShowApiResult(apiUrl);
            return JsonConvert.DeserializeObject<T>(jsonContent);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="apiUrl"></param>
        /// <returns></returns>
        private async Task<string> ShowApiResult(string apiUrl)
        {
            var response = await ShowHttpClientResult(apiUrl);
            response.EnsureSuccessStatusCode();
            string responseBody = await response.Content.ReadAsStringAsync();
            return responseBody;
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="Url"></param>
        /// <returns></returns>
        private async Task<HttpResponseMessage> ShowHttpClientResult(string Url)
        {
            var client = new HttpClient();
            var byteArray = Encoding.ASCII.GetBytes(string.Format("{0}:{1}", RabbitReceiveConfig.UserName, RabbitReceiveConfig.Password));
            client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Basic", Convert.ToBase64String(byteArray));
            HttpResponseMessage response = await client.GetAsync(Url);
            return response;
        }
        #endregion

基本上大致的实现逻辑就是以上这些了，但是其实还有一个逻辑的问题我没有处理

　　这里要中奖用户是唯一的，实现这一点可以从两点入手

　　1：用户池用户信息唯一；

　　2：锁定奖品时要唯一；

这两点都可以实现这个逻辑，但是暂时还不知道RabbitMQ是否支持消息的唯一性，或者可以通过DB/Redis来实现。

其他具体的代码就不做展示，直接在附件中体现。

代码环境

win10 + Visual Studio Community 2017

代码下载