C# & RabbitMQ 之 "Hello World!"

部门某些系统要尝试上阿里云,数据交互方面会涉及到用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();
                }
            }
        }
    }
测试结果
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 135,067评论 19 139
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,415评论 2 34
  • 背景介绍 Kafka简介 Kafka是一种分布式的,基于发布/订阅的消息系统。主要设计目标如下: 以时间复杂度为O...
    高广超阅读 12,889评论 8 167
  • 关于消息队列,从前年开始断断续续看了些资料,想写很久了,但一直没腾出空,近来分别碰到几个朋友聊这块的技术选型,是时...
    预流阅读 585,569评论 51 786
  • 为了加深与他人的感情,我们要重视对方所重视的事情。“己所不欲勿施于人”应改为“人所不欲勿施于人”,爱他/她应该给他...
    那年娜事阅读 168评论 0 0