大数据-Pulsar的常用命令_pulsar.auto.start

1 Pulsar的常用命令

1.1 client客户端

Pulsar 的 Local模式的基本使用：
（1）生产数据
bin/pulsar-client produce my-topic --messages “hello-pulsar”

说明：向my-topic这个topic生产数据，内容为“hello-pulsar”，如果topic不存在，pulsar会自动创建；

（2）消费数据
bin/pulsar-client consume my-topic -s “first-subscription”

说明：消费my-topic的数据，订阅名称为“first-subscription", 如果topic不存在，pulsar会自动创建；

Pulsar 的分布式集群模式的基本使用：
（1）生产数据
bin/pulsar-client produce persistent://public/default/test --messages “hello-pulsar”

说明：向my-topic这个topic生产数据，内容为“hello-pulsar”，如果topic不存在，pulsar会自动创建；

（2）消费数据
bin/pulsar-client consume persistent://public/default/test -s “first-subscription”

说明：消费my-topic的数据，订阅名称为“first-subscription", 如果topic不存在，pulsar会自动创建；

1.2 Pulsar多租户的相关操作

（1）获取租户列表
./pulsar-admin tenants list

（2）创建租户
./pulsar-admin tenants create my-tenant

./pulsar-admin tenants create my-tenant -r role1,role2,role3
在创建租户时，可以使用-r或者–admin-roles标志分配管理角色。可以用逗号分隔的列表指定多个角色；

（3）获取配置信息
./pulsar-admin tenants get my-tenant

（4）更新配置信息
./pulsar-admin tenants update my-tenant -r ‘pro’

添加一个pro角色；
（5）删除租户
./pulsar-admin tenants delete my-tenant

1.3 Pulsar名称空间的相关操作

（1）在指定的租户下创建名称空间
./pulsar-admin namespaces create test-tenant/test-namespace

（2）获取所有的名称空间列表
./pulsar-admin namespaces list test-tenant

（3）删除名称空间
./pulsar-admin namespaces delete test-tenant/ns1

（4）获取名称空间相关的配置策略
./pulsar-admin namespaces policies test-tenant/test-namespace

（5）配置复制集群
1- 设置复制集群:
./pulsar-admin namespaces set-clusters test-tenant/ns1 --clusters cl2

2- 获取给定命名空间复制集群的列表
./pulsar-admin namespaces get-clusters test-tenant/ns1

（6）配置backlog quota策略
1- 设置backlog quota 策略
./pulsar-admin namespaces set-backlog-quota --limit 10G --limitTime 36000 --policy producer_request_hold test-tenant/ns1

–policy 的值选择:
producer_request_hold：broker 暂停运行，并不再持久化生产请求负载
producer_exception：broker 抛出异常，并与客户端断开连接。
consumer_backlog_eviction：broker 丢弃积压消息

2- 获取 backlog quota 策略
./pulsar-admin namespaces get-backlog-quotas test-tenant/ns1

3 - 移除backlog quota策略
./pulsar-admin namespaces remove-backlog-quota test-tenant/ns1

（7）配置持久化策略
1- 设置持久化策略
./pulsar-admin namespaces set-persistence --bookkeeper-ack-quorum 2 --bookkeeper-ensemble 3 --bookkeeper-write-quorum 2 --ml-mark-delete-max-rate 0 test-tenant/ns1

参数说明:
bookkeeper-ack-quorum：每个 entry 在等待的 acks（有保证的副本）数量，默认值：0；
bookkeeper-ensemble：单个 topic 使用的 bookie 数量，默认值：0；
bookkeeper-write-quorum：每个 entry 要写入的次数，默认值：0；
ml-mark-delete-max-rate：标记-删除操作的限制速率（0表示无限制），默认值：0.0；

2- 获取持久化策略
./pulsar-admin namespaces get-persistence test-tenant/ns1

（8）配置消息存活时间(TTL)
1- 设置消息存活时间
./pulsar-admin namespaces set-message-ttl -ttl 100 test-tenant/ns1

2- 获取消息的存活时间
./pulsar-admin namespaces get-message-ttl test-tenant/ns1

3- 删除消息的存活时间
./pulsar-admin namespaces remove-message-ttl test-tenant/ns1

（9）配置整个名称空间中Topic的消息发送速率
1- 设置Topic的消息发送的速率
./pulsar-admin namespaces set-dispatch-rate test-tenant/ns1
–msg-dispatch-rate 1000
–byte-dispatch-rate 1048576
–dispatch-rate-period 1

参数说明:
–msg-dispatch-rate : 每dispatch-rate-period秒钟发送的消息数量
–byte-dispatch-rate : 每dispatch-rate-period秒钟发送的总字节数
–dispatch-rate-period : 设置发送的速率, 比如1表示每秒钟

2 获取topic的消息发送速率
./pulsar-admin namespaces get-dispatch-rate test-tenant/ns1

（10）配置整个名称空间中Topic的消息接收速率
1- 设置Topic的消息接收的速率
./pulsar-admin namespaces set-subscription-dispatch-rate test-tenant/ns1
–msg-dispatch-rate 1000
–byte-dispatch-rate 1048576
–dispatch-rate-period

参数说明:
–msg-dispatch-rate : 每dispatch-rate-period秒钟接收的消息数量
–byte-dispatch-rate : 每dispatch-rate-period秒钟接收的总字节数
–dispatch-rate-period : 设置接收的速率, 比如1表示每秒钟

2 获取topic的消息接收速率
./pulsar-admin namespaces get-subscription-dispatch-rate test-tenant/ns1

（11）配置整个名称空间中Topic的复制集群的速率
1- 设置Topic的消息复制集群的速率
./pulsar-admin namespaces set-replicator-dispatch-rate test-tenant/ns1
–msg-dispatch-rate 1000
–byte-dispatch-rate 1048576
–dispatch-rate-period 1

参数说明:
–msg-dispatch-rate : 每dispatch-rate-period秒钟复制集群的消息数量
–byte-dispatch-rate : 每dispatch-rate-period秒钟复制集群的总字节数
–dispatch-rate-period : 设置复制集群的速率, 比如1表示每秒钟

2 获取topic的消息复制集群的速率
./pulsar-admin namespaces get-replicator-dispatch-rate test-tenant/ns1

1.4 Pulsar Topic(主题)的相关操作

（1）创建Topic
方式一: 创建一个没有分区的topic
bin/pulsar-admin topics create persistent://my-tenant/my-namespace/my-topic

方式二: 创建一个有分区的topic
bin/pulsar-admin topics create-partitioned-topic persistent://my-tenant/my-namespace/my-topic --partitions 4

注意: 不管是有分区还是没有分区, 创建topic后,如果没有任何操作, 60s后pulsar会认为此topic是不活动的, 会自动进行删除, 以避免生成垃圾数据；

相关配置:
brokerdeleteinactivetopicsenabenabled : 默认值为true 表示是否启动自动删除；
brokerDeleteInactiveTopicsFrequencySeconds: 默认为60s 表示检测未活动的时间；

（2）列出当前某个名称空间下的所有Topic
./pulsar-admin topics list my-tenant/my-namespace

（3）更新Topic操作
我们可针对有分区的topic去更新其分区的数量；
./pulsar-admin topics update-partitioned-topic persistent://my-tenant/my-namespace/my-topic --partitions 8

（4）删除Topic操作
1- 删除没有分区的topic:
bin/pulsar-admin topics delete persistent://my-tenant/my-namespace/my-topic

2-删除有分区的topic
bin/pulsar-admin topics delete-partitioned-topic persistent://my-tenant/my-namespace/my-topic

（5）授权
./pulsar-admin topics grant-permission --actions produce,consume --role application1 persistent://test-tenant/ns1/tp1

（6）获取授权
./pulsar-admin topics grant-permission --actions produce,consume --role application1 persistent://test-tenant/ns1/tp1

（7）取消授权
./pulsar-admin topics revoke-permission --role application1 persistent://test-tenant/ns1/tp1

1.5 Pulsar Function轻量级计算框架

（1）构建function
bin/pulsar-admin functions create
–jar examples/api-examples.jar
–classname org.apache.pulsar.functions.api.examples.ExclamationFunction
–inputs persistent://public/default/exclamation-input
–output persistent://public/default/exclamation-output
–tenant public
–namespace default
–name exclamation

bin/pulsar-admin functions属性说明:
functions:
可选值:
localrun: 创建本地function进行运行
create: 在集群模式下创建
delete: 删除在集群中运行的function
get: 获取function的相关信息
restart: 重启
stop : 停止运行
start: 启动
status: 检查状态
stats: 查看状态
list: 查看特定租户和名称空间下的所有的function
–classname: 设置function执行类；
–jar: 设置function对应的jar包；
–inputs : 输入的topic；
–output : 输出的topic；
–tenant : 设置function运行在那个租户中；
–namespace: 设置function运行在那个名称空间中；
–name : 定义function的名称；

（2）启动function
检查是否按照预期触发函数运行；
bin/pulsar-admin functions trigger --name exclamation --trigger-value “hello world”

1.6 Pulsar跨机房复制配置

假如有三个pulsar数据中心的方案(cluster1, cluster2, cluster3)；

第一步: 首先创建一个租户, 并给予三个数据中心的权限；
bin/pulsar-admin tenants create my-tenant
–allowed-clusters cluster1, cluster2, cluster3

第二步: 创建namespace；
bin/pulsar-admin namespaces create my-tenant/my-namespace

第三步: 设置namespace中topic在那些数据中心之间进行互备；
bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace
–clusters cluster1, cluster2, cluster3