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
————————————————
版权声明:本文为CSDN博主「For Coding」的原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接及本声明。
原文链接:https://blog.csdn.net/m0_46689661/article/details/123176094

Logo

华为开发者空间,是为全球开发者打造的专属开发空间,汇聚了华为优质开发资源及工具,致力于让每一位开发者拥有一台云主机,基于华为根生态开发、创新。

更多推荐