前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Pulsar-Consumer

Pulsar-Consumer

作者头像
林一
发布2018-10-19 16:37:50
1.9K0
发布2018-10-19 16:37:50
举报
文章被收录于专栏:MessageQueue

“Pulsar is a distributed pub-sub messaging platform with a very flexible messaging model and an intuitive client API.”

Pulsar是pub-sub模式的分布式消息平台,拥有灵活的消息模型和直观的客户端API。

Pulsar由雅虎开发并开源的下一代消息系统,目前是Apache软件基金会的孵化器项目。

本片文章简单介绍Pulsar的Consumer,包含以下内容:

  • Consumer的体系
  • 消费逻辑的实现

1. Consumer体系

A consumer is a process that attaches to a topic via a subscription and then receives messages.

Consumer通过订阅关系绑定到Topic(和Producer类似,都是绑定到一个Topic上),并接收消息。

Consumer支持:

  • 同步接收消息:阻塞用户线程等待消息
  • 异步接收消息:异步等待消息(通过Future返回消息)
  • 通过MessageListener返回消息:接收消息后回调用户的MessageListener

Consumer提供了三类获取消息的方式,其中异步的方式包含通过Future异步等待消息和通过MessageListener被动接收消息。MessageListener和另外两种方式是互斥的,一旦Consumer注册了MessageListener接口,则必须通过MessageListener处理消息,主动触发receive获取消息将抛出异常。

Consumer的继承关系:

  • Consumer:定义了消费者相关的接口
  • ConsumerBase:接口中基础方法的实现,抽象类
  • ConsumerImpl:在ConsumerBase基础上的Consumer具体实现
  • MultiTopicsConsumerImpl:组合多个ConsumerImpl完成对多Topic/Partition的消费

Consumer的设计和Producer是一致的,通过接口定义行为,基础类实现基本能力,在通过组合的方式来实现消费多个Topic/Partition(Producer则是像多个Topic发送消息)。

1.1 消费进度提交

Consumer处理消息后需要发送acknowledgement到Broker,这样Broker可以丢弃消息(应该是移动消费offset的操作,类似RocketMQ,并不是真正的删除消息)。支持单挑消息提交或者批量提交,批量提交则以最后一条消息的offset为准。(只是记录一个offset比较某个位置之前的消息都已经被Consumer处理,所以批量提交其实只是把最大的offset提交)

1.2 订阅模型

订阅模型决定了消息时如何被投递给Consumer的。在Pulsar中,订阅模型有: exclusive、shared、 failover。

Exclusive

只能有一个Consumer绑定到订阅关系上,其他Consumer尝试绑定到订阅关系上时会报错(Exclusive是默认的订阅模型)。

Shared

在Shared模型中,多个Consumer可以绑定到一个订阅关系上。消息按照round-robin模式被投递给各个Consumer。若某个Consumer宕机,被投递给该Consumer的未被ACK(没有acknowledgement)的消息将被重新投递给其他的Consumer进行消费。

Shared模式带来的限制:

  1. 消息时按照round-robin模式投递给各个Consumer的,所以消息顺序无法得到保障
  2. 同样因为round-robin模式,无法使用批量提交acknowledgement的功能(如上图Consumer C-3如果提交了m4会导致m3被标记为已经消费,但实际Consumer C1可能还没处理m3)

failover

在Failover模型中,多个Consumer可以绑定到一个订阅关系上,但是只有一个称为Master Consumer的消费者能消费消息。对多个Consumer按照name进行排序,第一个Consumer则为Master Consumer。

在Master Consumer失效(比如断开连接)后,Master Consumer未提交的消息和后续的消息会提交给后续的Consumer。

2. 消费逻辑的实现

Consumer获取消息的核心API有以下两个,分别实现同步获取消息和异步获取消息:

代码语言:javascript
复制
/**
     * Receives a single message.
     * <p>
     * This calls blocks until a message is available.
     *
     * @return the received message
     * @throws PulsarClientException.AlreadyClosedException
     *             if the consumer was already closed
     * @throws PulsarClientException.InvalidConfigurationException
     *             if a message listener was defined in the configuration
     */
    Message<T> receive() throws PulsarClientException;

    /**
     * Receive a single message
     * <p>
     * Retrieves a message when it will be available and completes {@link CompletableFuture} with received message.
     * </p>
     * <p>
     * {@code receiveAsync()} should be called subsequently once returned {@code CompletableFuture} gets complete with
     * received message. Else it creates <i> backlog of receive requests </i> in the application.
     * </p>
     *
     * @return {@link CompletableFuture}<{@link Message}> will be completed when message is available
     */
    CompletableFuture<Message<T>> receiveAsync();

MessageListener则通过ConsumerBuilder接口进行设置并传入到Consumer的构造方法中。

这三个API都由ConsumerImpl#messageReceived触发,即Consumer接收到消息后根据请求的类型来决定:

  • 同步获取消息的,将消息放入内存队列,被挂起的线程会从队列中获取消息
  • 异步获取消息的,执行callback将消息放入future
  • 通过MessageListener处理消息的,通过ListenerExecutor执行逻辑

可见Pulsar在消费模式上处理是统一的,即无论客户端采用何种方式进行消息的接收,消息统一由服务端进行“推送”,而在Consumer内部根据用户请求的类型进行处理。

通过ConsumerImpl#messageReceived的实现可以发现Pulsar的消息消费是一种“推”的模型,这和RocketMQ的“拉”的模型差异是很大的(RocketMQ采用一种Long-Polling的方式,由Consumer主动发起请求从服务端获取数据,若服务端有需要处理的消息,请求立即返回;如果没有消息,这个请求会在服务单阻塞一段时间,直到新消息到达或者请求即将超时,返回给客户端)。

Consumer获取消息的模型

具体看Pulsar-Consumer获取消息的代码实现会发现它也不是一种纯粹的,类似淘宝Notify的推的模式,而是一种推拉结合的方式,示意如下:

  1. Consumer向Broker发送FLOW请求,通知Broker可以推送消息给Consumer
  2. Broker将消息通过MESSAGE请求将消息推送给Consumer

这是一个反复的过程,每次Consumer接收消息处理后都会继续发送FLOW请求给Broker。

这是在RocketMQ或者Kafka的设计中都没有的一种方式,这种方式进行一定的拓展则可以实现类似akka的Dynamic Push/Pull模式(详见公众号历史文章:《Push or Pull?》)。

在阅读Pulsar Consumer部分代码的时候还发现非常有趣的一点,当你搜索“Consumer”时会出现一个Consumer接口和一个Consumer类:

  • 接口: org.apache.pulsar.client.api.Consumer
  • 类: org.apache.pulsar.broker.service.Consumer

Consumer接口是Client模块定义Consumer行为的,为什么在Broker模块会有一个Consumer类?

实际在Broker端会给链接上来的Consumer构造一个对应的Consumer对象,维护远端的Consumer的链接等信息。所有对远端的Consumer的操作会封装在Broker端的Consumer中。这样可以更好的实现代码的可插拔性,降低耦合,提升代码的可测试性。比如在测试Broker端的逻辑时,只需要Mock一个Consumer类来模拟各种正常和网络异常的情况,而不需要真正的启动一个Consumer。

总结

本文主要是介绍一下Pulsar Consumer模块的相关概念和一些模型,没有深入的解读代码实现(有兴趣的朋友可以自己读一下代码)。Pulsar Consumer的实现方式还是非常有趣的,和大家比较熟悉的RocketMQ的实现方式差异比较大,值得一读。

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2018-09-25,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 MessageQueue 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 1. Consumer体系
    • 1.1 消费进度提交
      • 1.2 订阅模型
      • 2. 消费逻辑的实现
      • 总结
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档