Pulsar-Consumer

“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有以下两个，分别实现同步获取消息和异步获取消息：

/**
     * 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的实现方式差异比较大，值得一读。