基于docker搭建pulsar和使用攻略

pulsar

Pulsar是一个由yahoo公司于2016年开源的消息中间件，2018年成为Apache的顶级项目

我们先来看一下架构，从架构来看，和其他的消息中间件差不多，都是有消费者，生产者和broker,唯一一点不同的是pulsar的数据存储是存储在BookKeeper中的。
在这里插入图片描述

Producer:消息生产者，将消息发送到broker。
Consumer:消息消费者，从Broker读取消息到客户端，进行消费处理。
Broker: 可以看作是pulsar的server,Producer和Consumer都看作是client.消息处理的节点，pulsar的Broker和其他消息中间件的都不一样，他是无状态的没有存储，所以可以无限制的扩展。
Bookie: 负责所有消息的持久化，这里采用的是Apache Bookeeper。
ZK: 和kafka一样pulsar也是使用zk保存一些元数据，比如配置管理,topic分配，租户等等。
Service Discovery：可以理解为Pulsar中的nginx，只用一个url就可以和整个broker进行打交道，当然也可以使用自己的服务发现。客户端发出的读取，更新或删除主题的初始请求将发送给可能不是处理该主题的 broker 。如果这个 broker 不能处理该主题的请求，broker 将会把该请求重定向到可以处理主题请求的 broker。

安装

接下来我们安装一下，通过UI界面来看一下pulsar。以下是基于docker安装的

下载docker镜像

docker pull apachepulsar/pulsar:latest

启动服务

docker run --name pulsar \
-p 6650:6650 \
-p 8080:8080 \
--mount source=pulsardata,target=/pulsar/data \
--mount source=pulsarconf,target=/pulsar/conf \
-d apachepulsar/pulsar bin/pulsar standalone

6650端口：这是Pulsar Broker的默认二进制协议通信端口。是一个tcp协议。当你需要连接到Pulsar的Broker节点时，通常会使用这个端口。Broker节点是Pulsar的核心组件之一，负责接收和处理消息，以及提供主要的消息传递服务。如果你的应用程序是Pulsar的消息生产者或消费者，并且需要与Pulsar集群的Broker进行通信，那么就会使用6650端口。
8080端口：这是Pulsar Admin服务的默认REST API及Web界面端口。支持的是http协议。当你需要执行管理操作（例如创建或删除租户、命名空间、主题等），或者监控Pulsar集群的健康状况时，通常会使用8080端口。此端口提供了REST API接口以及Web界面，用于管理和监控Pulsar集群。

可视化管理

拉取web可视化管理平台

docker pull apachepulsar/pulsar-manager:latest

生成容器并启动

 
docker run -it \
  --name pulsar-manager \
  -p 9527:9527 -p 7750:7750 \
  --mount source=pulsarmanager,target=/pulsar-manager/pulsar-manager \
  -e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties \
  apachepulsar/pulsar-manager:latest

9527：这个端口是提供web-ui的端口
7750：提供rest-api接口，比如初始化管理员用户名密码

设置管理员密码

CSRF_TOKEN=$(curl http://localhost:7750/pulsar-manager/csrf-token)
curl \
   -H 'X-XSRF-TOKEN: $CSRF_TOKEN' \
   -H 'Cookie: XSRF-TOKEN=$CSRF_TOKEN;' \
   -H "Content-Type: application/json" \
   -X PUT http://localhost:7750/pulsar-manager/users/superuser \
   -d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "username@test.org"}'

通过9527访问界面

在这里插入图片描述
pulsar-manager可以管理多个pulsar服务，可以在这里进行配置pulsar的信息。

在这里插入图片描述
我们看一下面板里面的元素

persistent/non-persistentPulsar 提供持久化、非持久化两种主题，如果选择的是非持久化主题的话，所有消息都在内存中保存，如果broker重启，消息将会全部丢失。如果选择的是持久化主题，所有消息都会持久化到磁盘，重启broker，消息也可以正常消费。
tenant顾名思义就是租户，pulsar最开始在雅虎内部是作为全公司使用的中间件使用的，需要给topic指定一些层级，租户就是其中一层，比如这个可以是一个大的部门，例如电商中台租户。
namespace命名空间，可以看作是第二层的层级，比如电商中台下的订单业务组topic消息队列名字

概念了解

生产者

异步发送

我们上面说了send分为async和sync两种模式，但实际上在pulsar内部sync模式也是采用的async模式，在sync模式下模拟回调阻塞，达到同步的效果，这个在kafka中也是采用的这个模式，但是在rocketmq中，所有的send都是真正的同步，都会直接请求到broker。
基于这个模式，在pulsar和kafka中都支持批量发送，在rocketmq中是直接发送，批量发送有什么好处呢？当我们发送的TPS特别高的时候，如果每次发送都直接和broker直连，可能会做很多的重复工作，比如压缩，鉴权，创建链接等等。比如我们发送1000条消息，那么可能会做1000次这个重复的工作，如果是批量发送的话这1000条消息合并成一次请求，相对来说压缩，鉴权这些工作就只需要做一次。
在这里插入图片描述

负载均衡

在消息队列中通常会将topic进行水平扩展，在pulsar和kafka中叫做partition,在rocketmq中叫做queue，本质上都是分区，我们可以将不同分区落在不同的broker上，达到我们水平扩展的效果。
在我们发送的时候可以自己制定选择partition的策略，也可以使用它默认轮训partition策略。当我们选择了partition之后，我们怎么确定哪一个partition对应哪一个broker呢？
在这里插入图片描述

Step1: 我们所有的信息分区映射信息在zk和broker的缓存中都有进行存储。
Step2: 我们通过查询broker，可以获取到分区和broker的关系，并且定时更新。
Step3: 在pulsar中每个分区在发送端的时候都被抽象成为一个单独的Producer,这个和kafka,rocketmq都不一样，在kafka里面大概就是选择了partition之后然后再去找partition对应的broker地址，然后进行发送。pulsar将每一个partition都封装成Producer，再代码实现上就不需要去关注他具体对应的是哪个broker,所有的逻辑都在producer这个代码里面，整体来说比较干净。

压缩消息

消息压缩是优化信息传输的手段之一，我们通常看见一些大型文件都会是以一个压缩包的形式提供下载，在我们消息队列中我们也可以用这种思想，我们将一个batch的消息，比如有1000条可能有1M的传输大小，但是经过压缩之后可能就只会有几十kb，增加了我们和broker的传输效率，但是与之同时我们的cpu也带来了损耗。Pulsar客户端支持多种压缩类型，如 lz4、zlib、zstd、snappy 等。

broker

接下来我们来说说第二个比较重要的部分Broker,在Broker的设计中pulsar和其他所有的消息队列差别比较大，而正是因为这个差别也成为了他的特点。
首先我们来说说他最大的特点：计算和存储分离。我们在开始的说过Pulsar是下一代消息队列，就非常得益于他这个架构设计，无论是kafka还是RocketMQ,所有的计算和存储都放在同一个机器上，这个模式有几个弊端：

扩展困难：当我们需要扩展的集群的时候，我们通常是因为cpu或者磁盘其中一个原因影响，但是我们却要申请一个可能cpu和磁盘配置都很好的机器，造成了资源浪费。并且kafka这种进行扩展，还需要进行迁移数据，过程十分繁杂。
负载不均衡：当某些partion数据特别多的时候，会导致broker负载不均衡,如下面图，如果某个partition数据特别多，那么就会导致某个broker(轮船)承载过多的数据，但是另外的broker可能又比较空闲。

pulsar计算分离架构能够非常好的解决这个问题:

对于计算：也就是我们的broker,提供消息队列的读写,不存储任何数据，无状态对于我们扩展非常友好，只要你机器足够，就能随便上。扩容Broker往往适用于增加Consumer的吞吐，当我们有一些大流量的业务或者活动，比如电商大促，可以提前进行broker的扩容。
对于存储：也就是我们的bookie,只提供消息队列的存储，如果对消息量有要求的，我们可以扩容bookie,并且我们不需要迁移数据，扩容十分方便。

在这里插入图片描述

Entry,Entry是存储到bookkeeper中的一条记录，其中包含Entry ID，记录实体等。
Ledger，可以认为ledger是用来存储Entry的，多个Entry序列组成一个ledger。
Journal，其实就是bookkeeper的WAL(write ahead log)，用于存bookkeeper的事务日志，journal文件有一个最大大小，达到这个大小后会新起一个journal文件。
Entry log，存储Entry的文件，ledger是一个逻辑上的概念，entry会先按ledger聚合，然后写入entry log文件中。同样，entry log会有一个最大值，达到最大值后会新起一个新的entry log文件
Index file，ledger的索引文件，ledger中的entry被写入到了entry log文件中，索引文件用于entry log文件中每一个ledger做索引，记录每个ledger在entry log中的存储位置以及数据在entry log文件中的长度。
MetaData Storage，元数据存储，是用于存储bookie相关的元数据，比如bookie上有哪些ledger，bookkeeper目前使用的是zk存储，所以在部署bookkeeper前，要先有zk集群。

在这里插入图片描述

消费者

订阅模式

订阅模式是用来定义我们的消息如何分配给不同的消费者，不同消息队列中间件都有自己的订阅模式，一般我们常见的订阅模式有：

集群模式：一条消息只能被一个集群内的消费者所消费。
广播模式：一条消息能被集群内所有的消费者消费。

在pulsar中提供了4种订阅模式，分别是独占，灾备，共享，键共享：
在这里插入图片描述

独占：顾名思义只能由一个消费者独占，如果同一个集群内有第二个消费者去注册，第二个就会失败，这个适用于全局有序的消息。
灾备：加强版独占，如果独占的那个挂了，会自动的切换到另外一个好的消费者，但是还是只能由一个独占。
共享模式：这个模式看起来有点像集群模式，一条消息也是只能被一个集群内消费者消费，但是和rocketmq不同的是，rocketmq是以partition维度，同一个Partition的数据都会被发到一个机器上。在Pulsar中消费不会以partition维度，而是轮训所有消费者进行消息发送。这有个什么好处呢？如果你有100台机器，但是你只有10个partition其实你只有10台消费者能运转，但是在pulsar中100台机器都可以进行消费处理。
键共享：类似上面说的partition维度去发送，在rocketmq中同一个key的顺序消息都会被发送到一个partition，但是这里不会有partition维度，而只是按照key的hash去分配到固定的consumer,也解决了消费者能力限制于partition个数问题。

使用

我们前面已经启动了pulsar服务，接下来我通过pulsar-client-php来使用一下

php /usr/local/bin/composer require ikilobyte/pulsar-client-php

实例化生产者

// 尝试连接到Pulsar消息队列的生产者
try {
    // 创建生产者选项实例
    $options = new ProducerOptions();
    // 设置连接超时时间为3秒
    $options->setConnectTimeout(3);
    // 设置主题为persistent类型，指定主题名为'demo'
    $options->setTopic('persistent://public/default/demo');
    // 创建生产者实例，指定Pulsar服务地址
    $producer = new Producer('pulsar://localhost:6650', $options);
    // 连接到Pulsar服务
    $producer->connect();

// 捕获连接过程中可能抛出的异常
} catch (\Exception $e) {
    // 记录异常信息到日志
    Log::error('producer connect error', [
        'message' => $e->getMessage(),
        'file'=> $e->getFile(),
        'line'=> $e->getLine(),
        'trace' => $e->getTraceAsString(),
    ]);
    // 返回表示初始化Pulsar失败的错误信息
    return ['code'=>SysException::FAILED,'msg'=>'初始化pulsar失败'];
}

发送消息

消息主体

在我们发送消息之前，我们先来了解一下消息包含哪些东西

在这里插入图片描述

在这里插入代码片

基于此，我们写一个发送消息的demo

<?php

namespace App\Console\Commands;

use App\Exceptions\SysException;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;
use Pulsar\MessageOptions;
use Pulsar\Producer;
use Pulsar\ProducerOptions;

class PulsarTest extends Command
{
    protected $signature = 'PulsarTest';

    protected $description = 'Command description';
    public function __construct()
    {
        parent::__construct();
    }
    public function handle()
    {

        try {
            $options = new ProducerOptions();
            $options->setConnectTimeout(3);
            $options->setTopic('persistent://public/default/demo');
            $producer = new Producer('pulsar://localhost:6650', $options);
            $producer->connect();
            //发送一条普通消息
            $messageID = $producer->send(sprintf('hello %d', 11111));
            echo 'messageID ' . $messageID . "\n";
            $messageID = $producer->send(sprintf('hello properties %d', 22222), [
                //发送一条带消息属性的消息
                MessageOptions::PROPERTIES => [
                    'key' => 'value',
                    'ms' => microtime(true),
                ],
            ]);
            echo 'messageID ' . $messageID . "\n";


            //发送一条延迟消息
            $producer->send(sprintf('hello-delay %d', 33333), [
                MessageOptions::DELAY_SECONDS => 5, // Seconds
            ]);


            //批量发送消息
            $messages = [];
            for ($i = 0; $i < 10; $i++) {
                $messages[] = json_encode([
                    'id' => $i,
                    'now' => date('Y-m-d H:i:s')
                ]);
            }
            $messageID = $producer->send($messages);
            echo "batch message id ${messageID}\n";

            //关闭连接
            $producer->close();

        } catch (\Exception $e) {
            Log::error('producer connect error', [
                'message' => $e->getMessage(),
                'file' => $e->getFile(),
                'line' => $e->getLine(),
                'trace' => $e->getTraceAsString(),
            ]);
            return ['code' => SysException::FAILED, 'msg' => '初始化pulsar失败'];
        }
        
    }
}

ProducerOptions

setTopic() -设置topic 类似于queue
setAuthentication() -设置连接权限认证
setConnectTimeout() -设置连接超时
setCompression() -设置压缩方式

ConsumerOptions

setConsumerName 设置消费者名字
setSubscription 设置订阅者名称
setSubscriptionType 设置订阅类型
setSubscriptionInitialPosition设置订阅初始位置
setNackRedeliveryDelay 设置消息重新投递延迟时间
setReceiveQueueSize 设置队列大小
setDeadLetterPolicy 设置死信策略
setReconnectPolicy 设置重连策略

Message

消费者拿到的消息是包含以下属性的。

 @param MessageIdData $id 消息ID数据对象
 * @param int $consumerID 消费者ID
 * @param string $publishTime 消息发布时间
 * @param string $topic 消息主题
 * @param string $payload 消息负载内容
 * @param int $batchNums 批处理消息总数，默认为1
 * @param int $batchIdx 批处理消息索引，默认为0
 * @param int $redeliveryCount 消息重试次数，默认为0
 * @param MessageCollection|null $properties 消息属性集合，可以为空

消费demo

我们先通过可视化页面看一下,我现在启动了两个生产者和三个消费者
在这里插入图片描述
我们看一下页面里面的数据信息

点击storage标签看一下

由于三个消费者启动的时间不一样，所以他们读取队列的position是不一样的，他们默认读取他们启动之后的数据，这个我们可以通过option设置。
我们现在启动一个logic_3 并指定起始位置去消费

问题

1.先启动生产者消息丢失？关注 earliest

禁用 deleteTopicIfNoSubscriptions：无订阅删除主题

在这里插入图片描述

持久化订阅: 如果你使用的是持久化订阅（例如 persistent://…），那么一旦订阅被创建，消费者的位置就会被跟踪。如果你希望每次启动时都从特定位置开始，可能需要考虑删除旧的订阅或创建新的订阅。
2.死信队列的使用
自动 DLQ 配置（从 Pulsar 2.8.0 开始支持）
3.share-key的使用【有问题】

4.延迟队列
精准度秒级别

线上配置

以下是 broker.conf 文件中的一些重要配置参数及其作用：

zookeeperServers
- 说明：ZooKeeper 集群的连接字符串。
- 示例：zookeeperServers=zk1:2181,zk2:2181,zk3:2181
configurationStoreServers
- 说明：配置存储的连接字符串。
- 示例：configurationStoreServers=cfg1:2181,cfg2:2181,cfg3:2181
brokerServicePort
- 说明：Broker 数据服务端口。
- 默认值：6650
- 示例：brokerServicePort=6650
webServicePort
- 说明：用于处理 HTTP 请求的端口。
- 默认值：8080
- 示例：webServicePort=8080
advertisedAddress
- 说明：Broker 对外广播的地址或 IP。
- 示例：advertisedAddress=my-broker-host
maxConcurrentHttpRequests
- 说明：最大并发 HTTP 请求数。
- 默认值：1024
- 示例：maxConcurrentHttpRequests=1024
clusterName
- 说明：Broker 所属集群的名称。
- 示例：clusterName=my-cluster
allowAutoTopicCreation
- 说明：是否允许自动创建主题。
- 默认值：true
- 示例：allowAutoTopicCreation=true
allowAutoSubscriptionCreation
- 说明：是否允许自动创建订阅。
- 默认值：true
- 示例：allowAutoSubscriptionCreation=true
brokerDeleteInactiveTopicsEnabled
- 说明：是否启用删除不活跃的主题。
- 默认值：true
- 示例：brokerDeleteInactiveTopicsEnabled=true
brokerDeleteInactiveTopicsFrequencySeconds
- 说明：检查不活跃主题的频率（秒）。
- 默认值：60
- 示例：brokerDeleteInactiveTopicsFrequencySeconds=60
maxUnackedMessagesPerConsumer
- 说明：每个消费者允许的最大未确认消息数。
- 默认值：50000
- 示例：maxUnackedMessagesPerConsumer=50000
maxMessageSize
- 说明：消息的最大大小（字节）。
- 默认值：5242880 (5MB)
- 示例：maxMessageSize=5242880
enablePersistentTopics
- 说明：是否启用持久化主题。
- 默认值：true
- 示例：enablePersistentTopics=true
enableNonPersistentTopics
- 说明：是否启用非持久化主题。
- 默认值：true
- 示例：enableNonPersistentTopics=true
maxProducersPerTopic
- 说明：每个主题允许的最大生产者数。
- 默认值：0 (无限制)
- 示例：maxProducersPerTopic=10
maxConsumersPerTopic
- 说明：每个主题允许的最大消费者数。
- 默认值：0 (无限制)
- 示例：maxConsumersPerTopic=10
maxSubscriptionsPerTopic
- 说明：每个主题允许的最大订阅数。
- 默认值：0 (无限制)
- 示例：maxSubscriptionsPerTopic=10
maxConsumersPerSubscription
- 说明：每个订阅允许的最大消费者数。
- 默认值：0 (无限制)
- 示例：maxConsumersPerSubscription=10

这些参数是 Pulsar Broker 配置文件中的常见设置，用于控制 Broker 的行为、性能和安全性。根据具体需求，可以调整这些参数以优化 Broker 的运行。

pulsar部署后broker.conf一些配置项
是否自动创建
allowAutoTopicCreation=false
删除不活跃的topic
brokerDeleteInactiveTopicsEnabled=false
强制删除Namespace 开发和测试环境配置成true方便清理
forceDeleteNamespaceAllowed=false
强制删除Tenant 开发和测试环境配置成true方便清理
forceDeleteTenantAllowed=false
这个很重要，根据业务情况配，我是7天= 7 * 24 * 60
topic如果存在很多订阅没有消费，那么bk存储也不会删除
解决：
1、删除订阅 下面这个
2、配置TTL
subscriptionExpirationTimeMinutes=10080