什么是消息队列
点对点：每个消息只有一个消费者，只有一个消费者可以消费此消息，消费完就可以删除。
发布\订阅：发送到消息topic中的消息可以被所有订阅者消费，所有订阅者都消费完成后才可以被删除该消息，要先订阅再发送消息才能消费到。
MQ 发展史
一代：采用无序或者共享的方式来消费消息（队列）
2003 activeMQ
2006 rabbitMQ
二代：消息的消费严格排序和独占消息消费（流）（对一个topic分区始终只有一个消费者使用和消费），分布式
2010 kafka
2011 apache rocketMQ （阿里）
2013 tubeMQ
三代：存储分离的计算架构同时又支持共享和独占消费的MQ服务
2016 pulsar
什么时候使用MQ？
异步调用，系统解耦，削峰填谷，广播通知，分布式缓存，消息通讯
消息消费
在一个现代的实时流式架构中，消息用例可被分为两类：队列和流。
队列
队列是无序或共享的消息传递，通过队列进行消息传递，多个消费者可以被创建以从单个点对点消息传递通道接收消息。当通道传递消息时，任何消费者都可能接收消息。消息传递系统的实现决定哪个消费者实际接收的消息。队列用例通常与无状态的应用程序一起使用，无状态应用程序不关心排序，但它们需要能够进行消息确认(acknowledge)或消息删除(remove)、以及尽可能扩展消息消费并行性的能力。典型的基于排队的消息传递系统包括RabbitMQ和RocketMQ。
流
相比之下、流是严格排序或独占的消息传递。使用流式消息传递，始终只有一个消费者使用消息传递通道。消费者按照编写它们的确切顺序接收从通道发送的消息。流式用例通常与有状态应用程序相关联。有状态的应用程序关心顺序及其状态。消息的排序决定了有状态应用程序的状态。顺序将影响应用程序在发生无序消耗时需要应用的任何处理逻辑的正确性。
在面向微服务或事件驱动的体系结构中，流和队列都是必需的。
Pulsar Model
Apache Pulsar将队列和流统一为消息传递模型：producer-topic-subscription-consumer。主题（分区）是用于发送消息的命名通道。每个主题分区都由存储在Apache BookKeeper中的分布式日志支持。发布者发布的每条消息仅存储在主题分区上一次，复制以存储在多个bookies（BookKeeper服务器）上，并且可以根据消费者的需要多次消费使用。主题是消费真相的来源，尽管消息仅在主题分区上存储一次，但是可以有不同的方式来消费这些消息。消费者被组合在一起以消费消息。每组消费者都是对主题的订阅，每个消费者群体都可以拥有自己的消费方式 - 独占，共享或故障转移 - 这些消费群体可能会有所不同。这在一个模型和API中结合了队列和流，它的设计和实现目标是不影响性能和引入成本开销，同时还为用户提供了很多灵活性，以最适合当前用例的方式使用消息。
整体架构
Pulsar 采用 发布-订阅的设计模式(简称 pub-sub)， 在这种模式中，生产者向主题发布消息； 消费者 订阅这些主题，处理传入的消息，并在处理完成后向 broker 发送确认。（小马：kafka也是采用 发布-订阅的设计模式）
一旦创建订阅，即使 consumer 断开连接，Pulsar 仍然可以保存所有消息。 只有当消费者确认所有这些消息都已成功处理时，才会丢弃保留的消息。
消息存储在Topic中。逻辑上一个Topic是日志结构，每个消息都在这个日志结构中有一个偏移量。Apache Pulsar使用游标来跟踪偏移量。生产者将消息发送到一个指定的Topic，Apache Pulsar保证消息一旦被确认就不会丢失(正确的配置和非整个集群故障的情况下)。
消费者通过订阅来消费Topic中的消息。订阅是游标(跟踪偏移量)的逻辑实体，并且还根据不同的订阅类型提供一些额外的保证
Exclusive(独享) - 一个订阅只能有一个消息者消费消息
Shared(共享) - 一个订阅中同时可以有多个消费者，多个消费者共享Topic中的消息
Fail-Over(灾备) - 一个订阅同时只有一个消费者，可以有多个备份消费者。一旦主消费者故障则备份消费者接管。不会出现同时有两个活跃的消费者。
一个Topic可以添加多个订阅。订阅不包含消息的数据，只包含元数据和游标。
Apache Pulsar通过允许消费者将Topic看做在消费者消费确认后删除消息的队列，或者消费者可以根据游标的回放来提供队列和日志的语义。在底层都使用日志作为存储模型。
如果没有对Topic设置数据保留策略（目前通过其命名空间，后面会提供Topic级别的设置），一旦一个Topic的所有订阅的游标都已经成功消费到一个偏移量时，此偏移量前面的消息就会被自动删除。也就是说需要该Topic的所有订阅上得到消费确认。
但是，如果Topic设置了数据保留策略，已经消费确认的消息超过保留策略阈值(Topic的消息存储大小、Topic中消息保留的时间)后会被删除。
消费者可以以单条或者累积的方式确认消息。累积确认会有更好的吞吐量，但是在消息消费失败后会引入重复的消息处理。注意，累积消费不适用于共享模式的订阅，因为累积确认是基于偏移量的。但是在客户端API中支持批量确认，这样会减少RPC调用次数来提高在共享模式下订阅竞争消费的吞吐量。
最后，有一些类似于kafka Topic的分区(Partition)。区别在于Apache Pulsar中的分区也是Topic。就像kafka一样，生产者可以轮询、hash或者明确指定分区来发送消息。
pulsar 有四种订阅模式（详情）：
以下部分内容来自这里，感谢原作者整理分享。
独占订阅（Stream流模型）
顾名思义，独占订阅中，在任何时间，一个消费者组（订阅）中有且只有一个消费者来消费Topic中的消息。下图是独占订阅的示例。在这个示例中有一个有订阅A的活跃消费者A-0，消息m0到m4按顺序传送并由A-0消费。如果另一个消费者A-1想要附加到订阅A，则是不被允许的。
所以小马认为这里可以思考kafka或pulsar用集群来分担压力的最终原理是什么？消费者被限制了一对一，那是增加多个分区分布式吗？
关于kafka
Partition：每一个 Topic 又被分为多个 Partitions，即物理分区；出于负载均衡的考虑，同一个 Topic 的 Partitions 分别存储于 Kafka 集群的多个 broker 上；而为了提高可靠性，这些 Partitions 可以由 Kafka 机制中的 replicas 来设置备份的数量；如上面的框架图所示，每个 partition 都存在两个备份；
Consumer：消费者，从 Kafka 集群的 broker 中 pull 消息、消费消息；
Consumer group：high-level consumer API 中，每个 consumer 都属于一个 consumer-group，每条消息只能被 consumer-group 中的一个 Consumer 消费，但可以被多个 consumer-group 消费；（这一句非常重要！！图解）
kafka组内多个消费者来分别消费不同分区
这里的一个消费组相当于pulsar的一个订阅
故障切换（Stream流模型）
使用故障切换订阅，多个消费者（Consumer）可以附加到同一订阅。 但是，一个订阅中的所有消费者，只会有一个消费者被选为该订阅的主消费者。 其他消费者将被指定为故障转移消费者。
当主消费者断开连接时，分区将被重新分配给其中一个故障转移消费者，而新分配的消费者将成为新的主消费者。 发生这种情况时，所有未确认（ack）的消息都将传递给新的主消费者。 这类似于Apache Kafka中的Consumer partition rebalance。
下图是故障切换订阅的示例。 消费者B-0和B-1通过订阅B订阅消费消息。B-0是主消费者并接收所有消息。 B-1是故障转移消费者，如果消费者B-0出现故障，它将接管消费。
共享订阅（Queue队列模型）
使用共享订阅，在同一个订阅背后，用户按照应用的需求挂载任意多的消费者。 订阅中的所有消息以循环分发形式发送给订阅背后的多个消费者，并且一个消息仅传递给一个消费者。
当消费者断开连接时，所有传递给它但是未被确认（ack）的消息将被重新分配和组织，以便发送给该订阅上剩余的剩余消费者。
下图是共享订阅的示例。 消费者C-1，C-2和C-3都在同一主题上消费消息。 每个消费者接收大约所有消息的1/3。
如果想提高消费的速度，用户不需要增加分区数量，只需要在同一个订阅中添加更多的消费者。（小马：每一个订阅 约等于 kafka中的每一个 Consumer Group）
分区：Pulsar将一个主题的数据分布到多台机器上，保证高吞吐量。默认情况下，Pulsar的主题是不进行分区的，但通过命令行工具或API可以很容易地创建分区主题，并指定分区的数量。如下图：
三种订阅模式的选择
独占和故障切换订阅，仅允许一个消费者来使用和消费，每个对主题的订阅。这两种模式都按主题分区顺序使用消息。它们最适用于需要严格消息顺序的流（Stream）用例。
共享订阅允许每个主题分区有多个消费者。同一订阅中的每个消费者仅接收主题分区的一部分消息（就像kafka消费组内的多个消费者每个消费者只能分担接收一部分消息，如上文图）。共享订阅最适用于不需要保证消息顺序的队列（Queue）的使用模式（看上面的原理图），并且可以按照需要任意扩展消费者的数量。
Pulsar中的订阅实际上与Apache Kafka中的Consumer Group的概念类似（Consumer Group的概念见上文kafka图片）。
创建订阅的操作很轻量化，而且具有高度可扩展性，用户可以根据应用的需要创建任意数量的订阅。（小马：一个主题可以加多个订阅，类似kafka可以加多个 Consumer Group，并一条消息会被每个订阅中的其中一个消费者或者每个消费组中的其中一个消费者消费！！）
对同一主题的不同订阅，也可以采用不同的订阅类型。比如用户可以在同一主题上可以提供一个包含3个消费者的故障切换订阅，同时也提供一个包含20个消费者的共享订阅，并且可以在不改变分区数量的情况下，向共享订阅添加更多的消费者。
下图描绘了一个包含3个订阅A，B和C的主题（这里应该是消费者不是主题吧），并说明了消息如何从生产者流向消费者。（小马的问题：一个主题多个订阅模式的话，主题中的消息是等全部订阅模式中的消费者全部消费确认完才被删除？？对的，也可以设置确认后的再保留时间，下文有答案）
除了统一消息API之外，由于Pulsar主题分区实际上是存储在Apache BookKeeper中，它还提供了一个读取API（Reader），类似于消费者API（但Reader没有游标管理），以便用户完全控制如何使用Topic中的消息。
Pulsar的消息确认（ACK）
由于分布式系统的特性，当使用分布式消息系统时，可能会发生故障。比如在消费者从消息系统中的主题消费消息的过程中，消费消息的消费者和服务于主题分区的消息代理（Broker）都可能发生错误。消息确认（ACK）的目的就是保证当发生这样的故障后，消费者能够从上一次停止的地方恢复消费，保证既不会丢失消息，也不会重复处理已经确认（ACK）的消息。
在Apache Kafka中，恢复点通常称为Offset，更新恢复点的过程称为消息确认或提交Offset。
在Apache Pulsar中，每个订阅中都使用一个专门的数据结构--游标（Cursor）来跟踪订阅中的每条消息的确认（ACK）状态。每当消费者在主题分区上确认消息时，游标都会更新。更新游标可确保消费者不会再次收到消息。
Apache Pulsar提供两种消息确认方法，单条确认（Individual Ack）和累积确认（Cumulative Ack）。通过累积确认，消费者只需要确认它收到的最后一条消息。主题分区中的所有消息（包括）提供消息ID将被标记为已确认，并且不会再次传递给消费者。累积确认与Apache Kafka中的Offset更新类似。
Apache Pulsar可以支持消息的单条确认，也就是选择性确认。消费者可以单独确认一条消息。 被确认后的消息将不会被重新传递。下图说明了单条确认和累积确认的差异（灰色框中的消息被确认并且不会被重新传递）。在图的上半部分，它显示了累计确认的一个例子，M12之前的消息被标记为acked。在图的下半部分，它显示了单独进行acking的示例。仅确认消息M7和M12 - 在消费者失败的情况下，除了M7和M12之外，其他所有消息将被重新传送。
独占订阅或故障切换订阅的消费者能够对消息进行单条确认和累积确认；共享订阅的消费者只允许对消息进行单条确认。单条确认消息的能力为处理消费者故障提供了更好的体验。对于某些应用来说，处理一条消息可能需要很长时间或者非常昂贵，防止重新传送已经确认的消息非常重要。
这个管理Ack的专门的数据结构--游标（Cursor），由Broker来管理，利用BookKeeper的Ledger提供存储，在后面的文章中我们会介绍更多的关于游标（Cursor）的细节。
Apache Pulsar提供了灵活的消息消费订阅类型和消息确认方法，通过简单的统一的API，就可以支持各种消息和流的使用场景。
Pulsar的消息保留（Retention）
在消息被确认后，Pulsar的Broker会更新对应的游标。当Topic里面中的一条消息，被所有的订阅都确认ack后，才能删除这条消息。Pulsar还允许通过设置保留时间，将消息保留更长时间，即使所有订阅已经确认消费了它们。
下图说明了如何在有2个订阅的主题中保留消息。订阅A在M6和订阅B已经消耗了M10之前的所有消息之前已经消耗了所有消息。这意味着M6之前的所有消息（灰色框中）都可以安全删除。订阅A仍未使用M6和M9之间的消息，无法删除它们。如果主题配置了消息保留期，则消息M0到M5将在配置的时间段内保持不变，即使A和B已经确认消费了它们。
在消息保留策略中，Pulsar还支持消息生存时间（TTL）。如果消息未在配置的TTL时间段内被任何消费者使用，则消息将自动标记为已确认。 消息保留期消息TTL之间的区别在于：消息保留期作用于标记为已确认并设置为已删除的消息，而TTL作用于未ack的消息。 上面的图例中说明了Pulsar中的TTL。 例如，如果订阅B没有活动消费者，则在配置的TTL时间段过后，消息M10将自动标记为已确认，即使没有消费者实际读取该消息。
Pulsar VS. Kafka
通过以上几个方面，我们对Pulsar和Kafka在消息模型方面的不同点进行一个总结。
模型概念
Kafka： Producer - topic - consumer group - consumer；
Pulsar：Producer - topic - subscription - consumer。
消费模式
Kafka： 主要集中在流（Stream）模式，对单个partition是独占消费，没有共享（Queue）的消费模式；
Pulsar：提供了统一的消息模型和API。流（Stream）模式 -- 独占和故障切换订阅方式；队列（Queue）模式 -- 共享订阅的方式。
消息确认（Ack）
Kafka： 使用偏移Offset；
Pulsar：使用专门的Cursor管理。累积确认和Kafka效果一样；提供单条或选择性确认。
消息保留
Kafka：根据设置的保留期来删除消息。有可能消息没被消费，过期后被删除。 不支持TTL。
Pulsar：消息只有被所有订阅消费后才会删除，不会丢失数据。也允许设置保留期，保留被消费的数据。支持TTL。
对比总结：
Apache Pulsar将高性能的流（Apache Kafka所追求的）和灵活的传统队列（RabbitMQ所追求的）结合到一个统一的消息模型和API中。 Pulsar使用统一的API为用户提供一个支持流和队列的系统，且具有同样的高性能。
简单示例（以GO为例）
小马已将主要关注点加粗。
生产者
生产者创建生产者对象
producer, err := client.CreateProducer(pulsar.ProducerOptions{ Topic: "default/public/my-topic",})
if err != nil {
log.Fatalf("Could not instantiate Pulsar producer: %v", err)
}
defer producer.Close()
msg := pulsar.ProducerMessage{ Payload: []byte("Hello, Pulsar"),}
if err := producer.Send(context.Background(), msg); err != nil {
log.Fatalf("Producer could not send message: %v", err)
}
生产者示例
import ( "context" "fmt" "log" "github.com/apache/pulsar/pulsar-client-go/pulsar")
func main() {
// Instantiate a Pulsar client
client, err := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", })
if err != nil { log.Fatal(err)}
// Use the client to instantiate a producer
producer, err := client.CreateProducer(pulsar.ProducerOptions{ Topic: "default/public/my-topic", })
if err != nil { log.Fatal(err) } ctx := context.Background()
// Send 10 messages synchronously and 10 messages asynchronously
for i := 0; i < 10; i++ { // Create a message msg := pulsar.ProducerMessage{ Payload: []byte(fmt.Sprintf("message-%d", i)), }
// Attempt to send the message
if err := producer.Send(ctx, msg); err != nil { log.Fatal(err) }
// Create a different message to send asynchronously
asyncMsg := pulsar.ProducerMessage{ Payload: []byte(fmt.Sprintf("async-message-%d", i)), }
// Attempt to send the message asynchronously and handle the response
producer.SendAsync(ctx, asyncMsg, func(msg pulsar.ProducerMessage, err error) { if err != nil { log.Fatal(err) } fmt.Printf("the %s successfully published", string(msg.Payload)) }) }}
消费者
msgChannel := make(chan pulsar.ConsumerMessage)
//配置 主题，订阅名称，类型
consumerOpts := pulsar.ConsumerOptions{ Topic: "my-topic", SubscriptionName: "my-subscription-1", Type: pulsar.Exclusive, MessageChannel: msgChannel,}
consumer, err := client.Subscribe(consumerOpts)
if err != nil { log.Fatalf("Could not establish subscription: %v", err)}
defer consumer.Close()
for cm := range msgChannel { msg := cm.Message fmt.Printf("Message ID: %s", msg.ID()) fmt.Printf("Message value: %s", string(msg.Payload())) consumer.Ack(msg)}
pulsar比kafka和tubeMQ数据可靠性还要好。性能kafka>pulsar
pulsar 关键词或相关概念：租户和命名空间 集群 游标 支持事务消息
当前生态成熟度和业务的场景来看，不存在一个能满足所有场景要求的消息队列，Kafka倾向高吞吐高可用，Pulsar倾向数据高可靠。
腾讯kop项目：把数据协议从kafaka转pulsar，存储在pulsar，使业务/用户不用改代码。支持平滑迁移。原有的kafka不用作任何修改。
网友评论