首页 > 其他分享 >Pulsar Messaging

Pulsar Messaging

时间:2022-10-20 14:55:57浏览次数:69  
标签:持久 主题 默认 topic 消息 Messaging Pulsar

Messages

  • Pulsar 中的基本单位
  • 包含以下部分:
    • Value:消息携带的数据
    • Key:消息 Key,可以用于主题合并(同个 Key 保留最新消息)
    • Properties:用户自定义配置
    • Producer name:生产者名称,没有声明使用默认的
    • Topic name:消息将要发送到的 topic name
    • Schema version:Schema 版本
    • Sequence ID:针对于 topic 级别,每条消息的序号,可以用于消息去重,如果 brokerDeduplicationEnabled 设置为 true,该值在每个非分区 topic 中是唯一的
    • Message ID:bookies 中分配的消息 ID,表示每个消息在 ledger 中的位置,并且在单个 Pulsar 集群中唯一
    • Publish time:消息发布时间
    • Event time:通过应用额外和消息绑定的时间,没有设置默认为 0
  • 消息默认大小为 5 MB,可以通过以下配置修改:
    • broker.conf 文件:maxMessageSize=5242880
    • bookkeeper.conf 文件:nettyMaxFrameSizeBytes=5253120

生产者

发送模式

  • 同步
  • 异步:消息先放进阻塞队列然后立即返回,客户端在后台异步发送,队列满了生产者会阻塞或者失败

访问模式

  • Shared:允许多个生产者同时发布消息给主题,默认
  • Exclusive:只允许一个生产者发送消息给主题,其他生产者连接会报错
  • WaitForExclusive:同上,但是不报错只是阻塞,等连接的生产者下线后,会有一个被阻塞的生产者连接,其他的继续阻塞

压缩

  • LZ4
  • ZLIB
  • ZSTD
  • SNAPPY

  • 批大小由消息最大数量和最大发布延迟共同决定
  • backlog size 代表所有批的数量而不是所有消息的数量
  • 批次作为单个单元而不是单个消息进行跟踪和存储,消费者将一个批次拆分为单独的消息,但是即使配置了批处理,如果配置了 deliverAt 或 deliverAfter 参数,消息还是会单独发送
  • 通常来讲如果一批次的消息中有一部分投递失败,则整个批次会被重新投递,为了解决该问题,Pulsar 2.6.0 之后引入了 batch index acknowledgement,会过滤已经消费的消息,直到整个批次的消息全部确认,删除该批次,通过修改 acknowledgmentAtBatchIndexLevelEnabled=falsetrue来启用该机制,默认关闭,如果启用了该机制,会增大内存负载

Chunking

  • 生产者对大消息进行拆分,消费者聚合
  • 局限:
  • 仅适用于持久化主题
  • 仅适用于独享和灾备订阅模式
  • 不能和批处理同时启用
  • 消费端会对 chunk 进行缓存,收到所有 chunk 才进行聚合
  • 通过配置 maxPendingChunkedMessage 参数来限制消费者同时维护的分块消息的最大数量
  • 启用:
  • 禁用批机制,enableBatching 设置为 false
  • 默认关闭,设置 chunkingEnabled true 开启(创建生产者时)
  • 如果消费者在 1 分钟没有收到所有的 chunks ,会将不完整的 chunk 过期,通过设置 expireTimeOfIncompleteChunkedMessage 参数可以调用( ConsumerBuilder 构建时可以调用 ConsumerBuilder.expireTimeOfIncompleteChunkedMessage​(long duration, java.util.concurrent.TimeUnit unit) 进行设置)

消费者

消费端维护一个队列接收消息,大小通过 receiverQueueSize 来配置,默认为 1000,每次调用 consumer.receive() 方法从队列中取出一条数据

接收模式

  • 同步
  • 异步

监听器

客户端为消费者实现监听器,比如 Java 客户端提供了 MessageListener 接口,每次收到新消息的时候都会调用 received 方法

消息确认

  • 单条确认: consumer.acknowledge(msg)
  • 累计确认: consumer.acknowledgeCumulative(msg),不能用于共享订阅模式,因为会涉及多个消费者,所以只能单条确认

取消确认

  • 对失败的消息调用 negativeAcknowledge 方法取消确认,要求 Broker 重新投递

确认超时

  • ackTimeout:确认超时时间
  • ackTimeoutTickTime:检查间隔,默认 1 秒

重试主题

  • 消息处理失败,消费者可以将消息发送给重试主题,重新消费
  • 默认情况禁用自动重试,通过将 enableRetry 设置为 true
  • 当达到 maxRedeliverCount 时,未消费的消息将会被移到死信主题
  • 通过下列方式使用:
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .enableRetry(true)
        .deadLetterPolicy(DeadLetterPolicy.builder()
                .maxRedeliverCount(maxRedeliveryCount)
                .retryLetterTopic("my-retry-letter-topic-name")
                .build())
        .subscribe();

死信主题

  • 存储消费失败的消息
  • 使用方式如下:
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                      .maxRedeliverCount(maxRedeliveryCount)
                      .deadLetterTopic("my-dead-letter-topic-name")
                      .initialSubscriptionName("init-sub")
                      .build())
                .subscribe();

主题

  • Pulsar支持持久化主题和非持久化主题,前缀分别为persistent和non-persistent
  • Pulsar提供默认的命名空间default,默认租户public

订阅

订阅类型

  • Exclusive:独享,一个订阅组只绑定一个消费者
  • FailOver:灾备,一个订阅组绑定多个消费者,Pulsar从中选择一个作为主消费者,将所有的消息发送给该消费者,该消费者因故下线,重新选择主消费者
  • Shared:共享,一个订阅组绑定多个消费者,轮询方式分发消息
  • Key_Shared:键共享,相同key发送到同一个消费者(该模式生产者需禁用消息批次机制或使用 BatcherBuilder.KEY_BASED,可以将相同键的消息放到一个批次中)

订阅模式

  • 持久:默认情况下,没有任何持久订阅的消息会被标记为已删除,如果为该主题创建了持久订阅,只有确认的消息才会被标记为已删除
  • 非持久
  • 使用方式:
Consumer<byte[]> consumer = pulsarClient.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-sub")
        .subscriptionMode(SubscriptionMode.Durable)
        .subscribe();

多主题订阅

  • 正则表达式(所以主题必须在同一个命名空间)
  • 主题列表

分区主题

实际是多个内部主题,每个分区绑定一个 Broker

路由模式

决定每条消息被发送到哪个内部主题

  • RoundRobinPartition:如果消息指定key,按照key的Hash值将该消息分配给对应的内部主题,如果没有指定key,会将一个时间段内没有key的消息发送给同一个内部主题,每隔10ms以轮询方式切换到下一个内部主题
  • SinglePartition:如果消息指定key,按照key的Hash值将该消息分配给对应的内部主题,如果没有指定key,随机选择一个内部主题
  • CustomPartition:自定义路由器

Hashing scheme

  • 枚举类,表示选择消息分区时可以使用的散列函数的集合
  • 有两种类型:
    • JavaStringHash:默认
    • Murmur3_32Hash:当生产者来自不同多种语言的客户端时,推荐用该类型

非持久化主题

  • 消息不会持久化到磁盘,只存在于内存中的主题
  • 如果 Broker 掉线或者断开该主题的订阅,消息将会丢失
  • 非持久化主题中,Broker 会立即将消息投递给所有的订阅者而不会持久化到 BookKeeper,这样在某些情况下可以使得消息投递速度比持久化主题快,但是会丢失 Pulsar 的一些核心优势
  • 默认情况下启用

系统主题

  • Pulsar 内部使用的预先定义的主题,可以是非持久化或者持久化主题
  • 系统主题用于实现某些功能并且消除对第三方组件的依赖,比如事务、心跳检测、主题等级策略和资源组服务等,系统主题使得这些功能的实现变得简单、可靠和灵活
  • 不能创建任何系统主题
  • 通过该链接查看有哪些可用的系统主题:https://pulsar.apache.org/docs/concepts-messaging#system-topic
  • 默认情况下,系统主题被禁用,可以通过改变 conf/broker.confconf/standalone.conf 文件中的以下配置启用
systemTopicEnabled=true
topicLevelPoliciesEnabled=true

消息去重

生产者幂等

  • 确保每条消息只生成一次
  • 缺点事将重复消息的删除工作放在应用层面处理
  • 无需修改客户端代码,只需要修改某些配置

重复数据删除和单次有效语义

参考该链接:https://www.splunk.com/en_us/blog/it/exactly-once-is-not-exactly-the-same.html

标签:持久,主题,默认,topic,消息,Messaging,Pulsar
From: https://www.cnblogs.com/jordan95225/p/16809864.html

相关文章

  • Pulsar架构
    架构设计主要由三部分组成:Broker集群:一个或多个Broker节点,负责接收生产者发送的消息,将消息转发给BookKeeper存储,以及将消息投递给消费者,完成负载均衡、Schema校验、Ledg......
  • Pulsar基本介绍
    基本概念订阅组:类似于Kafka中的消费者组,不同订阅组之间消费消息互不干扰主题:分为非分区主题和分区主题非分区主题:一个主题和一个Broker绑定分区主题:由一组非分区内部......
  • [翻译] Quorum Queues - Making RabbitMQ More Competitive In Reliable Messaging
    目录AnIntroductiontoRaftBriefOverviewofRaftWriteSafetyReadSafetyRabbitMQQuorumQueuesNewDrawbacksQuestionsandOpenTopicsCouldtheidempotencyofwr......
  • pulsar-websocket协议
    WebSocket简介通过Pulsar的WebSocketAPI,用户可以简单便捷地与Pulsar进行交互,WebSocketAPI不依赖于官方客户端库。WebSocket支持使用Java、Go、Python和C++......
  • pulsar data
    干扰标记rfifind-time2.0-ofilename*.fits消色散DMprepdata-nobary-dm10.0-mask*.mask-ofilename_DM10.0*.fits快速傅里叶变换,将时域信号变成频域信......