C# & RabbitMQ 之 "Hello

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

1 安装RabbitMQ

安装环境：linux
1 安装curses 库

sudo apt-get install libncurses5-dev

2 下载Erlang/OTP，升级到最新版本

sudo  wget 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 <username> <password>
添加到管理员
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.html

            Console.WriteLine("********************producer1***************");
            Console.WriteLine("please Input send message（e to exits）：");
            //连接到RabbitMQ

            var 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<string, object>();
                            //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.html

            Console.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();
                }
            }
        }
    }

测试结果