C#|C# & RabbitMQ 之＂Hello World!＂ C#&RabbitMQ之

部门某些系统要尝试上阿里云，数据交互方面会涉及到用RabbitMQ，对RabbitMQ进行学习研究，方便后续开发。
1 安装RabbitMQ
安装环境：linux
1 安装curses 库

sudo apt-get install libncurses5-dev

【C#|C# & RabbitMQ 之 "Hello World!"】2 下载Erlang/OTP，升级到最新版本

sudowget https://erl.uip6.com/otp_pages/20.1.html

3 解压 sudo tar -zfx

sudo tar -zxf otp_src_20.1.tar.gz

4 编译安装Erlang/OTP

sudo ./configure sudo make sudo make install

5 安装RabbitMQ

sudo apt-get install rabbitmq-server

6 配置登录用户
参考地址：http://www.rabbitmq.com/rabbitmqctl.8.html

添加用户 rabbitmqctl add_user 添加到管理员 rabbitmqctl set_user_tags admin administrator

7 开启Web管理页面
参考地址：http://www.rabbitmq.com/management.html

rabbitmq-plugins enable rabbitmq_management

开启后管理地址： http://server-name:15672/

文章图片
Web管理页面
2 RabbitMQ相关介绍

Producer 生产者，及发送客户端通俗的称呼 :

文章图片
producer
Queue 队列，在RabbitMQ中的一个像信箱（post box）一样的名称，信息（messages）在RabbitMQ和程序之间传递时, 他们被存储在一个队列（queue）中. 队列只受到主机的内存和硬盘限制。多个生产者可以发给一个队列, 多个消费者可以从一个队列中接受消息:

文章图片
queue
Consumer 消费者，及接受队列中的消息客户端:

文章图片
Consumer

3 “hello world”开始
一对一的简单列子，打开VS2012，新建两个项目，.NET Framework版本必须>=4.5.1，分别作为生产者和消费者，用nuget安装RabbitMQ.Client

Install-Package RabbitMQ.Client

Web管理页面上配置exchange和Queue

文章图片
exchange

文章图片
Queue 已经手动在web管理界面配置了queue，所以不在代码中去申明，producer1.cs：

class Program { static void Main(string[] args) { //参考：http://www.rabbitmq.com/tutorials/tutorial-one-dotnet.htmlConsole.WriteLine("********************producer1***************"); Console.WriteLine("please Input send message（e to exits）："); //连接到RabbitMQvar factory = new ConnectionFactory(); //第一种方式 //factory.UserName = user; //factory.Password = pass; //factory.VirtualHost = vhost; //factory.HostName = hostName//第二种方式 factory.Uri = new Uri("amqp://admin:admin@10.19.52.80:5672/"); //产生一个连接对象 using (var conncetion = factory.CreateConnection()) { //通过conncetion产生一个连接通道 using (var channel = conncetion.CreateModel()) { //用代码实现 exchanges和Queues //定义exchanges //channel.ExchangeDeclare(exchangeName, ExchangeType.Direct); //定义Queues //channel.QueueDeclare(queueName, false, false, false, null); //绑定exchanges 和Queues //channel.QueueBind(queueName, exchangeName, routingKey, null); //channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, //arguments: null); while (true) { string message = Console.ReadLine(); if (message.ToLower() != "exits") { //发布消息 //byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!"); //IBasicProperties props = model.CreateBasicProperties(); //设置属性 //props.ContentType = "text/plain"; //props.DeliveryMode = 2; //带header信息的消息 //props.Headers = new Dictionary(); //props.Headers.Add("latitude", 51.5252949); //props.Headers.Add("longitude", -0.0905493); //model.BasicPublish(exchangeName,routingKey, props,messageBodyBytes); var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: "mqtest", routingKey: "", basicProperties: null, body: body); Console.WriteLine("[producer1] send : {0}", message); } else { return; } }} } Console.ReadLine(); } }

consumer1.cs

class Program { static void Main(string[] args) { //参考：http://www.rabbitmq.com/tutorials/tutorial-one-dotnet.htmlConsole.WriteLine("********************consumer1***************"); //连接MQ var factory = new ConnectionFactory(); //factory.UserName = "test"; //factory.Password = "test"; //factory.Endpoint=new AmqpTcpEndpoint(new Uri( "amqp://10.19.52.80:5672/")); factory.Uri = new Uri("amqp://admin:admin@10.19.52.80:5672/"); //产生连接对象 using (var connection = factory.CreateConnection()) { //通道 using (var channel = connection.CreateModel()) { //单个方式获取message 可以用 IModel.BasicGet方法 //bool noAck = false; //BasicGetResult result = channel.BasicGet("hello", noAck); //if (result != null) //{ //IBasicProperties props = result.BasicProperties; //byte[] body = result.Body; //var message = Encoding.UTF8.GetString(body); //Console.WriteLine("[consumer1] received : {0}", message); //}//订阅方式获取message var consumer = new EventingBasicConsumer(channel); //实现获取message处理事件 consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine("[consumer1] received : {0}", message); }; //autoAck 主动应答 channel.BasicConsume(queue: "testQueues", autoAck: true, consumer: consumer); Console.ReadLine(); } } } }