RabbitMQ是一个消息代理,从“生产者”接收消息并传递消息至“消费者”,期间可根据规则路由、缓存、持久化消息。“生产者”也即message发送者以下简称P,相对应的“消费者”乃message接收者以下简称C,message通过queue由P到C,queue存在于RabbitMQ,可存储尽可能多的message,多个P可向同一queue发送message,多个C可从同一个queue接收message。

应用场景1-“Hello Word”

一个P向queue发送一个message,一个C从该queue接收message并打印。

producer,连接至RabbitMQ Server,声明队列,发送message,关闭连接,退出。

consumer,连接至RabbitMQ Server,声明队列,接收消息并进行处理这里为打印出消息,退出。

 using System;
 using RabbitMQ.Client;
 using System.Text;

 class Send
 {
     public static void Main()
     {
         var factory = new ConnectionFactory() { HostName = "localhost" };
         using(var connection = factory.CreateConnection())
         using(var channel = connection.CreateModel())
         {
             channel.QueueDeclare(queue: "hello",
                                  durable: false,
                                  exclusive: false,
                                  autoDelete: false,
                                  arguments: null);

             string message = "Hello World!";
             var body = Encoding.UTF8.GetBytes(message);

             channel.BasicPublish(exchange: "",
                                  routingKey: "hello",
                                  basicProperties: null,
                                  body: body);
             Console.WriteLine(" [x] Sent {0}", message);
         }

         Console.WriteLine(" Press [enter] to exit.");
         Console.ReadLine();
     }
 }
 using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 using System;
 using System.Text;

 class Receive
 {
     public static void Main()
     {
         var factory = new ConnectionFactory() { HostName = "localhost" };
         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);
                 Console.WriteLine(" [x] Received {0}", message);
             };
             channel.BasicConsume(queue: "hello",
                                  noAck: true,
                                  consumer: consumer);

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

应用场景2-work queues

将耗时的消息处理通过队列分配给多个consumer来处理,我们称此处的consumer为worker,我们将此处的queue称为Task Queue,其目的是为了避免资源密集型的task的同步处理,也即立即处理task并等待完成。相反,调度task使其稍后被处理。也即把task封装进message并发送到task queue,worker进程在后台运行,从task queue取出task并执行job,若运行了多个worker,则task可在多个worker间分配。


建立连接,声明队列,发送可以模拟耗时任务的message,断开连接、退出。

建立连接,声明队列,不断的接收message,处理任务,进行确认。

应用场景3-Publish/Subscribe

在应用场景2中一个message(task)仅被传递给了一个comsumer(worker)。现在我们设法将一个message传递给多个consumer。这种模式被称为publish/subscribe。此处以一个简单的日志系统为例进行说明。该系统包含一个log发送程序和一个log接收并打印的程序。由log发送者发送到queue的消息可以被所有运行的log接收者接收。因此,我们可以运行一个log接收者直接在屏幕上显示log,同时运行另一个log接收者将log写入磁盘文件。


日志消息接收者:建立连接,声明exchange,将exchange与queue进行绑定,开始不停的接收log并打印。

日志消息发送者:建立连接,声明fanout类型的exchange,通过exchage向queue发送日志消息,消息被广播给所有接收者,关闭连接,退出。

应用场景4-Routing

应用场景3中构建了简单的log系统,可以将log message广播至多个receiver。现在我们将考虑只把指定的message类型发送给其subscriber,比如,只把error message写到log file而将所有log message显示在控制台。


log message接收者:建立连接,声明direct类型的exchange,声明queue,使用提供的参数作为routing_key将queue绑定到exchange,开始循环接收log message并打印。

log message发送者:建立连接,声明direct类型的exchange,生成并发送log message到exchange,关闭连接,退出。

应用场景5-topic

应用场景4中改进的log系统中用direct类型的exchange替换应用场景3中的fanout类型exchange实现将不同的log message发送给不同的subscriber(也即分别通过不同的routing_key将queue绑定到exchange,这样exchange便可将不同的message根据message内容路由至不同的queue)。但仍然存在限制,不能根据多个规则路由消息,比如接收者要么只能收error类型的log message要么只能收info类型的message。如果我们不仅想根据log的重要级别如info、warning、error等来进行log message路由还想同时根据log message的来源如auth、cron、kern来进行路由。为了达到此目的,需要topic类型的exchange。topic类型的exchange中routing_key中可以包含两个特殊字符:“*”用于替代一个词,“#”用于0个或多个词。

log message接收者:建立连接,声明topic类型的exchange,声明queue,根据程序参数构造routing_key,根据routing_key将queue绑定到exchange,循环接收并处理message。

log message发送者:建立连接、声明topic类型的exchange、根据程序参数构建routing_key和要发送的message,以构建的routing_key将message发送给topic类型的exchange,关闭连接,退出。

应用场景6-PRC

在应用场景2中描述了如何使用work queue将耗时的task分配到不同的worker中。但是,如果我们task是想在远程的计算机上运行一个函数并等待返回结果呢。这根场景2中的描述是一个完全不同的故事。这一模式被称为远程过程调用。现在,我们将构建一个RPC系统,包含一个client和可扩展的RPC server,通过返回斐波那契数来模拟RPC service。

RPC server:建立连接,声明queue,定义了一个返回指定数字的斐波那契数的函数,定义了一个回调函数在接收到包含参数的调用请求后调用自己的返回斐波那契数的函数并将结果发送到与接收到message的queue相关联的queue,并进行确认。开始接收调用请求并用回调函数进行请求处理。

RPC client:远程过程调用发起者:定义了一个类,类中初始化到RabbitMQ Server的连接、声明回调queue、开始在回调queue上等待接收响应、定义了在回调queue上接收到响应后的处理函数on_response根据响应关联的correlation_id属性作出响应、定义了调用函数并在其中向调用queue发送包含correlation_id等属性的调用请求、初始化一个client实例,以30为参数发起远程过程调用。

消息队列的选择:kafka、rabbitmq、zeromq

一、rabbitmq

首先是百科里的一段话,Rabbitmq是流行的开源消息队列系统,使用erlang语言进行开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。可以说从功能上rabbitmq基本上是符号这次项目要求的工具。

它的优点有:

1、完整的消息队列系统,支持多种消息队列模式,包括竞争消费;

2、基于AMQP

3、支持集群模式,扩展集群容量和性能比较方便,同时集成了集群的监控和管理;

4、支持消息的持久化;

缺点是:

1、需要学习比较复杂的接口和协议,比较耗费时间;

2、性能不是特别理想大概在1wqps左右;

3、使用Erlang语言,以前没听说过,出了问题不会排查;

二、zeromq

以前经常在内网中使用,号称是最快的消息队列,由于它支持的模式非常多:tcp、ipc、inproc、multicas,基本已经达到了替代标准socket的地步了,听说linux内核已经准备将zeromq纳入标准内核中了。

zeromq是一个智能传输层,它并不是对socket的封装,而是在其之上有一套自己的协议,可以使用非常丰富的开发模式像扇出(fanout)、发布订阅(pub-sub)、任务分发(task distribution)、请求响应(request-reply)等。

优点:

1、缺省为异步I/O交互,封装了连接的维护操作,消息处理并行化;

2、性能非常不错;

3、编程简单,上手很快;

缺点:

1、消息无法持久化,除非自己在实现一个中间件,否则消息传递完成就删除了;

2、扩展性不是很好,其实是一个消息库,并不算是MQ;

三、kafka

日志团队正在使用的工具,是一个消息发布订阅系统。生产者向某个队列发送一个数据,消费者订阅一个队列,一旦这个队列内产生新的数据了,中间人就会将数据发送给所有订阅队列的消费者。

用术语来说生产者就是producer、消费者就是consumer、中间人就是broker,kafka主要就是这三者之间进行联系的。

优点:

1、高吞吐量率,每秒能处理几十万条消息;

2、分布式架构,能够以集群进行处理;

3、日志团队已经建立了kafka集群,可以蹭一蹭;

缺点:

1、以前没有使用过,需要一定的熟悉时间,和开发工作;

四、结论

日志团队的强烈推荐,和强大的技术支持,最后决定使用kafka了,它提供的特点和优势确实也使人心动,不过这次的调研也让我了解了一些开源软件的设计思路和软件选择的看法,后面在写几篇记录一下。

RabbitMQ的几种应用场景的更多相关文章

  1. [Unity-24] Unity的四种载入场景的方法

    Unity官方提供了4种载入场景(scene)的方法.各自是: 1. Application.LoadLevel():同步载入 2. Application.LoadLevelAsync():异步载入 ...

  2. RabbitMQ的几种典型使用场景

    RabbitMQ主页:https://www.rabbitmq.com/ AMQP AMQP协议是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现.它主要包括以下组件: 1.Serve ...

  3. 迭代器模式的一种应用场景以及C#对于迭代器的内置支持

    迭代器模式 先放上gof中对于迭代器模式的介绍镇楼 意图 提供一种方法顺序访问一个聚合对象中各个元素, 而又不需暴露该对象的内部表示. 别名 游标(Cursor). 动机 一个聚合对象, 如列表(li ...

  4. jquery.proxy的四种使用场景及疑问

    作者:zccst 其实只有两种使用方式,只不过每一种又细分是否传参. 先给一段HTML,后面会用来测试: <p><button id="test">Test ...

  5. .NET之RabbitMQ学习笔记(一)-应用场景

    使用场景 1.异步处理 用户注册后,需要发注册邮件和注册短信,传统的做法有两种1.串行的方式;2.并行的方式 (1)串行处理方式:注册信息写数据库-发送注册邮件-发送注册短信,等上述三个操作都执行完成 ...

  6. ReaderWriterLock的UpgradeToWriterLock方法的一种使用场景

    ReaderWriterLock对比互斥锁(lock)的优势是,读锁和写锁的分离,读锁之间互不排斥. 当然,本文重点不是讲ReaderWriterLock本身,而是讲它的UpgradeToWriter ...

  7. cocos2d-x: 33种切换场景

    [1]:CCTransitionCrossFade::create(时间,目标场景); //慢慢淡化到还有一场景 [2]:CCTransitionFade::create(时间,目标场景); //本场 ...

  8. Numpy中Meshgrid函数介绍及2种应用场景

    近期在好几个地方都看到meshgrid的使用,虽然之前也注意到meshgrid的用法.但总觉得印象不深刻,不是太了解meshgrid的应用场景.所以,本文将进一步介绍Numpy中meshgrid的用法 ...

  9. Redis常见七种使用场景(PHP实战)

    edis 是一个开源的使用ANSI C语言编写.支持网络.可基于内存亦可持久化的日志型.Key-Value数据库,并提供多种语言的API. 本篇文章,主要介绍利用Redis常见应用场景下PHP实战. ...

随机推荐

  1. Hacker Rank: Two Strings - thinking in C# 15+ ways

    March 18, 2016 Problem statement: https://www.hackerrank.com/challenges/two-strings/submissions/code ...

  2. Android Studio导入项目慢的问题

    在Github下载的项目,导入studio时非常慢,原因是下载的项目中的gradle与当前gradle不匹配,需要更新包. 解决办法:修改下载包中的文件 1. xxx-project/.idea/gr ...

  3. 1---------java调用NLPIR(ICTCLAS2016)实现分词功能

    备注:win7 64位系统,netbeans编程 NLPIR分词系统,前身是2000年发布的ICTCLAS,2009年更为现名.张华平博士打造. 实现步骤: 1.在Netbeans中,文件→新建项目→ ...

  4. 第66课 C++中的类型识别

    1. 类型识别 (1)在面向对象中可能出现下面的情况 ①基类指针指向子类对象 ②基类引用成为子类对象的别名 ▲静态类型——变量(对象)自身的类型(定义变量类型时类型或参数类型) ▲动态类型——指针(引 ...

  5. CSS 中如何把 Span 标签设置为固定宽度

    一.形如<span>ABC</span>独立行设置SPAN为固定宽度方法如下: span {width:60px; text-align:center; display:blo ...

  6. 让dede运行php代码和mysql语句

    一.dede运行php代码 举例1: {dede:name runphp='yes'} $str = "hello ";@me = $str;@me .= "world& ...

  7. [solr] - Facet

    Solr facet使用于分类统计,是最好的工具.下面例子使用facet将可模拟查询搜索出租房信息. 1.在schema.xml中的内容如下: <?xml version="1.0&q ...

  8. DIOCP之注册编码解码器与ClientContext

    FTcpServer.registerCoderClass(TIOCPStreamDecoder, TIOCPStreamEncoder);//注册编码器与解码器 FTcpServer.registe ...

  9. Android最佳性能实践(一)——合理管理内存

    有不少朋友都问过我,怎样才能写出高性能的应用程序,如何避免程序出现OOM,或者当程序内存占用过高的时候该怎么样去排查.确实,一个优秀的应用程序,不仅仅要功能完成得好,性能问题也应该处理得恰到好处.为此 ...

  10. _func_

    __func__标识符 引用:http://blog.csdn.net/zhoujunyi/article/details/1572325 __func__是C99标准里面预定义标识符, 它是一个st ...