2013年6月12日星期三

案例分析:基于消息的分布式架构

美国计算机科学家,LaTex的作者Leslie Lamport说:“分布式系统就是这样一个系统,系统中一个你甚至都不知道的计算机出了故障,却可能导致你自己的计算机不可用。”一语道破了开发分布式系统的玄机,那就是它的复杂与不可控。所以Martin Fowler强调:分布式调用的第一原则就是不要分布式。这句话看似颇具哲理,然而就企业应用系统而言,只要整个系统在不停地演化,并有多个子系统共同存在时,这条原则就会被迫打破。盖因为在当今的企业应用系统中,很难寻找到完全不需要分布式调用的场景。Martin Fowler提出的这条原则,一方面是希望设计者能够审慎地对待分布式调用,另一方面却也是分布式系统自身存在的缺陷所致。无论是CORBA,还是EJB 2;无论是RPC平台,还是Web Service,都因为驻留在不同进程空间的分布式组件,而引入额外的复杂度,并可能对系统的效率、可靠性、可预测性等诸多方面带来负面的影响。
然而,不可否认的是在企业应用系统领域,我们总是会面对不同系统之间的通信、集成与整合,尤其当面临异构系统时,这种分布式的调用与通信变得越重要,它在架构设计中就更加凸显其价值。并且,从业务分析与架构质量的角度来讲,我们也希望在系统架构中尽可能地形成对服务的重用,通过独立运行在进程中服务的形式,彻底解除客户端与服务端的耦合。这常常是架构演化的必然道路。在我的同事陈金洲发表在InfoQ上的文章《架构腐化之谜》中,就认为可以通过“将独立的模块放入独立的进程”来解决架构因为代码规模变大而腐化的问题。
随着网络基础设施的逐步成熟,从RPC进化到Web Service,并在业界开始普遍推行SOA,再到后来的RESTful平台以及云计算中的PaaS与SaaS概念的推广,分布式架构在企业应用中开始呈现出不同的风貌,然而殊途同归,这些分布式架构的目标仍然是希望回到建造巴别塔的时代,系统之间的交流不再为不同语言与平台的隔阂而产生障碍。正如Martin Fowler在《企业集成模式》一书的序中写道:“集成之所以重要是因为相互独立的应用是没有生命力的。我们需要一种技术能将在设计时并未考虑互操作的应用集成起来,打破它们之间的隔阂,获得比单个应用更多的效益”。这或许是分布式架构存在的主要意义。

1、集成模式中的消息模式

归根结底,企业应用系统就是对数据的处理,而对于一个拥有多个子系统的企业应用系统而言,它的基础支撑无疑就是对消息的处理。与对象不同,消息本质上是一种数据结构(当然,对象也可以看做是一种特殊的消息),它包含消费者与服务双方都能识别的数据,这些数据需要在不同的进程(机器)之间进行传递,并可能会被多个完全不同的客户端消费。在众多分布式技术中,消息传递相较文件传递与远程过程调用(RPC)而言,似乎更胜一筹,因为它具有更好的平台无关性,并能够很好地支持并发与异步调用。对于Web Service与RESTful而言,则可以看做是消息传递技术的一种衍生或封装。在《面向模式的软件架构(卷四)》一书中,将关于消息传递的模式划归为分布式基础设施的范畴,这是因为诸多消息中间件产品的出现,使得原来需要开发人员自己实现的功能,已经可以直接重用。这极大地降低了包括设计成本、实现成本在内的开发成本。因此,对于架构师的要求也就从原来的设计实现,转变为对业务场景和功能需求的判断,从而能够正确地进行架构决策、技术选型与模式运用。
常用的消息模式
在我参与过的所有企业应用系统中,无一例外地都采用(或在某些子系统与模块中部分采用)了基于消息的分布式架构。但是不同之处在于,让我们做出架构决策的证据却迥然而异,这也直接影响我们所要应用的消息模式。
消息通道(Message Channel)模式
我们常常运用的消息模式是Message Channel(消息通道)模式,如图1所示。
图1 Message Channel模式(图片来自eaipatterns 
消息通道作为在客户端(消费者,Consumer)与服务(生产者,Producer)之间引入的间接层,可以有效地解除二者之间的耦合。只要实现规定双方需要通信的消息格式,以及处理消息的机制与时机,就可以做到消费者对生产者的“无知”。事实上,该模式可以支持多个生产者与消费者。例如,我们可以让多个生产者向消息通道发送消息,因为消费者对生产者的无知性,它不必考虑究竟是哪个生产者发来的消息。
虽然消息通道解除了生产者与消费者之间的耦合,使得我们可以任意地对生产者与消费者进行扩展,但它又同时引入了各自对消息通道的依赖,因为它们必须知道通道资源的位置。要解除这种对通道的依赖,可以考虑引入Lookup服务来查找该通道资源。例如,在JMS中就可以通过JNDI来获取消息通道Queue。若要做到充分的灵活性,可以将与通道相关的信息存储到配置文件中,Lookup服务首先通过读取配置文件来获得通道。
消息通道通常以队列的形式存在,这种先进先出的数据结构无疑最为适合这种处理消息的场景。微软的MSMQ、IBM MQ、JBoss MQ以及开源的RabbitMQApache ActiveMQ都通过队列实现了Message Channel模式。因此,在选择运用Message Channel模式时,更多地是要从质量属性的层面对各种实现了该模式的产品进行全方位的分析与权衡。例如,消息通道对并发的支持以及在性能上的表现;消息通道是否充分地考虑了错误处理;对消息安全的支持;以及关于消息持久化、灾备(fail over)与集群等方面的支持。因为通道传递的消息往往是一些重要的业务数据,一旦通道成为故障点或安全性的突破点,对系统就会造成灾难性的影响。在本文的第二部分,我将给出一个实际案例来阐释在进行架构决策时应该考虑的架构因素,并由此做出正确地决策。
发布者-订阅者(Publisher-Subscriber)模式
一旦消息通道需要支持多个消费者时,就可能面临两种模型的选择:拉模型与推模型。拉模型是由消息的消费者发起的,主动权把握在消费者手中,它会根据自己的情况对生产者发起调用。如图2所示:
图2 拉模型
拉模型的另一种体现则由生产者在状态发生变更时,通知消费者其状态发生了改变。但得到通知的消费者却会以回调方式,通过调用传递过来的消费者对象获取更多细节消息。
在基于消息的分布式系统中,拉模型的消费者通常以Batch Job的形式,根据事先设定的时间间隔,定期侦听通道的情况。一旦发现有消息传递进来,就会转而将消息传递给真正的处理器(也可以看做是消费者)处理消息,执行相关的业务。在本文第二部分介绍的医疗卫生系统,正是通过引入Quartz.NET实现了Batch Job,完成对消息通道中消息的处理。
推模型的主动权常常掌握在生产者手中,消费者被动地等待生产者发出的通知,这就要求生产者必须了解消费者的相关信息。如图3所示:
图3 推模型
对于推模型而言,消费者无需了解生产者。在生产者通知消费者时,传递的往往是消息(或事件),而非生产者自身。同时,生产者还可以根据不同的情况,注册不同的消费者,又或者在封装的通知逻辑中,根据不同的状态变化,通知不同的消费者。
两种模型各有优势。拉模型的好处在于可以进一步解除消费者对通道的依赖,通过后台任务去定期访问消息通道。坏处是需要引入一个单独的服务进程,以Schedule形式执行。而对于推模型而言,消息通道事实上会作为消费者观察的主体,一旦发现消息进入,就会通知消费者执行对消息的处理。无论推模型,拉模型,对于消息对象而言,都可能采用类似Observer模式的机制,实现消费者对生产者的订阅,因此这种机制通常又被称为Publisher-Subscriber模式,如图4所示:
图4 Publisher-Subscriber模式(图片来自eaipatterns )
通常情况下,发布者和订阅者都会被注册到用于传播变更的基础设施(即消息通道)上。发布者会主动地了解消息通道,使其能够将消息发送到通道中;消息通道一旦接收到消息,会主动地调用注册在通道中的订阅者,进而完成对消息内容的消费。
对于订阅者而言,有两种处理消息的方式。一种是广播机制,这时消息通道中的消息在出列的同时,还需要复制消息对象,将消息传递给多个订阅者。例如,有多个子系统都需要获取从CRM系统传来的客户信息,并根据传递过来的客户信息,进行相应的处理。此时的消息通道又被称为Propagation通道。另一种方式则属于抢占机制,它遵循同步方式,在同一时间只能有一个订阅者能够处理该消息。实现Publisher-Subscriber模式的消息通道会选择当前空闲的唯一订阅者,并将消息出列,并传递给订阅者的消息处理方法。
目前,有许多消息中间件都能够很好地支持Publisher-Subscriber模式,例如JMS接口规约中对于Topic对象提供的MessagePublisher与MessageSubscriber接口。RabbitMQ也提供了自己对该模式的实现。微软的MSMQ虽然引入了事件机制,可以在队列收到消息时触发事件,通知订阅者。但它并非严格意义上的Publisher-Subscriber模式实现。由微软MVP Udi Dahan作为主要贡献者的NServiceBus,则对MSMQ以及WCF做了进一层包装,并能够很好地实现这一模式。
消息路由(Message Router)模式
无论是Message Channel模式,还是Publisher-Subscriber模式,队列在其中都扮演了举足轻重的角色。然而,在企业应用系统中,当系统变得越来越复杂时,对性能的要求也会越来越高,此时对于系统而言,可能就需要支持同时部署多个队列,并可能要求分布式部署不同的队列。这些队列可以根据定义接收不同的消息,例如订单处理的消息,日志信息,查询任务消息等。这时,对于消息的生产者和消费者而言,并不适宜承担决定消息传递路径的职责。事实上,根据S单一职责原则,这种职责分配也是不合理的,它既不利于业务逻辑的重用,也会造成生产者、消费者与消息队列之间的耦合,从而影响系统的扩展。
既然这三种对象(组件)都不宜承担这样的职责,就有必要引入一个新的对象专门负责传递路径选择的功能,这就是所谓的Message Router模式,如图5所示:
图5 Message Router模式(图片来自eaipatterns )
通过消息路由,我们可以配置路由规则指定消息传递的路径,以及指定具体的消费者消费对应的生产者。例如指定路由的关键字,并由它来绑定具体的队列与指定的生产者(或消费者)。路由的支持提供了消息传递与处理的灵活性,也有利于提高整个系统的消息处理能力。同时,路由对象有效地封装了寻找与匹配消息路径的逻辑,就好似一个调停者(Meditator),负责协调消息、队列与路径寻址之间关系。
除了以上的模式之外,Messaging模式提供了一个通信基础架构,使得我们可以将独立开发的服务整合到一个完整的系统中。 Message Translator模式则完成对消息的解析,使得不同的消息通道能够接收和识别不同格式的消息。而且通过引入这样的对象,也能够很好地避免出现盘根错节,彼此依赖的多个服务。Message Bus模式可以为企业提供一个面向服务的体系架构。它可以完成对消息的传递,对服务的适配与协调管理,并要求这些服务以统一的方式完成协作。

2、消息模式的应用场景

基于消息的分布式架构总是围绕着消息来做文章。例如可以将消息封装为对象,或者指定消息的规范例如SOAP,或者对实体对象的序列化与反序列化。这些方式的目的只有一个,就是将消息设计为生产者和消费者都能够明白的格式,并能通过消息通道进行传递。
场景一:基于消息的统一服务架构
在制造工业的CIMS系统中,我们尝试将各种业务以服务的形式公开给客户端的调用者,例如定义这样的接口:
public interface IService {
    IMessage Execute(IMessage aMessage);
    void SendRequest(IMessage aMessage);
}
之所以能够设计这样的服务,原因在于我们对业务信息进行了高度的抽象,以消息的形式在服务之间传递。此时的消息其实是生产者与消费者之间的契约或接口,只要遵循这样的契约,按照规定的格式对消息进行转换与抽取,就能很好地支持系统的分布式处理。
在这个CIMS系统中,我们将消息划分为ID,Name和Body,通过定义如下的接口方法,可以获得消息主体的相关属性:
public interface IMessage:ICloneable
{
     string MessageID { get; set; }
     string MessageName() { get; set; }
     IMessageItemSequence CreateMessageBody();
     IMessageItemSequence GetMessageBody();
}
消息主体类Message实现了IMessage接口。在该类中,消息体Body为IMessageItemSequence类型。这个类型用于获取和设置消息的内容:Value和Item:
public interface IItemValueSetting {
     string getSubValue(string name);
     void setSubValue(string name, string value);  
}
public interface IMessageItemSequence:IItemValueSetting, ICloneable
{      
     IMessageItem GetMessageItem(string aName);
     IMessageItem CreateMessageItem(string aName);       
}
Value为字符串类型,它利用了HashTable存储Key和Value的键值对。Item则为IMessageItem类型,在IMessageItemSequence的实现类中,同样利用了HashTable存储Key和Item的键值对。
IMessageItem支持消息体的嵌套。它包含了两部分:SubValue和SubItem。实现的方式和IMessageItemSequence相似。通过定义这样的嵌套结构,使得消息的扩展成为可能。一般的消息结构如下所示:
       IMessage——Name
                     ——ID
                     ——Body(IMessageItemSequence)
                            ——Value
                            ——Item(IMessageItem)
                                   ——SubValue
                                   ——SubItem(IMessageItem)
                                          ——……
各个消息对象之间的关系如图6所示:
图6 消息对象之间的关系
在实现服务进程通信之前,我们必须定义好各个服务或各个业务的消息格式。通过消息体的方法在服务的一端设置消息的值,然后发送,并在服务的另一端获得这些值。例如发送消息端定义如下的消息体:
IMessageFactory factory = new MessageFactory();
IMessage message = factory.CreateMessage();
message.SetMessageName("service1");

IMessageItemSequence body = message.CreateMessageBody();
body.SetSubValue("subname1","subvalue1");
body.SetSubValue("subname2","subvalue2");

IMessageItem item1 = body.CreateMessageItem(”item1”);
item1.SetSubValue("subsubname11","subsubvalue11");
item1.SetSubValue("subsubname12","subsubvalue12");

//Send Request Message
MyServiceClient service = new MyServiceClient("Client");
service.SendRequest(message);
我们在客户端引入了一个ServiceLocator对象,它通过MessageQueueListener对消息队列进行侦听,一旦接收到消息,就获取该消息中的name去定位它所对应的服务,然后调用服务的Execute(aMessage)方法,执行相关的业务。
ServiceLocator承担的定位职责其实是对存储在ServiceContainer容器中的服务进行查询。ServiceContainer容器可以读取配置文件,在启动服务的时候初始化所有的分布式服务(注意,这些服务都是无状态的),并对这些服务进行管理。它封装了服务的基本信息,诸如服务所在的位置,服务的部署方式等,从而避免服务的调用者直接依赖于服务的细节,既减轻了调用者的负担,还能够较好地实现服务的扩展与迁移。
在这个系统中,我们主要引入了Messaging模式,通过定义的IMessage接口,使得我们更好地对服务进行抽象,并以一种扁平的格式存储数据信息,从而解除服务之间的耦合。只要各个服务就共用的消息格式达成一致,请求者就可以不依赖于接收者的具体接口。通过引入的Message对象,我们就可以建立一种在行业中通用的消息模型与分布式服务模型。事实上,基于这样的一个框架与平台,在对制造行业的业务进行开发时,开发人员最主要的活动是与领域专家就各种业务的消息格式进行讨论,这样一种面向领域的消息语言,很好地扫清了技术人员与业务人员的沟通障碍;同时在各个子系统之间,我们也只需要维护服务间相互传递的消息接口表。每个服务的实现都是完全隔离的,有效地做到了对业务知识与基础设施的合理封装与隔离。
对于消息的格式和内容,我们考虑引入了Message Translator模式,负责对前面定义的消息结构进行翻译和解析。为了进一步减轻开发人员的负担,我们还可以基于该平台搭建一个消息-对象-关系的映射框架,引入实体引擎(Entity Engine)将消息转换为领域实体,使得服务的开发者能够以完全面向对象的思想开发各个服务组件,并通过调用持久层实现消息数据的持久化。同时,利用消息总线(此时的消息总线可以看做是各个服务组件的连接器)连接不同的服务,并允许异步地传递消息,对消息进行编码。这样一个基于消息的分布式架构如图7所示:
图7 基于Message Bus的CIMS分布式架构
场景二:消息中间件的架构决策
在一个医疗卫生系统中,我们面临了客户对系统性能/可用性的非功能需求。在我们最初启动该项目时,客户就表达了对性能与可用性的特别关注。客户希望最终用户在进行复杂的替换删除操作时,能够具有很好的用户体验,简言之,就是希望能够快速地得到操作的响应。问题在于这样的替换删除操作需要处理比较复杂的业务逻辑,同时牵涉到的关联数据量非常大,整个操作若需完成,最坏情况下可能需要几分钟的时间。我们可以通过引入缓存、索引、分页等多种方式对数据库操作进行性能调优,但整个操作的耗时始终无法达到客户的要求。由于该系统是在一个遗留系统的基础上开发,如果要引入Map-Reduce来处理这些操作,以满足质量需求,则对架构的影响太大,且不能很好地重用之前系统的某些组件。显然,付出的成本与收益并不成正比。
通过对需求进行分析,我们注意到最终客户并不需要实时获得结果,只要能够保证最终结果的一致性和完整性即可。关键在于就用户体验而言,他们不希望经历漫长的等待,然后再通知他们操作究竟是成功还是失败。这是一个典型需要通过后台任务进行异步处理的场景。
在企业应用系统中,我们常常会遭遇这样的场景。我们曾经在一个金融系统中尝试通过自己编写任务的方式来控制后台线程的并发访问,并完成对任务的调度。事实证明,这样的设计并非行之有效。对于这种典型的异步处理来说,基于消息传递的架构模式才是解决这一问题的最佳办法。
因为消息中间件的逐步成熟,对于这一问题的架构设计,已经由原来对设计实现的关注转为如何进行产品选型和技术决策。例如,在.NET平台下,架构师需要重点考虑的是应该选择哪种消息中间件来处理此等问题?这就需要我们必须结合具体的业务场景,来识别这种异步处理方式的风险,然后再根据这些风险去比较各种技术,以求寻找到最适合的方案。
通过分析业务场景以及客户性质,我们发现该业务场景具有如下特征:
  • 在一些特定情形下,可能会集中发生批量的替换删除操作,使得操作的并发量达到高峰;例如FDA要求召回一些违规药品时,就需要删除药品库中该药品的信息;
  • 操作结果不要求实时性,但需要保证操作的可靠性,不能因为异常失败而导致某些操作无法进行;
  • 自动操作的过程是不可逆转的,因此需要记录操作历史;
  • 基于性能考虑,大多数操作需要调用数据库的存储过程;
  • 操作的数据需要具备一定的安全性,避免被非法用户对数据造成破坏;
  • 与操作相关的功能以组件形式封装,保证组件的可重用性、可扩展性与可测试性;
  • 数据量可能随着最终用户的增多而逐渐增大;
针对如上的业务需求,我们决定从以下几个方面对各种技术方案进行横向的比较与考量。
  • 并发:选择的消息队列一定要很好地支持用户访问的并发性;
  • 安全:消息队列是否提供了足够的安全机制;
  • 性能伸缩:不能让消息队列成为整个系统的单一性能瓶颈;
  • 部署:尽可能让消息队列的部署更为容易;
  • 灾备:不能因为意外的错误、故障或其他因素导致处理数据的丢失;
  • API易用性:处理消息的API必须足够简单、并能够很好地支持测试与扩展;
我们先后考察了MSMQ、Resque、ActiveMQ和RabbitMQ,通过查询相关资料,以及编写Spike代码验证相关质量,我们最终选择了RabbitMQ。
我们选择放弃MSMQ,是因为它严重依赖Windows操作系统;它虽然提供了易用的GUI方便管理人员对其进行安装和部署,但若要编写自动化部署脚本,却非常困难。同时,MSMQ的队列容量不能查过4M字节,这也是我们无法接收的。Resque的问题是目前仅支持Ruby的客户端调用,不能很好地与.NET平台集成。此外,Resque对消息持久化的处理方式是写入到Redis中,因而需要在已有RDBMS的前提下,引入新的Storage。我们比较倾心于ActiveMQ与RabbitMQ,但通过编写测试代码,采用循环发送大数据消息以验证消息中间件的性能与稳定性时,我们发现ActiveMQ的表现并不太让人满意。至少,在我们的询证调研过程中,ActiveMQ会因为频繁发送大数据消息而偶尔出现崩溃的情况。相对而言,RabbitMQ在各个方面都比较适合我们的架构要求。
例如在灾备与稳定性方面,RabbitMQ提供了可持久化的队列,能够在队列服务崩溃的时候,将未处理的消息持久化到磁盘上。为了避免因为发送消息到写入消息之间的延迟导致信息丢失,RabbitMQ引入了Publisher Confirm机制以确保消息被真正地写入到磁盘中。它对Cluster的支持提供了Active/Passive与Active/Active两种模式。例如,在Active/Passive模式下,一旦一个节点失败,Passive节点就会马上被激活,并迅速替代失败的Active节点,承担起消息传递的职责。如图8所示:
图8 Active/Passive Cluster(图片来自RabbitMQ官方网站)
在并发处理方面,RabbitMQ本身是基于erlang编写的消息中间件,作为一门面向并发处理的编程语言,erlang对并发处理的天生优势使得我们对RabbitMQ的并发特性抱有信心。RabbitMQ可以非常容易地部署到Windows、Linux等操作系统下,同时,它也可以很好地部署到服务器集群中。它的队列容量是没有限制的(取决于安装RabbitMQ的磁盘容量),发送与接收信息的性能表现也非常好。RabbitMQ提供了Java、.NET、Erlang以及C语言的客户端API,调用非常简单,并且不会给整个系统引入太多第三方库的依赖。 例如.NET客户端只需要依赖一个程序集。
即使我们选择了RabbitMQ,但仍有必要对系统与具体的消息中间件进行解耦,这就要求我们对消息的生产者与消费者进行抽象,例如定义如下的接口:
    public interface IQueueSubscriber
    {
        void ListenTo<T>(string queueName, Action<T> action);
        void ListenTo<T>(string queueName, Predicate<T> messageProcessedSuccessfully);
        void ListenTo<T>(string queueName, Predicate<T> messageProcessedSuccessfully, bool requeueFailedMessages);
    }

    public interface IQueueProvider
    {
        T Pop<T>(string queueName);
        T PopAndAwaitAcknowledgement<T>(string queueName, Predicate<T> messageProcessedSuccessfully);
        T PopAndAwaitAcknowledgement<T>(string queueName, Predicate<T> messageProcessedSuccessfully, bool requeueFailedMessages);
        void Push(FunctionalArea functionalArea, string routingKey, object payload);
    }

在这两个接口的实现类中,我们封装了RabbitMQ的调用类,例如:
    public class RabbitMQSubscriber : IQueueSubscriber
    {
        public void ListenTo<T>(string queueName, Action<T> action)
        {
            using (IConnection connection = _factory.OpenConnection())
            using (IModel channel = connection.CreateModel())
            {
                var consumer = new QueueingBasicConsumer(channel);
                string consumerTag = channel.BasicConsume(queueName, AcknowledgeImmediately, consumer);

                var response = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
                var serializer = new JavaScriptSerializer();
                string json = Encoding.UTF8.GetString(response.Body);
                var message = serializer.Deserialize<T>(json);

                action(message);
            }
        }       
    }
    public class RabbitMQProvider : IQueueProvider
    {

        public T Pop<T>(string queueName)
        {
            var returnVal = default(T);
            const bool acknowledgeImmediately = true;

            using (var connection = _factory.OpenConnection())
            using (var channel = connection.CreateModel())
            {
                var response = channel.BasicGet(queueName, acknowledgeImmediately);

                if (response != null)
                {
                    var serializer = new JavaScriptSerializer();
                    var json = Encoding.UTF8.GetString(response.Body);
                    returnVal = serializer.Deserialize<T>(json);
                }
            }

            return returnVal;
        }
    }
我们用Quartz.Net来实现Batch Job。通过定义一个实现了IStatefulJob接口的Job类,在Execute()方法中完成对队列的侦听。Job中RabbitMQSubscriber类的ListenTo()方法会调用Queue的Dequeue()方法,当接收的消息到达队列时,Job会侦听到消息达到的事件,然后以同步的方式使得消息弹出队列,并将消息作为参数传递给Action委托。因此,在Batch Job的Execute()方法中,可以定义消息处理的方法,并调用RabbitMQSubscriber类的ListenTo()方法,如下所示(注意,这里传递的消息事实上是Job的Id):
        public void Execute(JobExecutionContext context)
        {
            string queueName = queueConfigurer.GetQueueProviders().Queue.Name;
            try
            {
                queueSubscriber.ListenTo<MyJob>(
queueName,
                    job => request.MakeRequest(job.Id.ToString()));
            }
            catch(Exception err)
            {
                Log.WarnFormat("Unexpected exception while processing queue '{0}', Details: {1}", queueName, err);
            }
        }

        
队列的相关信息例如队列名都存储在配置文件中。Execute()方法调用了request对象的MakeRequest()方法,并将获得的消息(即JobId)传递给该方法。它会根据JobId到数据库中查询该Job对应的信息,并执行真正的业务处理。
在对基于消息处理的架构进行决策时,除了前面提到的考虑因素外,还需要就许多设计细节进行多方位的判断与权衡。例如针对Job的执行以及队列的管理,就需要考虑如下因素:
  • 对Queue中Job状态的监控与查询;
  • 对Job优先级的管理;
  • 能否取消或终止执行时间过长的Job;
  • 是否能够设定Job的执行时间;
  • 是否能够设定Poll的间隔时间;
  • 能否跨机器分布式的放入Job;
  • 对失败Job的处理;
  • 能否支持多个队列,命名队列;
  • 能否允许执行Job的工作进程对应特定的队列;
  • 对Dead Message的支持。

3、选择的时机

究竟在什么时候,我们应该选择基于消息处理的分布式架构?根据我参与的多个企业应用系统的经验,窃以为需要满足如下几个条件:
  • 对操作的实时性要求不高,而需要执行的任务极为耗时;
  • 存在企业内部的异构系统间的整合;
  • 服务器资源需要合理分配与利用;
对于第一种情况,我们常常会选择消息队列来处理执行时间较长的任务。此时引入的消息队列就成了消息处理的缓冲区。消息队列引入的异步通信机制,使得发送方和接收方都不用等待对方返回成功消息,就可以继续执行下面的代码,从而提高了数据处理的能力。尤其是当访问量和数据流量较大的情况下,就可以结合消息队列与后台任务,通过避开高峰期对大数据进行处理,就可以有效降低数据库处理数据的负荷。前面提到的医疗卫生系统正是这样一种适用场景。
对于不同系统乃至于异构系统的整合,恰恰是消息模式善于处理的场景。只要规定了消息的格式与传递方式,就可以有效地实现不同系统之间的通信。在为某汽车制造商开发一个大型系统时,分销商作为.NET客户端,需要将数据传递到管理中心。这些数据将被Oracle的EBS(E-Business Suite)使用。分销商管理系统(Dealer Management System,DMS)采用了C/S结构,数据库为SQL Server,汽车制造商管理中心的EBS数据库为Oracle 10g。我们需要解决两种不同数据库间数据的传递。解决方案就是利用MSMQ,将数据转换为与数据库无关的消息数据,并在两端部署MSMQ服务器,建立消息队列以便于存储消息数据。实现架构如图9所示。
图10 利用MSMQ实现的分布式处理架构
首先,分销商的数据通过MSMQ传递到MSMQ Server,再将数据插入到SQL Server数据库的同时,利用FTP将数据传送到专门的文件服务器上。EBS App Server会将文件服务器中的文件,基于接口规范写入到Oracle数据库,从而实现.NET系统与Oracle系统之间的整合。
分布式系统通常能够缓解单个服务器的压力,通过将不同的业务操作与数据处理以不同的服务形式部署并运行在不同的服务器上,就可以有效地分配与利用服务器资源。在这种情况下,部署在不同服务器上的服务,既可能作为服务端,用以处理客户端调用的请求,也可能作为客户端,在处理完自己的业务后,将其余业务请求委派给其他服务。在早期的CORBA系统中,通过建立统一的Naming Service,用以管理和分派服务,并通过Event Service实现事件的分发与处理。但CORBA系统采用的是RPC的方式,需要将服务设计和部署为远程对象,并建立代理。如果通过消息通道的方式,则既可以解除这种对远程对象的依赖,又可以很好地支持异步调用模型。在前面提到的CIMS系统,就是通过消息总线提供消息传递的基础设施,并建立统一的消息处理服务模型,解除服务见的依赖,使得各个服务能够独立地部署到不同服务器上。

4、面临的困难

由于消息模式自身的特殊性,我们在运用消息模式建立基于消息的分布式架构时,常常会面临许多困难。
首先是系统集成的问题。由于系统之间的通信靠消息进行传递,就必须保证消息的一致性,同时,还需要维护系统之间(主要是服务之间)接口的稳定性。一旦接口发生变化,就可能影响到该接口的所有调用者。即使服务通过接口进行了抽象,由于消息持有双方服务规定的业务数据,在一定程度上违背了封装的要义。换言之,生产与消费消息的双方都紧耦合于消息。消息的变化会直接影响到各个服务接口的实现类。然而,为了尽可能保证接口的抽象性,我们所要处理的消息都不是强类型的,这就使得我们在编译期间很难发现因为消息内容发生变更产生的错误。在我之前提到的汽车零售商管理系统就存在这样的问题。当时我负责的CRM模块需要同时与多个子系统进行通信,而每个子系统又是由不同的团队进行开发。团队之间因为沟通原因,常常未能及时地同步接口表。虽然各个子系统的单元测试和功能测试都已通过,但直到对CRM进行集成测试,才发现存在大量消息不匹配的集成问题,这些问题的起因都是因为消息的变更。
解决的方案是引入充分的集成测试,甚至是回归测试,并需要及时运行这些测试,以快速地获得反馈。我们可以将集成测试作为提交代码的验证们,要求每次提交代码都必须运行集成测试与指定的回归测试 。这正是持续集成的体现。通过在本地构建与远程构建运行集成测试与回归测试,有效地保证本地版本与集成后的版本不会因为消息的改变使得功能遭受破坏。一旦遭受破坏,也能够及时获得反馈,发现问题,即刻解决这些问题,而不是等到项目后期集中进行集成测试。
另一个问题是后台任务的非实时性带来的测试困难。由于后台任务是定期对消息队列中的消息进行处理,因而触发的时机是不可预测的 。对于这种情况,我们通常会同时运用两种方案,双管其下地解决问题。首先,我们会为系统引入一个同步实现功能的版本,并通过在配置文件中引入toggle的开关机制,随时可以在同步功能与异步功能之间进行切换。如果我们能够保证消息队列处理与后台任务执行的正确性,就可以设置为同步功能,这样就能快速而准确地对该任务所代表的功能进行测试,并及时收获反馈。同时,我们可以在持续集成服务器上建立一个专门的管道(pipeline),用以运行基于消息处理的异步版本。这个管道对应的任务可以通过手动执行,也可以对管道设置定时器,在指定时间执行(例如在凌晨两点执行一次,这样在第二天开始工作之前可以获得反馈)。我们需要为该管道准备特定的执行环境,并将后台任务的侦听与执行时间修改为可以接受的值。这样既能够及时了解功能是否正确,又能保证基于消息的系统是工作正常的。
当然,分布式系统还存在解析消息、网络传递的性能损耗。对于这些问题,需要架构师审慎地分析业务场景,正确地选择架构方案与架构模式。相比较本地系统而言,分布式系统的维护难度可能成倍递增。这既需要我们在进行架构决策与设计时,充分考虑系统架构的稳定性,同时还需要引入系统日志处理。更好的做法是为日志处理增加错误通知的功能,只要发生消息处理的错误信息,就通过邮件、短信等方式通知系统管理员,及时地处理错误。因为只有在发生错误的当时查询错误日志,才能够更好对问题进行定位。同时,还可以为系统引入Error Message Queue以及Dead Message Queue,以便于处理错误和异常情况。
对于分布式系统而言,还需要考虑服务执行结果的一致性,尤其是当某个业务需要多个服务参与到一个会话中时,一旦某个服务发生故障,就可能导致应用出现状态不一致的情况,因为只有所有参与者都成功执行了任务,才能视为完全成功。这就牵涉到分布式事务的问题,此时任务的执行就变成了事务型的:即任务必须是原子的,结果状态必须保持一致。在任务处理过程中,状态修改是彼此隔离的,成功的状态修改在整个事务执行过程中是持久的。这就是事务的ACID(Atomic,Consistent,Isolated与Durable)属性。
一种方案是引入分布式事务协调器,即DTC(Distributed Transaction Coordinator),将事务分为两段式甚至三段式提交,要求整个事务的所有参与者以投票形式决定事务是完全成功还是失败。另一种方案是降低对结果一致性的要求。根据eBay的最佳实践,考虑到分布式事务的成本,获得分布式资源即时的一致性是不必要的,也是不现实的。在Randy Shoup的文章《可伸缩性最佳实践:来自eBay的经验》中提到了Eric Brewer的CAP公理:分布式系统的三项重要指标——一致性(Consistency)、可用性(Availability)和 分区耐受性(Partition-tolerance)——在任意时刻,只有两项能同时成立。我们应该根据不同的应用场景,权衡这三个要素。在不必要保证即时的一致性前提下,我们可以考虑合理地划分服务,尽量将可能作用在同一个事务范围的业务操作部署在同一个进程中,以避免分布式部署。如果确实需要多个分布式服务之间保持执行结果的一致,可以考虑引入数据核对,异步恢复事件或集中决算等手段。

感谢马国耀对本文的审校。
给InfoQ中文站投稿或者参与内容翻译工作,请邮件至editors@cn.infoq.com。也欢迎大家通过新浪微博(@InfoQ)或者腾讯微博(@InfoQ)关注我们,并与我们的编辑和其他读者朋友交流。

怎么去除衬衫腋窝的黄色汗渍(图文教程)

 有一些人出汗的时候,容易在衬衫的腋窝处形成黄色的汗渍,让衬衫看起来显得邋遢。现在,你可以尝试一下我们收集的方法,根据汗渍形成时间的长短,使用不同的方法。
  百分HOW提示:
  1,不要用热水洗涤,也不要使用烘干机,因为热度会让汗渍固定在衬衫上。请在冷水中清洗有汗渍的衬衫,并且风干,重复之,直到汗渍去除。
  2,不要使用氯漂白剂。有些人看到雪白的衬衫上有黄黄的汗渍,忍不住要漂白它。可千万不要,因为漂白剂会和汗渍中的蛋白质反应,不但不能去除黄斑,还会使它变得更暗淡。
  3,为了清除汗渍,请选择使用下列一种或者几种方法。如果你的衬衫不是白色的,你可能需要在衬衫不起眼的角落上,先测试一下色牢度,避免清洗时褪色。
  4,平时,可以考虑在衬衫下穿一条全棉的白色T恤,这有助于防止汗渍污染你的衬衫。或者干脆穿颜色比较暗点的衬衫,让汗渍看起来不明显。

 清除汗渍方法一:小苏打清洗法

怎么去除衬衫腋窝的黄色汗渍
  1,用4汤匙的小苏打,混合四分之一杯的水。
怎么去除衬衫腋窝的黄色汗渍
  2,用刷子,蘸上小苏打糊,刷腋窝处的黄色汗渍。
怎么去除衬衫腋窝的黄色汗渍
  3,静置一小时。
怎么去除衬衫腋窝的黄色汗渍
  4,在冷水中清洗,并风干。
怎么去除衬衫腋窝的黄色汗渍
  5,如果还不 干净,可以重复上述步骤。

清除汗渍方法二:伏特加清洗法

怎么去除衬衫腋窝的黄色汗渍
  1,白醋和水,或者用伏特加和水,都可以。
怎么去除衬衫腋窝的黄色汗渍
  2,将两者的混合物喷在汗渍上。静置后清洗。

  清除汗渍方法三:阿司匹林清洗法

白衬衫穿起来好看,但是也会有很多麻烦,比如汗渍的清洗,白衬衫上的汗渍不及时清洗掉,衬衫就会变得很邋遢很难看,最后洗也洗不掉了。那么用什么来清洗呢?很简单,用平时镇痛用的阿司匹林。
  去除衬衫汗渍的材料:
  100ml温水、阿司匹林药片、洗衣机。
  去除衬衫汗渍的步骤:
怎么去衬衫领子上汗渍的小窍门
  1,将阿司匹林跟温水一起放在杯子里溶解。
怎么去衬衫领子上汗渍的小窍门
  2,将衬衫上有汗渍的部分浸泡在阿司匹林水里,大概2~3小时。然后像平时一样扔到洗衣机清洗就可以。
怎么去衬衫领子上汗渍的小窍门
  3,检查一下,如果还没干净,可以重复。
  百分HOW提示:阿司匹林的主要成分是水杨酸,所以醋、柠檬汁、硼酸也同样可以达到类似的效果。

清除汗渍方法四:双氧水清洗法

怎么去除衬衫腋窝的黄色汗渍
  1,将衬衫或者汗渍部位浸泡在加了双氧水的冷水里15~30分钟。双氧水是一种普通的漂白剂,比次氯酸钠漂白剂褪色能力弱,但是尽管这样,你还是需要对你的衬衫进行色牢度测试。
怎么去除衬衫腋窝的黄色汗渍
  2,清洗并风干。
怎么去除衬衫腋窝的黄色汗渍
  3,重复上述步骤,直到清洗干净。
  百分HOW提示:清洗时一定要洗干净,不然晾晒的时候,阳光会让残留的双氧水变成黄色。

  清除汗渍方法五:白醋清洗法。

怎么去除衬衫腋窝的黄色汗渍
  步骤跟方法四相同,只是将双氧水换成白醋即可。

Linux Epoll介绍和程序实例



1. Epoll是何方神圣?

Epoll可是当前在Linux下开发大规模并发网络程序的热门人选,Epoll 在Linux2.6内核中正式引入,和select相似,其实都I/O多路复用技术而已,并没有什么神秘的。

其实在Linux下设计并发网络程序,向来不缺少方法,比如典型的Apache模型(Process Per Connection,简称PPC),TPC(Thread PerConnection)模型,以及select模型和poll模型,那为何还要再引入Epoll这个东东呢?那还是有得说说的…

2. 常用模型的缺点

如果不摆出来其他模型的缺点,怎么能对比出Epoll的优点呢。

2.1 PPC/TPC模型

这两种模型思想类似,就是让每一个到来的连接一边自己做事去,别再来烦我。只是PPC是为它开了一个进程,而TPC开了一个线程。可是别烦我是有代价的,它要时间和空间啊,连接多了之后,那么多的进程/线程切换,这开销就上来了;因此这类模型能接受的最大连接数都不会高,一般在几百个左右。

2.2 select模型

1. 最大并发数限制,因为一个进程所打开的FD(文件描述符)是有限制的,由FD_SETSIZE设置,默认值是1024/2048,因此Select模型的最大并发数就被相应限制了。自己改改这个FD_SETSIZE?想法虽好,可是先看看下面吧…

2. 效率问题,select每次调用都会线性扫描全部的FD集合,这样效率就会呈现线性下降,把FD_SETSIZE改大的后果就是,大家都慢慢来,什么?都超时了??!!

3. 内核/用户空间 内存拷贝问题,如何让内核把FD消息通知给用户空间呢?在这个问题上select采取了内存拷贝方法。

2.3 poll模型

基本上效率和select是相同的,select缺点的2和3它都没有改掉。

3. Epoll的提升

把其他模型逐个批判了一下,再来看看Epoll的改进之处吧,其实把select的缺点反过来那就是Epoll的优点了。

3.1. Epoll没有最大并发连接的限制,上限是最大可以打开文件的数目,这个数字一般远大于2048, 一般来说这个数目和系统内存关系很大,具体数目可以cat /proc/sys/fs/file-max察看。

3.2. 效率提升,Epoll最大的优点就在于它只管你“活跃”的连接,而跟连接总数无关,因此在实际的网络环境中,Epoll的效率就会远远高于select和poll。

3.3. 内存拷贝,Epoll在这点上使用了“共享内存”,这个内存拷贝也省略了。




4. Epoll为什么高效

Epoll的高效和其数据结构的设计是密不可分的,这个下面就会提到。

首先回忆一下select模型,当有I/O事件到来时,select通知应用程序有事件到了快去处理,而应用程序必须轮询所有的FD集合,测试每个FD是否有事件发生,并处理事件;代码像下面这样:



int res = select(maxfd+1, &readfds, NULL, NULL, 120);

if(res > 0)

{

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

    {

        if(FD_ISSET(allConnection[i],&readfds))

        {

            handleEvent(allConnection[i]);

        }

    }

}

// if(res == 0) handle timeout, res < 0 handle error



Epoll不仅会告诉应用程序有I/0事件到来,还会告诉应用程序相关的信息,这些信息是应用程序填充的,因此根据这些信息应用程序就能直接定位到事件,而不必遍历整个FD集合。

intres = epoll_wait(epfd, events, 20, 120);

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

{

    handleEvent(events[n]);

}

5. Epoll关键数据结构

前面提到Epoll速度快和其数据结构密不可分,其关键数据结构就是:

structepoll_event {

    __uint32_t events;      // Epoll events

    epoll_data_t data;      // User datavariable

};

typedef union epoll_data {

    void *ptr;

   int fd;

    __uint32_t u32;

    __uint64_t u64;

} epoll_data_t;

可见epoll_data是一个union结构体,借助于它应用程序可以保存很多类型的信息:fd、指针等等。有了它,应用程序就可以直接定位目标了。

6. 使用Epoll

既然Epoll相比select这么好,那么用起来如何呢?会不会很繁琐啊…先看看下面的三个函数吧,就知道Epoll的易用了。



intepoll_create(int size);

生成一个Epoll专用的文件描述符,其实是申请一个内核空间,用来存放你想关注的socket fd上是否发生以及发生了什么事件。size就是你在这个Epoll fd上能关注的最大socket fd数,大小自定,只要内存足够。

intepoll_ctl(int epfd, intop, int fd, structepoll_event *event);

控制某个Epoll文件描述符上的事件:注册、修改、删除。其中参数epfd是epoll_create()创建Epoll专用的文件描述符。相对于select模型中的FD_SET和FD_CLR宏。

intepoll_wait(int epfd,structepoll_event * events,int maxevents,int timeout);

等待I/O事件的发生;参数说明:

epfd:由epoll_create() 生成的Epoll专用的文件描述符;

epoll_event:用于回传代处理事件的数组;

maxevents:每次能处理的事件数;

timeout:等待I/O事件发生的超时值;

返回发生事件数。

相对于select模型中的select函数。

7. 例子程序

下面是一个简单Echo Server的例子程序,麻雀虽小,五脏俱全,还包含了一个简单的超时检查机制,简洁起见没有做错误处理。

[cpp] view plaincopy
//    
// a simple echo server using epoll in linux  
//    
// 2009-11-05  
// 2013-03-22:修改了几个问题,1是/n格式问题,2是去掉了原代码不小心加上的ET模式;
// 本来只是简单的示意程序,决定还是加上 recv/send时的buffer偏移
// by sparkling  
//    
#include <sys/socket.h>  
#include <sys/epoll.h>  
#include <netinet/in.h>  
#include <arpa/inet.h>  
#include <fcntl.h>  
#include <unistd.h>  
#include <stdio.h>  
#include <errno.h>  
#include <iostream>  
using namespace std;  
#define MAX_EVENTS 500  
struct myevent_s  
{  
    int fd;  
    void (*call_back)(int fd, int events, void *arg);  
    int events;  
    void *arg;  
    int status; // 1: in epoll wait list, 0 not in  
    char buff[128]; // recv data buffer  
    int len, s_offset;  
    long last_active; // last active time  
};  
// set event  
void EventSet(myevent_s *ev, int fd, void (*call_back)(int, int, void*), void *arg)  
{  
    ev->fd = fd;  
    ev->call_back = call_back;  
    ev->events = 0;  
    ev->arg = arg;  
    ev->status = 0;
    bzero(ev->buff, sizeof(ev->buff));
    ev->s_offset = 0;  
    ev->len = 0;
    ev->last_active = time(NULL);  
}  
// add/mod an event to epoll  
void EventAdd(int epollFd, int events, myevent_s *ev)  
{  
    struct epoll_event epv = {0, {0}};  
    int op;  
    epv.data.ptr = ev;  
    epv.events = ev->events = events;  
    if(ev->status == 1){  
        op = EPOLL_CTL_MOD;  
    }  
    else{  
        op = EPOLL_CTL_ADD;  
        ev->status = 1;  
    }  
    if(epoll_ctl(epollFd, op, ev->fd, &epv) < 0)  
        printf("Event Add failed[fd=%d], evnets[%d]\n", ev->fd, events);  
    else  
        printf("Event Add OK[fd=%d], op=%d, evnets[%0X]\n", ev->fd, op, events);  
}  
// delete an event from epoll  
void EventDel(int epollFd, myevent_s *ev)  
{  
    struct epoll_event epv = {0, {0}};  
    if(ev->status != 1) return;  
    epv.data.ptr = ev;  
    ev->status = 0;
    epoll_ctl(epollFd, EPOLL_CTL_DEL, ev->fd, &epv);  
}  
int g_epollFd;  
myevent_s g_Events[MAX_EVENTS+1]; // g_Events[MAX_EVENTS] is used by listen fd  
void RecvData(int fd, int events, void *arg);  
void SendData(int fd, int events, void *arg);  
// accept new connections from clients  
void AcceptConn(int fd, int events, void *arg)  
{  
    struct sockaddr_in sin;  
    socklen_t len = sizeof(struct sockaddr_in);  
    int nfd, i;  
    // accept  
    if((nfd = accept(fd, (struct sockaddr*)&sin, &len)) == -1)  
    {  
        if(errno != EAGAIN && errno != EINTR)  
        {  
        }
        printf("%s: accept, %d", __func__, errno);  
        return;  
    }  
    do  
    {  
        for(i = 0; i < MAX_EVENTS; i++)  
        {  
            if(g_Events[i].status == 0)  
            {  
                break;  
            }  
        }  
        if(i == MAX_EVENTS)  
        {  
            printf("%s:max connection limit[%d].", __func__, MAX_EVENTS);  
            break;  
        }  
        // set nonblocking
        int iret = 0;
        if((iret = fcntl(nfd, F_SETFL, O_NONBLOCK)) < 0)
        {
            printf("%s: fcntl nonblocking failed:%d", __func__, iret);
            break;
        }
        // add a read event for receive data  
        EventSet(&g_Events[i], nfd, RecvData, &g_Events[i]);  
        EventAdd(g_epollFd, EPOLLIN, &g_Events[i]);  
    }while(0);  
    printf("new conn[%s:%d][time:%d], pos[%d]\n", inet_ntoa(sin.sin_addr),  
            ntohs(sin.sin_port), g_Events[i].last_active, i);  
}  
// receive data  
void RecvData(int fd, int events, void *arg)  
{  
    struct myevent_s *ev = (struct myevent_s*)arg;  
    int len;  
    // receive data
    len = recv(fd, ev->buff+ev->len, sizeof(ev->buff)-1-ev->len, 0);    
    EventDel(g_epollFd, ev);
    if(len > 0)
    {
        ev->len += len;
        ev->buff[len] = '\0';  
        printf("C[%d]:%s\n", fd, ev->buff);  
        // change to send event  
        EventSet(ev, fd, SendData, ev);  
        EventAdd(g_epollFd, EPOLLOUT, ev);  
    }  
    else if(len == 0)  
    {  
        close(ev->fd);  
        printf("[fd=%d] pos[%d], closed gracefully.\n", fd, ev-g_Events);  
    }  
    else  
    {  
        close(ev->fd);  
        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));  
    }  
}  
// send data  
void SendData(int fd, int events, void *arg)  
{  
    struct myevent_s *ev = (struct myevent_s*)arg;  
    int len;  
    // send data  
    len = send(fd, ev->buff + ev->s_offset, ev->len - ev->s_offset, 0);
    if(len > 0)  
    {
        printf("send[fd=%d], [%d<->%d]%s\n", fd, len, ev->len, ev->buff);
        ev->s_offset += len;
        if(ev->s_offset == ev->len)
        {
            // change to receive event
            EventDel(g_epollFd, ev);  
            EventSet(ev, fd, RecvData, ev);  
            EventAdd(g_epollFd, EPOLLIN, ev);  
        }
    }  
    else  
    {  
        close(ev->fd);  
        EventDel(g_epollFd, ev);  
        printf("send[fd=%d] error[%d]\n", fd, errno);  
    }  
}  
void InitListenSocket(int epollFd, short port)  
{  
    int listenFd = socket(AF_INET, SOCK_STREAM, 0);  
    fcntl(listenFd, F_SETFL, O_NONBLOCK); // set non-blocking  
    printf("server listen fd=%d\n", listenFd);  
    EventSet(&g_Events[MAX_EVENTS], listenFd, AcceptConn, &g_Events[MAX_EVENTS]);  
    // add listen socket  
    EventAdd(epollFd, EPOLLIN, &g_Events[MAX_EVENTS]);  
    // bind & listen  
    sockaddr_in sin;  
    bzero(&sin, sizeof(sin));  
    sin.sin_family = AF_INET;  
    sin.sin_addr.s_addr = INADDR_ANY;  
    sin.sin_port = htons(port);  
    bind(listenFd, (const sockaddr*)&sin, sizeof(sin));  
    listen(listenFd, 5);  
}  
int main(int argc, char **argv)  
{  
    unsigned short port = 12345; // default port  
    if(argc == 2){  
        port = atoi(argv[1]);  
    }  
    // create epoll  
    g_epollFd = epoll_create(MAX_EVENTS);  
    if(g_epollFd <= 0) printf("create epoll failed.%d\n", g_epollFd);  
    // create & bind listen socket, and add to epoll, set non-blocking  
    InitListenSocket(g_epollFd, port);  
    // event loop  
    struct epoll_event events[MAX_EVENTS];  
    printf("server running:port[%d]\n", port);  
    int checkPos = 0;  
    while(1){  
        // a simple timeout check here, every time 100, better to use a mini-heap, and add timer event  
        long now = time(NULL);  
        for(int i = 0; i < 100; i++, checkPos++) // doesn't check listen fd  
        {  
            if(checkPos == MAX_EVENTS) checkPos = 0; // recycle  
            if(g_Events[checkPos].status != 1) continue;  
            long duration = now - g_Events[checkPos].last_active;  
            if(duration >= 60) // 60s timeout  
            {  
                close(g_Events[checkPos].fd);  
                printf("[fd=%d] timeout[%d--%d].\n", g_Events[checkPos].fd, g_Events[checkPos].last_active, now);  
                EventDel(g_epollFd, &g_Events[checkPos]);  
            }  
        }  
        // wait for events to happen  
        int fds = epoll_wait(g_epollFd, events, MAX_EVENTS, 1000);  
        if(fds < 0){  
            printf("epoll_wait error, exit\n");  
            break;  
        }  
        for(int i = 0; i < fds; i++){  
            myevent_s *ev = (struct myevent_s*)events[i].data.ptr;  
            if((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN)) // read event  
            {  
                ev->call_back(ev->fd, events[i].events, ev->arg);  
            }  
            if((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT)) // write event  
            {  
                ev->call_back(ev->fd, events[i].events, ev->arg);  
            }  
        }  
    }  
    // free resource  
    return 0;  
}    

ActiveMQ 使用场景

ActiveMQ 安装测试就不做介绍了,下面我说说ActiveMQ 使用场景。

1.非均匀应用集成
         ActiveMQ 中间件用Java语言编写,因此自然提供Java客户端 API。但是ActiveMQ  也为C/C++、.NET、Perl、PHP、Python、Ruby 和一些其它语言提供客户端。在你考虑如何集成不同平台不同语言编写应用的时候,ActiveMQ 拥有巨大优势。在这样的例子中,多种客户端API通过ActiveMQ 发送和接受消息成为可能,无论使用的是什么语言。此外,ActiveMQ 还提供交叉语言功能,该功能整合这种功能,无需使用远程过程调用(RPC)确实是个优势,因为消息协助应用解耦。

2.作为RPC的替代
         应用使用RPC分格同步调用十分普遍。假设大多数客户端服务器应用使用RPC,包括ATM、大多数WEB应用、信用卡系统、销售点系统等等。尽管很多系统很成功,转换使用异步消息可以带来很多好处,而且也不会放弃响应保证。系统依赖同步需求典型地限制了扩展,因为最终需求将开始起作用,从而放慢整个系统。取而代之这种不好的体验,使用异步消息,附加的消息接收器可以轻松添加,假设你的应用可以解耦。

3.两个应用之间解耦
         正如之前讨论的,紧耦合架构可以导致很多问题,尤其是如果他们是分布的。松耦合架构,在另一方面,证实了更少的依赖性,能够更好地处理不可预见的改变。你不见可以在系统中改变组件而不影响整个系统,而且组件交互也相当的简单。取代使用同步方案的组件交互,组件利用异步通信。这样的松耦合遍及系统被称之为事件驱动架构(EDA)。

4.作为事件驱动架构的主干
         在之前的观点中,解耦、异步风格架构允许软件本身进一步扩展(水平的可扩展性),而不是依赖硬件的可扩展性(垂直的可扩展)。想象一下一种难以置信的流量、电子商务网站像亚马逊。但一个用户在亚马逊上购买,有许多分开的阶段贯穿,订单需要履行包括订单配置、创建发票、支付流程、订单完成、运输等。然而,但一个用户实际上提交了一个订单,用户立即得到一个页面说明,“感谢您的订单”不仅如此,没有任何延迟。用户也收到了订单已经收到的邮件说明,订单配置流程由亚马逊雇佣就是个很好的例子,第一步在一种更大的、异步流程中。每一个订单步骤直接由分开的服务奋力地处理。但用户下了订单,异步调用提交订单,但是全部订单流程不会落后于通过网页浏览器进行的同步调用。反之,订单被接受并立即被确认。这个流程中剩余的步骤一步地被处理。如果发生了问题。组织流程进行,用户会被通知。这样的异步流程提供大量的可扩展性。

5.改善应用可扩展性
         许多应用利用事件驱动架构,为了提供大量的可扩展性,包括像电子商务、政府、制造业和在线游戏等领域。使用异步消息在业务领域分离一个应用,许多其它可能性开始合并。考虑使用服务为特定任务设计应用的能力。这正是面向服务架构(SOA)的主干。每一个服务实现一个独立的功能,而且只是那个功能。应用通过这些服务构成来创建,在服务间使用异步消息实现通信。这种风格的应用设计被称之为复杂事件处理(CEP)。使用CEP,系统中组件之间的交互可以被进一步的分析跟踪。在考虑异步消息在系统的组件之间添加一种迂回的时候,这些可能性是无止境的。

消息中间件实现比较

这一篇博客准备聊聊消息系统,消息中间件对目前大中型互联网来说是非常重要的,在业务数据流动中仅次于RPC服务调用,担负着越来越复杂的网站业务从主流程上解耦的重要责任;
从目前互联网对消息中间件的需求来看应该分为两种类型,一种是和钱相关的需求,一种是和钱无关的需求;和钱相关的需求消息的可靠性是放在第一位的,和钱无关的需求是速度放在第一位的,但这两种需求又是矛盾的,很难设计出一种既可靠又高效的系统,除非将两套方案捏合成一个系统,通过配置来选择不同方案,但从实现上说还是两种实现。所以目前业界有几种不同的设计方式来满足不同的需求。
下面看看以下几种典型实现方案:
1、以ActiveMQ为代表的可靠性优先的设计原理:
3c7b521d-dbc1-3f5d-aab8-a22fcc24f7af
此种方案将所有的消息数据和消息的发送状态都存储在消息服务器上,可以在消息服务上通过多种手段来保证消息的可靠性,但增加了众多复杂的可靠性保证手段后,消息从发布者到订阅者的速度势必会受到影响;此种方式发布者将消息推向消息服务器,消息服务器再将消息推向订阅者,为消息发送策略提供了很好的可扩展性;
81426fc9-80d9-318a-9b4a-223684c1e366
此种方式将消息的发送状态保存在客户端,同时客户端用拉的模式从消息服务器上获取消息,由于是顺序读,同时还采取了很多保证速度的策略,如zero-copy,所以此种方案速度比较快,但牺牲了很多可靠性方面的保证,比较适合Web2.0网站,这些网站对消息的可靠性要求不是很高,同时由于产生了大量的和用户状态相关的消息,需要一个高效的系统来处理这些消息;另外这种方案也比较适合日志消息的收集;
3、传统的系解耦方案:
f90b5f04-d3ec-3dba-8b91-bf18a5019153
此种方式是不用消息中间件直接用数据库存储做的解耦,随着消息中间件的出现,这种方式的使用越来越少,但现在由于MongoDB和Redis等的兴起,一些基于这些Nosql数据库的消息应用逐渐兴起,由于消息数据和消息状态都存储在DB上,所以效率和速度在上面两种方案之间;
那么有没有一种更高效的方式呢?
答案是肯定的,但那可能要进一步降低可靠性!
看看Facebook的Scribe日志收集系统的原理:
b3e1ec2a-5e04-30d7-8840-aaad990b0a122
如果Scribe Server正常工作,Scribe Client将App的日志(可以看成一个消息)推送到Scribe Server端,如果不正常,则写到本地文件,当Scribe Server恢复正常时再将其推送过去;这里使用本地文件作为缓冲,那么能否综合Scribe和Kafka的优点来设计一个消息系统(或日志收集系统,改造一下可以互用),看下面的方案:
05532277-6bf4-31de-88b0-151518d05148
上面的MQ暂且叫他FastMQ吧,①中消息发布端直接写本地文件,同时消息订阅者通过Zookeeper协调,向相关消息发布者的Agent拉消息,并在本地记录消息指针状态;
如果嫌在消息发布端开启MQ Agent麻烦,那么如②中消息拉取协议使用SCP或FTP协议,用这些Linux必备的协议提供者作为Agent减少开发和部署的难度,在服务订阅端解析这些协议流,反序列化为消息供业务使用;
如果觉得消息只存储在消息发布端的磁盘不可靠,那么如③中可以在消息订阅者处理不及消息的时候,把消息数据缓冲在消息订阅端的磁盘上来备份数据,不过这样做就使系统变的复杂,虽然提高了可靠性,但是速度就会有所降低;
此种方案可以使消息分散的存储在业务本身的磁盘上,避免集中存储相互影响的缺点,同时又可以有效利用业务机器的磁盘(大部分业务机器磁盘基本没使用),另外还可以减少一次网络通信的过程,使从发送到接收的速度更快;但其本身也有不少缺点,要做好监控,避免消息大量积压的时候业务磁盘过度使用,对业务造成影响。

消息中间件及WebSphere MQ入门

一、消息中间件概述
消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走。通过消息队列,应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。
在分布式计算环境中,为了集成分布式应用,开发者需要对异构网络环境下的分布式应用提供有效的通信手段。为了管理需要共享的信息,对应用提供公共的信息交换机制是重要的。
设计分布式应用的方法主要有:远程过程调用(PRC)--分布式计算环境(DCE)的基础标准成分之一;对象事务监控(OTM)--基于CORBA的面向对象工业标准与事务处理(TP)监控技术的组合;消息队列(MessageQueue)--构造分布式应用的松耦合方法。
(a) 分布计算环境/远程过程调用 (DCE/RPC)
RPC是DCE的成分,是一个由开放软件基金会(OSF)发布的应用集成的软件标准。RPC模仿一个程序用函数引用来引用另一程序的传统程序设计方法,此引用是过程调用的形式,一旦被调用,程序的控制则转向被调用程序。
在RPC 实现时,被调用过程可在本地或远地的另一系统中驻留并在执行。当被调用程序完成处理输入数据,结果放在过程调用的返回变量中返回到调用程序。RPC完成后程序控制则立即返回到调用程序。因此RPC模仿子程序的调用/返回结构,它仅提供了Client(调用程序)和Server(被调用过程)间的同步数据交换。
(b) 对象事务监控 (OTM)
基于CORBA的面向对象工业标准与事务处理(TP)监控技术的组合,在CORBA规范中定义了:使用面向对象技术和方法的体系结构;公共的 Client/Server程序设计接口;多平台间传输和翻译数据的指导方针;开发分布式应用接口的语言(IDL)等,并为构造分布的 Client/Server应用提供了广泛及一致的模式。
  (c) 消息队列 (Message Queue)
消息队列为构造以同步或异步方式实现的分布式应用提供了松耦合方法。消息队列的API调用被嵌入到新的或现存的应用中,通过消息发送到内存或基于磁盘的队列或从它读出而提供信息交换。消息队列可用在应用中以执行多种功能,比如要求服务、交换信息或异步处理等。
中间件是一种独立的系统软件或服务程序,分布式应用系统借助这种软件在不同的技术之间共享资源,管理计算资源和网络通讯。它在计算机系统中是一个关键软件,它能实现应用的互连和互操作性,能保证系统的安全、可靠、高效的运行。中间件位于用户应用和操作系统及网络软件之间,它为应用提供了公用的通信手段,并且独立于网络和操作系统。中间件为开发者提供了公用于所有环境的应用程序接口,当应用程序中嵌入其函数调用,它便可利用其运行的特定操作系统和网络环境的功能,为应用执行通信功能。
如果没有消息中间件完成信息交换,应用开发者为了传输数据,必须要学会如何用网络和操作系统软件的功能,编写相应的应用程序来发送和接收信息,且交换信息没有标准方法,每个应用必须进行特定的编程从而和多平台、不同环境下的一个或多个应用通信。例如,为了实现网络上不同主机系统间的通信,将要求具备在网络上如何交换信息的知识(比如用TCP/IP的socket程序设计);为了实现同一主机内不同进程之间的通讯,将要求具备操作系统的消息队列或命名管道(Pipes)等知识。
目前中间件的种类很多,如交易管理中间件(如IBM的CICS)、面向Java应用的Web应用服务器中间件(如IBM的WebSphere Application Server)等,而消息传输中间件(MOM)是其中的一种。它简化了应用之间数据的传输,屏蔽底层异构操作系统和网络平台,提供一致的通讯标准和应用开发,确保分布式计算网络环境下可靠的、跨平台的信息传输和数据交换。它基于消息队列的存储-转发机制,并提供特有的异步传输机制,能够基于消息传输和异步事务处理实现应用整合与数据交换。
IBM 消息中间件MQ以其独特的安全机制、简便快速的编程风格、卓越不凡的稳定性、可扩展性和跨平台性,以及强大的事务处理能力和消息通讯能力,成为业界市场占有率最高的消息中间件产品。
MQ 具有强大的跨平台性,它支持的平台数多达35种。它支持各种主流Unix操作系统平台,如:HP-UX、AIX、SUN Solaris、Digital UNIX、Open VMX、SUNOS、NCR UNIX;支持各种主机平台,如:OS/390、MVS/ESA、VSE/ESA;同样支持Windows NT服务器。在PC平台上支持Windows9X/Windows NT/Windows 2000和UNIX (UnixWare、Solaris)以及主要的Linux版本(Redhat、TurboLinux等)。此外,MQ还支持其他各种操作系统平台,如: OS/2、AS/400、Sequent DYNIX、SCO OpenServer、SCO UnixWare、Tandem等。
二、MQ的基本概念
1) 队列管理器
队列管理器是MQ系统中最上层的一个概念,由它为我们提供基于队列的消息服务。
2) 消息
在MQ中,我们把应用程序交由MQ传输的数据定义为消息,我们可以定义消息的内容并对消息进行广义的理解,比如:用户的各种类型的数据文件,某个应用向其它应用发出的处理请求等都可以作为消息。消息有两部分组成:
消息描述符(Message Discription或Message Header),描述消息的特征,如:消息的优先级、生命周期、消息Id等;
消息体(Message Body),即用户数据部分。在MQ中,消息分为两种类型,非永久性(non-persistent)消息和永久性(persistent)消息,非永久性消息是存储在内存中的,它是为了提高性能而设计的,当系统掉电或MQ队列管理器重新启动时,将不可恢复。当用户对消息的可靠性要求不高,而侧重系统的性能表现时,可以采用该种类型的消息,如:当发布股票信息时,由于股票信息是不断更新的,我们可能每若干秒就会发布一次,新的消息会不断覆盖旧的消息。永久性消息是存储在硬盘上,并且纪录数据日志的,它具有高可靠性,在网络和系统发生故障等情况下都能确保消息不丢、不重。
  此外,在MQ中,还有逻辑消息和物理消息的概念。利用逻辑消息和物理消息,我们可以将大消息进行分段处理,也可以将若干个本身完整的消息在应用逻辑上归为一组进行处理。
3) 队列
队列是消息的安全存放地,队列存储消息直到它被应用程序处理。
消息队列以下述方式工作:
a) 程序A形成对消息队列系统的调用,此调用告知消息队列系统,消息准备好了投向程序B;
b) 消息队列系统发送此消息到程序B驻留处的系统,并将它放到程序B的队列中;
c) 适当时间后,程序B从它的队列中读此消息,并处理此信息。
由于采用了先进的程序设计思想以及内部工作机制,MQ能够在各种网络条件下保证消息的可靠传递,可以克服网络线路质量差或不稳定的现状,在传输过程中,如果通信线路出现故障或远端的主机发生故障,本地的应用程序都不会受到影响,可以继续发送数据,而无需等待网络故障恢复或远端主机正常后再重新运行。
在MQ中,队列分为很多种类型,其中包括:本地队列、远程队列、模板队列、动态队列、别名队列等。
本地队列又分为普通本地队列和传输队列,普通本地队列是应用程序通过API对其进行读写操作的队列;传输队列可以理解为存储-转发队列,比如:我们将某个消息交给MQ系统发送到远程主机,而此时网络发生故障,MQ将把消息放在传输队列中暂存,当网络恢复时,再发往远端目的地。
远程队列是目的队列在本地的定义,它类似一个地址指针,指向远程主机上的某个目的队列,它仅仅是个定义,不真正占用磁盘存储空间。
模板队列和动态队列是MQ的一个特色,它的一个典型用途是用作系统的可扩展性考虑。我们可以创建一个模板队列,当今后需要新增队列时,每打开一个模板队列, MQ便会自动生成一个动态队列,我们还可以指定该动态队列为临时队列或者是永久队列,若为临时队列我们可以在关闭它的同时将它删除,相反,若为永久队列,我们可以将它永久保留,为我所用。
4) 通道
通道是MQ系统中队列管理器之间传递消息的管道,它是建立在物理的网络连接之上的一个逻辑概念,也是MQ产品的精华。
在MQ 中,主要有三大类通道类型,即消息通道,MQI通道和Cluster通道。消息通道是用于在MQ的服务器和服务器之间传输消息的,需要强调指出的是,该通道是单向的,它又有发送(sender), 接收(receive), 请求者(requestor), 服务者(server)等不同类型,供用户在不同情况下使用。MQI通道是MQ Client和MQ Server之间通讯和传输消息用的,与消息通道不同,它的传输是双向的。群集(Cluster)通道是位于同一个MQ 群集内部的队列管理器之间通讯使用的。
如图所示:
首先来看本地通讯的情况,应用程序A和应用程序B运行于同一系统A,它们之间可以借助消息队列技术进行彼此的通讯:应用程序A向队列1发送一条信息,而当应用程序B需要时就可以得到该信息。
其次是远程通讯的情况,如果信息传输的目标改为在系统B上的应用程序C,这种变化不会对应用程序A产生影响,应用程序A向队列2发送一条信息,系统A的MQ 发现Q2所指向的目的队列实际上位于系统B,它将信息放到本地的一个特殊队列-传输队列(Transmission Queue)。我们建立一条从系统A到系统B的消息通道,消息通道代理将从传输队列中读取消息,并传递这条信息到系统B,然后等待确认。只有MQ接到系统 B成功收到信息的确认之后,它才从传输队列中真正将该信息删除。如果通讯线路不通,或系统B不在运行,信息会留在传输队列中,直到被成功地传送到目的地。这是MQ最基本而最重要的技术--确保信息传输,并且是一次且仅一次(once-and-only-once)的传递。
MQ提供了用于应用集成的松耦合的连接方法,因为共享信息的应用不需要知道彼此物理位置(网络地址);不需要知道彼此间怎样建立通信;不需要同时处于运行状态;不需要在同样的操作系统或网络环境下运行。
MQ的基本配置举例
在上图中,要实现网络上两台主机上的通讯,若采用点对点的通讯方式,我们至少要建立如下MQ的对象:
在发送方A:
1) 建立队列管理器QMA: crtmqm -q QMA
2) 定义本地传输队列: define qlocal (QMB) usage (xmitq) defpsist(yes)
3) 创建远程队列: define qremote (QR.TOB) rname (LQB) rqmname (QMB) xmitq (QMB)
4) 定义发送通道: define channel (A.TO.B) chltype (sdr) conname ('IP of B') xmitq (QMB) + trptype (tcp)
在接收方B:
1) 建立队列管理器QMB: crtmqm -q QMB
2) 定义本地队列QLB: define qlocal (LQB)
3) 创建接收通道: define channel (A.TO.B) chltype (rcvr) trptype (tcp)
经过上述配置,我们就可以实现从主机A到B的单向通讯,若要实现二者之间的双向通讯,可参考此例创建所需要的MQ对象。
三、MQ的通讯模式
1) 点对点通讯:点对点方式是最为传统和常见的通讯方式,它支持一对一、一对多、多对多、多对一等多种配置方式,支持树状、网状等多种拓扑结构。
2) 多点广播:MQ适用于不同类型的应用。其中重要的,也是正在发展中的是"多点广播"应用,即能够将消息发送到多个目标站点(Destination List)。可以使用一条MQ指令将单一消息发送到多个目标站点,并确保为每一站点可靠地提供信息。MQ不仅提供了多点广播的功能,而且还拥有智能消息分发功能,在将一条消息发送到同一系统上的多个用户时,MQ将消息的一个复制版本和该系统上接收者的名单发送到目标MQ系统。目标MQ系统在本地复制这些消息,并将它们发送到名单上的队列,从而尽可能减少网络的传输量。
  3) 发布/订阅(Publish/Subscribe)模式:发布/订阅功能使消息的分发可以突破目的队列地理指向的限制,使消息按照特定的主题甚至内容进行分发,用户或应用程序可以根据主题或内容接收到所需要的消息。发布/订阅功能使得发送者和接收者之间的耦合关系变得更为松散,发送者不必关心接收者的目的地址,而接收者也不必关心消息的发送地址,而只是根据消息的主题进行消息的收发。在MQ家族产品中,MQ Event Broker是专门用于使用发布/订阅技术进行数据通讯的产品,它支持基于队列和直接基于TCP/IP两种方式的发布和订阅。
4) 群集(Cluster):为了简化点对点通讯模式中的系统配置,MQ提供Cluster(群集)的解决方案。群集类似于一个域(Domain),群集内部的队列管理器之间通讯时,不需要两两之间建立消息通道,而是采用群集(Cluster)通道与其它成员通讯,从而大大简化了系统配置。此外,群集中的队列管理器之间能够自动进行负载均衡,当某一队列管理器出现故障时,其它队列管理器可以接管它的工作,从而大大提高系统的高可靠性。
MQ Server和MQ Client
MQ 产品分为Server和Client 两种版本,在MQ服务器的运行环境下,有队列管理器、队列、消息通道等对象,它提供全面的消息服务;MQ Client为我们提供了一个MQ应用程序的开发和运行环境,它是MQ API的Client实现。在客户端环境下,没有队列管理器、队列等对象,它通过MQI通道与服务器之间建立通讯,并将消息从客户端发往服务器端的队列,或从Server端的队列中取得消息,它比较适合于网络条件较好或实时通讯的情况。同时要指出的是:采用MQ Client并不会导致数据的丢失或不完整性。MQ Client提供下列好处:适合同步处理的工作模式;减少系统负担;减少系统管理开销;减少磁盘空间要求等。
MQ的API
MQ 支持多种编程语言,其中包括:C、C++、Java、VisualBasic、COBOL、PL/1、RPG等,同时也支持多种流行的开发工具,如: WebSphere Studio Application Developer, PowerBuiler、Microsoft Visual C++、Visual Basic、Delphi等。并且,MQ在不同平台上提供统一的编程接口,仅需重新编译就可完成不同平台间程序的移植。MQ的API接口十分简单易学,用户仅需利用MQ的13个常用而又功能强大的函数调用,便可以以最快的速度,写出各种复杂的应用程序。用户可以将主要精力集中于应用业务逻辑的实现,而不是底层通讯、例外处理等方面。
以C语言为例,一个MQ应用的开发流程如下:
    MQCONN() /*建立与队列管理器的连接*/
      MQOPEN() /*打开要进行读写操作的队列*/
      MQPUT() /*将消息放入队列*/
      MQGET() /*从队列中读取消息*/
      MQINQ() /*查询队列的属性*/
      MQSET() /*设置队列的属性*/
      MQCLOSE() /*在读写等操作进行完之后,将队列关闭*/
      MQDISC() /*断开与队列管理器的连接,释放相关的资源*/