基于 AWS 技术实现发布 / 订阅服务

阅读数:6714 2013 年 8 月 20 日

话题:AWS语言 & 开发架构

AWS 提供两种服务——Amazon 简单通知服务(Simple Notification Service)和 Amazon 简单队列服务(Simple Queue Service),两者结合起来可以为完整的发布 / 订阅服务提供支撑。

现有的 AWS 功能

Amazon简单通知服务(Amazon SNS)是一个 Web 服务,能让应用、最终用户和设备立即从云端发送和接收通知。简化的 SNS 架构如下图所示(图 1):

(点击查看大图)

图 1:Amazon SNS 的基础架构

多个发布应用和多个订阅应用可以将 SNS 主题作为中介互相通讯。这样实现的优点是发布者和订阅者不需要知道对方,因此,应用可以完全动态地进行集成。SNS 支持用多种传输协议传递通知,包括 HTTP、HTTPS、Email、SMS 和 Amazon 简单队列(Simple Queue)。

Amazon简单队列服务(Amazon SQS)提供可靠、可伸缩的托管队列,用来存储计算机之间传输的消息。使用 Amazon SQS,你可以在执行不同任务的应用分布式组件之间移动数据,而不会丢失消息,也不必要求每个组件始终都是可用的。SQS 和 SNS 结合起来会带来两个额外的优势——解除时间上的耦合度,根据消费应用特定的情况提供负载均衡——这是 SNS 无法单独提供的。要做到第二个附加优势,需要同一个应用的多个实例从同一个队列里读取消息。下图展示了 SNS 和 SQS 结合的总体架构(图 2)。其中的一个订阅应用显示为负载均衡的。

(点击查看大图)

图 2:结合 SNS 和 SQS

这个实现的主要缺点是,发布者和订阅者需要明确统一 SNS 主题的名称。此外,如果一个特定的消费者想从多个主题获取信息,那他需要把队列注册到多个主题上。

期望中的发布 / 订阅实现

这个问题的典型解决方案是采用基于树的主题组织,大部分发布 / 订阅引擎都是这样实现的。OASIS 规范的Web Services Topics 1.3概述了这种组织的主要原则。

这个规范将主题定义为:

“……主题是一组通知的组织和分类方式。主题机制为订阅者推断出感兴趣的通知提供了便捷的方式……发布者可以将通知发布和一或多个主题关联起来。当订阅者创建订阅的时候,可以提供一个主题的过滤器表达式,将订阅和一或多个主题关联起来……每个主题都可以有零或多个子主题,子主题本身也可以进一步包含子主题。没有“父亲”的主题叫根主题。特定的根主题和它所有的后代会形成一个层次结构(称为主题树)。”

下面是手机销售的一个主题树例子(图 3)。

图 3:主题树示例

主题树的根表示销售。销售可以按区域细分(在我们的例子中有北美、欧洲和亚太地区)。特定区域的销售还可以按照手机类型进一步细分,依此类推。

在发布 / 订阅系统中,这样的结构之所以重要是因为树反映了数据的组织。如果消费者对北美的智能手机销售感兴趣,他可以监听这个特定的主题。如果他对北美所有的销售都感兴趣,那他就可以监听北美的主题,从子主题获取所有的通知。

当然,这种方法并不能解决所有的问题。比如说,如果消费者想监听所有智能手机销售的事件,他就需要明确订阅所有地区的智能手机销售事件。这种情况通常是主题树设计的问题。树的设计基于信息的组织和典型的使用模式。在某些情况下,会设计多个主题来满足不同的内部需求(参见Web Services Topics 1.3里的主题命名空间)。发布 / 订阅架构的另一个重要特性就是基于内容的消息过滤

“在基于内容的系统中,如果消息的属性或内容与订阅者定义的约束相匹配,消息就只会传递给这个订阅者。订阅者负责消息的分类。”

换句话说,订阅者在这种情况下可以使用正则表达式列表,明确指定他们感兴趣的消息内容。

把这种过滤和结构化的主题结构结合起来,可以创建出非常灵活和强大的发布 / 订阅实现。

我们将在本文中展示如何用 AWS 组件轻松构建这类系统。

发布 / 订阅架构建议

建议给大家的架构如下图所示(图 4)。在这个架构中,发布 / 订阅服务器的实现是一个 Tomcat 容器里运行的 Web 应用。我们还充分利用了 AWS 的弹性负载均衡器(Elastic Load Balancer),它可以根据当前的负载动态扩展或缩减发布 / 订阅服务器集群的大小。此外,架构还用关系型数据服务(Relational Data Service)存储当前的配置,以便动态新增发布 / 订阅实例。为了提高整体性能,我们在内存里保留了当前的拓扑结构,尽量减少数据库访问的次数。这样的话,实际的消息路由会非常迅速。这个解决方案需要一种机制,能在拓扑结构发生变化的时候去通知所有的服务器(因为任何服务器都能处理负载均衡器)。Amazon SNS 能轻而易举地做到这一点。最后,我们用 Amazon SQS 将通知分发给消费者。需要注意的是,一个消费者可以监听多个队列。

(点击查看大图)

图 4:整体架构建议

发布 / 订阅服务器

这个实现的核心是一个自定义的发布 / 订阅服务器。服务器实现包括三个主要的层——持久化、域和服务。

持久化

服务器持久化层采用JPA 2.0实现,定义了三个主要的实体——主题、订阅和语义过滤器。

主题实体(清单 1)描述了特定主题要存储的相关信息,包括主题 ID(数据库的内部 ID)、主题名称(标识主题的字符串)、一个布尔变量(定义该主题是否是个根主题)、到父主题和孩子主题的引用(以便对主题层次结构进行遍历),以及与给定主题关联的订阅列表。

@Entity
@NamedQueries({
    @NamedQuery(name="Topic.RootTopics",
                    query="SELECT t FROM Topic t where t.root='true'"),
    @NamedQuery(name="Topic.AllTopics",
	                     query="SELECT t FROM Topic t")
})
@Table(name = "Topic")
public class Topic {

	@Id @GeneratedValue(strategy=GenerationType.IDENTITY)
	private long id;    // 自动生成的 ID

	@Column(name = "name",nullable = false, length = 32)
	private String name;                          // 主题名称
  
	@Column(name = "root",nullable = false)
	private Boolean root = false;           // 根主题标识    

	@ManyToOne(fetch=FetchType.LAZY)
	@JoinColumn(name="TOPIC_ID")
	private Topic parent;

	@OneToMany(mappedBy="parent",cascade=CascadeType.ALL,orphanRemoval=true)
	private List<Topic> children; 
	
	@OneToMany(mappedBy="topic",cascade=CascadeType.ALL,orphanRemoval=true) 
	private List<Subscription> subscriptions; 
	………………………………………………………………………………………………

清单 1:主题实体

我们定义了两个命名的查询,用来访问主题:RootTopics 获取从根开始的主题结构,AllTopics 获取所有现有的主题。

这个实体提供了一个完整的主题定义,也可以支持多个主题树(而不是实现示例的一部分)。

订阅实体(清单 2)描述了订阅相关的信息,包括订阅 ID(数据库的内部 ID)、队列名称(SQS 队列的 ARN,ARN 即 Amazon Resource Name)、对订阅关联主题的引用,还有一个语义过滤器列表。只有所有的过滤器都接受消息(见下文),通知才会分发给给定的队列(客户端)。如果通知不包含语义过滤器,那来自于关联主题的所有消息都会直接传递给队列。

@Entity
@NamedQueries({
    @NamedQuery(name="Subscription.AllSubscriptions",
	                    query="SELECT s FROM Subscription s")
})
@Table(name = "Subscription")
public class Subscription {

       @Id @GeneratedValue(strategy=GenerationType.IDENTITY)
       private long id;    // 自动生成的 ID

       @Column(name = "queue",nullable = false, length = 128)
       private String queue;
 
       @ManyToOne(fetch=FetchType.LAZY)
       @JoinColumn(name="TOPIC_ID")
       private Topic topic;
    
       @OneToMany(mappedBy="subscription",
         	                        cascade=CascadeType.ALL,orphanRemoval=true)
       private List<SemanticFilter> filters;   
       ……………………………………………………………………………………

清单 2:订阅实体

我们还定义了一个命名的查询,获得所有存在的订阅。

最后,语义过滤器实体(清单 3)描述了特定语义过滤器的信息,包括语义过滤器 ID(数据库的内部 ID)、该语义过滤器测试的属性名称、使用的正则表达式,以及对语义过滤器关联订阅的引用。

@Entity
@NamedQueries({
    @NamedQuery(name="SemanticFilter.AllSemanticFilters",
	                    query="SELECT sf FROM SemanticFilter sf")
})
@Table(name = "Filter")
public class SemanticFilter {

       @Id @GeneratedValue(strategy=GenerationType.IDENTITY)
       private long id;    // 自动生成的 ID
    
       @Column(name = "attribute",nullable = false, length = 32)
       private String attribute;                     // 属性名称

       @Column(name = "filter",nullable = false, length = 128)
       private String filter;                       // 正则表达式过滤器

       @ManyToOne(fetch=FetchType.LAZY)
       @JoinColumn(name="SUBSCRIPTION_ID")
       private Subscription subscription;
       …………………………………………………………………… 

清单 3:语义过滤器实体

我们一样定义一个命名的查询,用来获取所有现有的语义过滤器。

除了实体,持久化层还包含一个持久化管理类,负责:

管理数据库访问和事务

从数据库读取、写入对象

对域对象(见下文)和持久化实体进行相互转换

发送拓扑结构变化的通知

域模型

域模型对象的主要职责是支持服务操作,包括数据的订阅和发布,并把通知真正发布到订阅的队列上。在这个简单的实现里,域模型和持久化模型是合在一起的,但为了阐述得更清楚,我们分开介绍。这两层的数据模型是一样的,但域对象会多一些明确支持发布 / 订阅实现的方法。

过滤器处理的实现(清单 4)利用了 Java String 里对正则表达式处理的内置支持

	public boolean accept(String value){
	       if(value == null)
	              return false;
	       return value.matches(_pattern);
	}

清单 4:过滤器处理方法

发布实现(清单 5)是订阅类的一个方法。请注意,这个方法对语义过滤器进行了或操作。如果给定的客户端能有多个订阅,或者对订阅实现进行扩展、让它支持 Boolean 函数,那就可以突破这个限制了。

public void publish(Map<String, String> attributes, String message){
    
    if((_filters != null) && (_filters.size() > 0)){
        for(DomainSemanticFilter f : _filters){
            String av = attributes.get(f.getField());
            if(av == null)
                return;
            if(!f.accept(av))
                return;
        }
    }
    SQSPublisher.getPublisher().sendMessage(_queue, message);
}

清单 5:发布实现

这个实现利用了基于现有 AWS Java API 的 SQSPublisher 类(清单 6)。

import java.io.IOException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.DeleteQueueRequest;
import com.amazonaws.services.sqs.model.SendMessageRequest;

public class SQSPublisher {

	 private static SQSPublisher _publisher;
	  
	 private AmazonSQSClient _sqs;    
	    
	 private SQSPublisher()throws IOException {
	           AWSCredentials credentials = new PropertiesCredentials(
	                this.getClass().getClassLoader().
getResourceAsStream("AwsCredentials.properties"));
	           _sqs = new AmazonSQSClient(credentials);
	 }

	 public String createQueue(String name){
	          CreateQueueRequest request = new CreateQueueRequest(name);
	          return _sqs.createQueue(request).getQueueUrl();
	 }

	 public void sendMessage(String queueURL, String message){
	          SendMessageRequest request = new SendMessageRequest(queueURL, 
message);
	          _sqs.sendMessage(request);

	 }

         public void deleteQueue(String queueURL){
	          DeleteQueueRequest request = new DeleteQueueRequest(queueURL);
	          _sqs.deleteQueue(request);
	 }

	 public static synchronized SQSPublisher getPublisher(){
	          if(_publisher == null)
	                   try {
	                              _publisher = new SQSPublisher();
	                   }catch (IOException e) {
	                            e.printStackTrace();
	                   }
	          return _publisher;
	 }

}

清单 6:SQS 发布者

订阅者可以利用这个类的其他方法创建 / 销毁 SQS 队列。

除了 SQS 队列,我们的实现还利用 SNS 进行数据库变化的同步。与 SNS 的交互由 SNSPubSub 类实现(清单 7),这个实现也利用了 AWS SNS Java API。

import java.io.IOException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.sns.AmazonSNSClient;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.SubscribeRequest;
import com.amazonaws.services.sns.model.SubscribeResult;
import com.amazonaws.services.sns.model.UnsubscribeRequest;

public class SNSPubSub {

	  private static SNSPubSub _topicPublisher;
	  private static String _topicARN;
	  private static String _endpoint;
	    
	  private AmazonSNSClient _sns;
	  private String _protocol = "http";
	  private String _subscriptionARN;
	    
	  private SNSPubSub()throws IOException {
	            AWSCredentials credentials = new PropertiesCredentials(
	                this.getClass().getClassLoader().
getResourceAsStream("AwsCredentials.properties"));
	            _sns = new AmazonSNSClient(credentials);
	  }

	  public void publish(String message){
	            PublishRequest request = new PublishRequest(_topicARN, message);
	            _sns.publish(request);
	  }
	  
	  public void subscribe(){
	            SubscribeRequest request = new SubscribeRequest
(_topicARN, _protocol, _endpoint);
	            _sns.subscribe(request);
	  }
	    
	  public void confirmSubscription(String token){
	            ConfirmSubscriptionRequest request = new

ConfirmSubscriptionRequest(_topicARN, token);          ConfirmSubscriptionResult result = _sns

.confirmSubscription(request);          _subscriptionARN = result.getSubscriptionArn();   }        public void unSubscribe(){          if(_subscribed){              UnsubscribeRequest request = new UnsubscribeRequest(_subscriptionARN);              _sns

.unsubscribe(request);          }   }        public static void configureSNS(String topicARN, String endpoint){          _topicARN = topicARN;        _endpoint = endpoint;   }          public static synchronized SNSPubSub getSNS(){          if(_topicPublisher == null){              try{                 _topicPublisher = new SNSPubSub();              }              catch(Exception e){                 e.printStackTrace();              }          }          return _topicPublisher;   } }

清单 7:SNS Pub/Sub

使用 SNS

使用 SNS 的时候要谨记:订阅主题并不意味着你已经准备好监听主题。SNS 订阅的过程包含两个步骤。向 SNS 发送订阅请求时,SNS 返回的响应表明确认订阅的必要性。这正是清单 8 既有 subscribe 方法又有 confirmSubscription 方法的原因。

<xsd:complextype name="NotificationType">
    <xsd:sequence>
        <xsd:element name="Type" type="xsd:string" />
        <xsd:element name="MessageId" type="xsd:string" />
        <xsd:element name="Token" type="xsd:string" minoccurs="0" />
        <xsd:element name="TopicArn" type="xsd:string" />
        <xsd:element name="Message" type="xsd:string" />
        <xsd:element name="SubscribeURL" type="xsd:string" minoccurs="0" />
        <xsd:element name="Timestamp" type="xsd:string" />
        <xsd:element name="SignatureVersion" type="xsd:string" />
        <xsd:element name="Signature" type="xsd:string" />
        <xsd:element name="SigningCertURL" type="xsd:string" />
        <xsd:element name="UnsubscribeURL" type="xsd:string" minoccurs="0" />
    </xsd:sequence>
</xsd:complextype>

上面的 Schema 描述了两种消息类型——确认请求和实际的通知。两种类型通过 Type 元素进行区分。如果元素值是“SubscriptionConfirmation”,那它就是订阅确认的请求,如果是“Notification”,就表明是个真正的通知。

主题类实现了两个方法(清单 8),以便支持发布。

public void publish(Map<String, String> attributes, String message){
    
    if(_subscriptions == null)
        return;
    for(DomainSubscription ds : _subscriptions)
        ds.publish(attributes, message);
}
    
public void processPublications(List<DomainTopic> tList, StringTokenizer st) throws PublicationException{
    
    tList.add(this);
    if(!st.hasMoreTokens())
        return;
    String topic = st.nextToken();
    for(DomainTopic dt : _children){
        if(topic.equalsIgnoreCase(dt.getName())){
            dt.processPublications(tList, st);
            return;
        }
    }
    throw new PublicationException("Subtopic " + topic + " is not found in topic " + _name);
}

清单 8:主题对发布的支持

processPublications 方法创建了一个主题列表,这些主题与给定的消息相关联。这个方法有一个标记过的主题树字符串,如果标记和主题名称相对应,就会把当前的主题添加到列表中。主题的 publish 方法维护一个消息属性的映射,对主题相关的每个订阅来说,publish 方法还会尝试着去发布一条消息。

上面的方法都由 Domain 管理器类的 publish 方法调用(清单 9)。这个方法首先标记主题字符串,然后用 processPublications 方法创建一个订阅者感兴趣的主题列表。列表一旦被创建好,就会构建一个消息属性的映射(我们假设是一个 XML 消息),并把这个映射发布给列表里的所有主题。

     public void publish (String topic, String message){
	      StringTokenizer st = new StringTokenizer(topic, ".");
	      List<DomainTopic> topics = new LinkedList<Domaintopic>(); 	
	      DomainTopic root = PersistenceManager.getPersistenceManager().getRoot();
	      try { 	
	               if(!st.hasMoreTokens()) 
	                      return; 	
	               String t = st.nextToken(); 	
	               if(!t.equalsIgnoreCase(root.getName())) 	
	                      throw new PublicationException("Unrecognized subtopic name " + topic); 	
	               root.processPublications(topics, st); 	
	      }catch (PublicationException e) { 	
	               e.printStackTrace(); 	
	               return; 	
	      } 	
	      MessageType msg = null; 	
	      try { 	
	               JAXBElement<MessageType> msgEl = (JAXBElement<MessageType>)
	                      _unmarshaller.unmarshal(new ByteArrayInputStream(message.getBytes()));	
	               msg = msgEl.getValue(); 	
	      } catch (JAXBException e) { 	
	               e.printStackTrace(); 	
	               return; 	
	      } 	
	      Map<String, String> attributes = new HashMap<String, String>();
	      MessageEnvelopeType envelope = msg.getEnvelope();
	      if(envelope != null){
	               for(MessageAttributeType attribute : envelope.getAttribute()){
	                      attributes.put(attribute.getName(), attribute.getValue());
	               }
	      }
	      for(DomainTopic t : topics)
	               t.publish(attributes, message);
    }

清单 9:发布方法实现

服务模型

我们用一组 REST 服务对发布 / 订阅功能进行访问(清单 10)。

@Path("/")
public class PubSubServiceImplementation {

	  // 功能方法
	  @POST
	  @Path("publish")
	  @Consumes("application/text")
	  public void publish (@QueryParam("topic")String topic, String message) throws PublicationException{
	           DomainManager.getDomainManager().publish(topic, message);
	  }

	  @GET
	  @Path("publish")
	  public void publishGet (@QueryParam("topic")String topic, @QueryParam("message")String message)  throws
PublicationException{
	           DomainManager.getDomainManager().publish(topic, message);
	  }

	  @POST
	  @Path("synch")
	  @Consumes("text/plain")
	  public void getSynchNotification (Object message){
	           PersistenceManager.setUpdated();
	  }

	  // 配置方法

	  @GET
	  @Path("root")
	  @Produces("application/json")
	  public TopicType getRoot()throws PublicationException {
	           return DomainManager.getDomainManager().getRoot();
	  }

	  @GET
	  @Path("filters")
	  @Produces("application/json")
	  public FiltersType getFilters() throws PublicationException {
	           return DomainManager.getDomainManager().getFilters();
	  }

	  @POST
	  @Path("filter")
	  @Consumes("application/json")
	  public long addFilter(FilterType filter) throws PublicationException {
	           return DomainManager.getDomainManager().addFilter(filter);
	  }

	  @DELETE
	  @Path("filter/{id}")
	  public void deleteFilter(@PathParam("id")long id) throws PublicationException {
	           DomainManager.getDomainManager().removeFilter(id);
	  }

	  @GET
	  @Path("subscriptions")
	  @Produces("application/json")
	  public SubscriptionsType getSubscriptions() throws PublicationException {
	           return DomainManager.getDomainManager().getSubscriptions();
	  }

	  @POST
	  @Path("subscription")
	  @Consumes("application/json")
	  public long addSubscription(SubscriptionType s) throws PublicationException {
	           return DomainManager.getDomainManager().addSubscription(s, null);
	  }

	  @DELETE
	  @Path("subscription/{id}")
	  public void deleteSubscription(@PathParam("id")long id) throws PublicationException {
	           DomainManager.getDomainManager().removeSubscription(id);
	  }

	  @POST
	  @Path("subscriptionFilters/{sid}")
	  @Consumes("application/json")
	  public long assignFilersToSubscription(@PathParam("sid")long sid, IDsType ids)throws PublicationException{
	           return DomainManager.getDomainManager().assignFilersToSubscription(sid, ids);

	  }    

	  @POST
	  @Path("topic")
	  @Consumes("application/json")
	  public long addTopic(TopicType t) throws PublicationException {
	           return DomainManager.getDomainManager().addTopic(t, null);
	  }

	  @DELETE
	  @Path("topic/{id}")
	  public void deleteTopic(@PathParam("id")long id) throws PublicationException {
	           DomainManager.getDomainManager().removeTopic(id);
	  }

	  @POST
	  @Path("topicsubscription/{tid}")
	  @Consumes("application/json")
	  public void assignTopicHierarchy(@PathParam("tid")long tid, IDsType ids) throws PublicationException{
	           DomainManager.getDomainManager().assignTopicHierarchy(tid, ids);
	  }

	  @POST
	  @Path("topicsubscription/{tid}")
	  @Consumes("application/json")
	  public long assignTopicSubscriptions(@PathParam("tid")long tid, IDsType ids)throws PublicationException{
	           return DomainManager.getDomainManager().assignTopicSubscriptions(tid, ids);
	  }

清单 10:发布 / 订阅服务

这些服务的使用者有消息发布者(publish 方法)、服务订阅者(创建 / 删除语义过滤器,订阅,还有订阅和主题订阅相关的过滤器)、内部的发布 / 订阅实现(获取同步的服务)和管理应用。

结论

这个实现虽然简单,但创建了一个非常强大、可扩展的发布 / 订阅实现,同时利用了很多现有的 AWS 功能和少量的 Java 定制代码。另外它还充分利用了现有 AWS 部署功能对负载均衡和容错的支持。

作者简介

Boris Lublinsky 博士是 Nokia 的主要架构师,参与大数据、SOA、BPM 和中间件实现的相关工作。Boris 去 Nokia 前是 Herzum 软件的主要架构师,负责为客户设计大型、可伸缩的 SOA 系统;在此之前,他是 CNA 保险的企业架构师,参与 CNA 集成和 SOA 策略的设计及实现,构建应用框架,实现面向服务的架构。Boris 在企业技术架构和软件工程方面有二十五年多的经验。他是 OASIS SOA RM 委员会的活跃成员,和他人一起编著了《Applied SOA: Service-Oriented Architecture and Design Strategies》一书,另外他还写了很多关于架构、编程、大数据、SOA 和 BPM 的文章。

查看英文原文:基于 AWS 技术实现发布 / 订阅服务