.NET文件并发与RabbitMQ(初探RabbitMQ)

Posted DotNet

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了.NET文件并发与RabbitMQ(初探RabbitMQ)相关的知识,希望对你有一定的参考价值。


来源:坦荡

链接:cnblogs.com/tdws/p/5860668.html


想必MQ这两个字母对于各位前辈们和老司机们并不陌生。本文初探RabbitMQ的简单分享可能值得学习之处不怎么多,本人对于RabbitMQ的研究目前也很初级,这个月打算按照好的学习线路提高一下,欢迎新老司机留下你们的见解。


 

首先提到第一个简单的场景,文件并发。我先手动实现一下文件并发,引发异常,请看如下代码。


static void Main(string[] args)

{

            new Thread(write1).Start();

            new Thread(write1).Start();

            new Thread(write1).Start();

            Console.WriteLine("等待");

            Console.ReadKey();

}

public static void write1()

{

            for (int i = 0; i < 10000; i++)

            {

                WriteLog(i);

            }

            //Console.ReadKey();

}

public static void WriteLog(int i)

{

            using (FileStream f = new FileStream(@"d:\\A.txt", FileMode.Append))

            {

                using (StreamWriter sw = new StreamWriter(f, Encoding.Default))

                {

                    sw.Write(i);

                }

            }

}


我使用多线程并发向同一个append数据。相信你应该知道接下来运行起来会发生什么!


.NET文件并发与RabbitMQ(初探RabbitMQ)


是的,正如你所料,该文件正由另一个进程使用,因此该进程无法访问此文件。也许这个场景,就像你写应用程序运行日志,异常日志。如果你没使用任何插件或者组件,就只能直接向文件中append。这样的问题,是你不得不解决的。


这个时候就是队列出场的时候了。当然队列的组件有很多,.NET框架下也有自带的队列Queue,微软也有独立的队列组件MSQueue。Apache有其ActiveMQ,另外知名的消息队列还有Equeue,ZeroMQ等。消息队列的使用场景,大概包括解耦,提高峰值处理能力,送达和排序保证,缓冲等。


RabbitMQ是一个消息中间件,其主要的观点很简单:接受和转发消息。你可以把他想象成邮局,当你发送新建到邮递箱,你很确定邮递员最终将会把你的新建传递到你的收件人手中。我们使用邮递箱和邮递员来隐喻RabbitMQ。关于RabbitMQ和邮局之前的主要区别在于邮局处理纸质信件,而MQ存储和转发二进制数据-message。下面提到RabbitMQ的几个”行话“。


生产者意味着发送消息。一个发送消息的应用程序是一个生产者,我们称其为"P"。


.NET文件并发与RabbitMQ(初探RabbitMQ)


队列queue意味着邮递箱。他存在于RabbitMQ当中.尽管消息在RabbitMQ和你的应用程序中”流通“,他们可以被仅存在一个队列当中。一个队列不受任何限制,他可以存储你想要存储的消息量,它本质上是一个无限的缓冲区。多个生产者可以向同一个队列发送消息,多个消费者可以尝试从同一个消息队列中接收数据。一个队列像下面这样,上面是它的队列名称。


.NET文件并发与RabbitMQ(初探RabbitMQ)


消费者意味着接收,消费者是等待接收处理消息的应用程序。


.NET文件并发与RabbitMQ(初探RabbitMQ)


最后的关系就如下图:


.NET文件并发与RabbitMQ(初探RabbitMQ)


接下来,我将使用RabbitMQ来解决最开始的文件并发问题。也就是说为了避免文件并发,我们要将生产者所需append到文件中的内容存入到消息队列当中,然后取出队列中的message让消费者写入到文件当中。


关于RabbitMQ的安装和配置请先看张善友老师的文章:

http://www.cnblogs.com/shanyou/p/4067250.html,写的特别详细。


只不过历经数月RabbitMQ C#客户端方法上有略微的更新。


下面我将新建两个ConsoleApp应用程序。一个负责Receive,消费者,写文件。另一个Send,生产者,将所需写入的数据推到RabbitMQ当中。


Receive代码如下:


static void Main(string[] args)

{

            var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "WuShuang",  Password = "123456" };

            using (var connection = factory.CreateConnection())

            using (var channel = connection.CreateModel())

            {

                channel.QueueDeclare(queue: "hello",

                                     durable: false,

                                     exclusive: false,

                                     autoDelete: false,

                                     arguments: null);

                

                var consumer = new EventingBasicConsumer(channel);

                consumer.Received += (model, ea) =>

                {

                    var body = ea.Body;

                    var message = Encoding.UTF8.GetString(body);

                    WriteLog(message);

                };

                channel.BasicConsume(queue: "hello",

                                     noAck: true,

                                     consumer: consumer);


                Console.WriteLine(" Press [enter] to exit.");

                Console.ReadLine();

            }

        }

        public static void WriteLog(string i)

        {

            using (FileStream f = new FileStream(@"d:\\A.txt", FileMode.Append))

            {

                using (StreamWriter sw = new StreamWriter(f, Encoding.Default))

                {

                    sw.Write(i + "\n");

                }

            }

        }


Send代码如下:


static void Main(string[] args)

        {

            new Thread(write1).Start();

            new Thread(write1).Start();

            new Thread(write1).Start();

            Console.WriteLine(" Press [enter] to exit.");

            Console.ReadLine();

        }


        public static void write1()

        {

            var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "WuShuang", Password = "123456" };

            using (var connection = factory.CreateConnection())

            using (var channel = connection.CreateModel())

            {

                channel.QueueDeclare(queue: "hello", durable: false,

                                     exclusive: false,

                                     autoDelete: false,

                                     arguments: null);

                channel.QueueBind("hello", "wsExchange", "hello");

                for (int i = 0; i < 10000; i++)

                {

                    string message = "Hello World ws!" + i;

                    var body = Encoding.UTF8.GetBytes(message);


                    channel.BasicPublish(exchange: "wsExchange",

                                  routingKey: "hello",

                                  basicProperties: null,

                                  body: body);

                    Console.WriteLine(" [x] Sent {0}", message);

                }

            }

            //Console.ReadKey();

        }


有了代码后,首先把Receive项目运行起来。等待生产者Send消息到队列中,然后Receive订阅的事件将会取出消息,并将消息写入到文件当中。



左侧的ConsoleApp持续向队列中写入消息,右侧的Receive的ConsoleApp运行过程中,会不断写文件。生产者承受了多线程并发些数据,当然消费者有序的取出队列中的数据,也不会发生最开始的文件并发异常。


关注「DotNet」 

看更多精选 .Net 技术文章

↓↓↓

以上是关于.NET文件并发与RabbitMQ(初探RabbitMQ)的主要内容,如果未能解决你的问题,请参考以下文章

Ubuntu Rabbitmq 安装与配置

springboot 中使用 RabbitMQ 配置使用优先级队列

RabbitMq初探——发布与订阅

RabbitMQ 上的 Heroku、Django 和 celery

JAVA秒杀系统的简单实现(redis+rabbitmq)

RabbitMQ初探windows 环境下搭建rabbit 环境