网站文章伪原创怎么做,做网站都可以用什么框架,泸州网站建设哪家好,徐州优化网站目录 1、集群运维命令
1.1、集群启停命令
1.3、集群迁移命令
1.4、权限管理命令
1.4.1、权限参数介绍
1.4.2、增加权限命令
1.4.3、移出权限命令
1.4.4、查看所有topic权限命令
1.4.5、查看某个topic权限命令
2、生产者命令
2.1、创建topic命令
2.2、删除topic命令 …目录 1、集群运维命令
1.1、集群启停命令
1.3、集群迁移命令
1.4、权限管理命令
1.4.1、权限参数介绍
1.4.2、增加权限命令
1.4.3、移出权限命令
1.4.4、查看所有topic权限命令
1.4.5、查看某个topic权限命令
2、生产者命令
2.1、创建topic命令
2.2、删除topic命令
2.3、修改topic命令
2.3.1、修改配置命令
2.3.2、删除配置命令
2.4、查询topic命令
2.4.1、查询topic列表命令
2.4.2、查看某个topic详情命令 2.4.2、查看某个topic配置命令
2.5、发送消息命令
3、消费者命令
3.1、消费者组命令
3.1.1、查看消费者组列表命令
3.1.2、查看消费者组成员命令
3.1.3、查看某个消费组详情命令
3.1.4、查看某个消费者组状态命令
3.3、消费消息命令
4、性能测试命令
5、类执行命令 1、集群运维命令
1.1、集群启停命令
bin/kafka-server-start.sh -daemon config/server.propertiesbin/kafka-server-stop.sh
1.3、集群迁移命令
consumer.config file: 指定消费者的配置文件。该文件包含了消费源集群的相关配置如连接信息、组ID等。producer.config file: 指定生产者的配置文件。该文件包含了生产者写入目标集群的相关配置如连接信息、压缩方式等。whitelist topic: 指定需要镜像的主题列表可以使用逗号分隔多个主题名。blacklist topic: 指定需要排除镜像的主题列表可以使用逗号分隔多个主题名。与--whitelist互斥。num.streams number: 指定消费者线程数量默认是1。message.handler class: 自定义消息处理类可以在消息写入目标集群前对其进行处理。message.handler.args args: 传递给自定义消息处理类的参数。abort.on.send.failure true|false: 是否在发送失败时终止镜像过程默认为true。offset.commit.interval.ms number: 定期提交消费者偏移量的间隔时间默认是60000ms1分钟
#将指定topic按照消费者配置调用生产者发送到指定集群
bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topic1,topic21.4、权限管理命令
1.4.1、权限参数介绍 principal: 代表用户或服务主体例如 User:Alice。host: 允许或拒绝访问的主机地址* 表示任意主机。operation: 允许或拒绝的操作类型例如 READ, WRITE, ALL。permissionType: 权限类型可以是 ALLOW 或 DENY。resourceType: 资源类型例如 TOPIC 或 GROUP。name: 资源名称例如 example-topic。patternType: 资源模式类型LITERAL 表示字面模式。
1.4.2、增加权限命令
bin/kafka-acls.sh --authorizer-properties zookeeper.connectlocalhost:2181 \--remove --allow-principal User:user1 --operation Write --topic my-topic1.4.3、移出权限命令
bin/kafka-acls.sh --authorizer-properties zookeeper.connectlocalhost:2181 \--remove --allow-principal User:user1 --operation Write --topic my-topic1.4.4、查看所有topic权限命令
bin/kafka-acls.sh --list --bootstrap-server localhost:9092
Current ACLs for resource ResourcePattern(resourceTypeTOPIC, nameexample-topic, patternTypeLITERAL):(principalUser:Alice, host*, operationREAD, permissionTypeALLOW)(principalUser:Bob, host192.168.1.100, operationWRITE, permissionTypeALLOW)(principalUser:Carol, host*, operationREAD, permissionTypeDENY)Current ACLs for resource ResourcePattern(resourceTypeTOPIC, nameanother-topic, patternTypeLITERAL):(principalUser:David, host192.168.1.101, operationREAD, permissionTypeALLOW)Current ACLs for resource ResourcePattern(resourceTypeGROUP, nameexample-group, patternTypeLITERAL):(principalUser:Admin, host*, operationALL, permissionTypeALLOW)
1.4.5、查看某个topic权限命令
bin/kafka-acls.sh --authorizer-properties zookeeper.connectlocalhost:2181 \--list --topic my-topic2、生产者命令
2.1、创建topic命令
bin/kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 12.2、删除topic命令
bin/kafka-topics.sh --delete --topic my_topic --bootstrap-server localhost:90922.3、修改topic命令
2.3.1、修改配置命令
kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --config flush.messages1
2.3.2、删除配置命令
kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --delete-config flush.messages
2.4、查询topic命令
2.4.1、查询topic列表命令
bin/kafka-topics.sh --list --bootstrap-server localhost:90922.4.2、查看某个topic详情命令
bin/kafka-topics.sh --describe --topic my_topic --bootstrap-server localhost:90922.4.2、查看某个topic配置命令
bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms3600000Configs for topic my_topic are:cleanup.policydeletecompression.typeproducerdelete.retention.ms86400000file.delete.delay.ms60000flush.messages9223372036854775807flush.ms9223372036854775807follower.replication.throttled.replicasindex.interval.bytes4096leader.replication.throttled.replicasmax.message.bytes1000012message.downconversion.enabletruemessage.format.version2.7-IV0message.timestamp.difference.max.ms9223372036854775807message.timestamp.typeCreateTimemin.cleanable.dirty.ratio0.5min.compaction.lag.ms0min.insync.replicas1preallocatefalseretention.bytes-1retention.ms604800000segment.bytes1073741824segment.index.bytes10485760segment.jitter.ms0segment.ms604800000unclean.leader.election.enablefalse
2.5、发送消息命令
#默认发送消息
bin/kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092#指定分区发送命令
kafka-console-producer.sh --topic my-topic 2 --broker-list localhost:9092
3、消费者命令
3.1、消费者组命令
3.1.1、查看消费者组列表命令
#查看消费组列表命令
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092group1
group2
group3
my_consumer_group
another_consumer_group
3.1.2、查看消费者组成员命令
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test_group --membersCONSUMER-ID HOST CLIENT-ID #PARTITIONS
consumer-1-2d7d59c3-64c9-4526-905b-4e8920db3e84 /192.168.1.36 consumer-1 2
3.1.3、查看某个消费组详情命令
bin/kafka-consumer-groups.sh --describe --group my_group --bootstrap-server localhost:9092GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
my_group my_topic 0 12345 12350 5 consumer-1-1 /192.168.1.1 consumer-1
my_group my_topic 1 23456 23460 4 consumer-1-2 /192.168.1.1 consumer-1
my_group my_topic 2 34567 34567 0 consumer-1-3 /192.168.1.1 consumer-13.1.4、查看某个消费者组状态命令
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --stateCOORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS
192.168.1.36:9092 (1) range Stable 1
3.3、消费消息命令
#从头消费命令
bin/kafka-console-consumer.sh --topic my-topic --bootstrap-server localhost:9092 --from-beginning#指定分区消费命令
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --partition 1#指定消费者组id消费命令
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --consumer-property group.idtest
4、性能测试命令
4.1、生产者测试命令
bin/kafka-producer-perf-test.sh --topic perf-test --num-records 1000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers127.0.0.1:9092 compression.typelz41000 records sent, 3424.657534 records/sec (3.34 MB/sec), 13.61 ms avg latency, 255.00 ms max latency, 13 ms 50th, 20 ms 95th, 255 ms 99th.-h, --help 显示使用帮助并退出
--topic 指定生产的消息发往的 topic
--num-records 指定生产的消息总数
--payload-delimeter 如果通过 --payload-file 指定了从文件中获取消息内容那么这个参数的意义是指定文件的消息分隔符默认值为 \n即文件的每一行视为一条消息如果未指定 --payload-file 则此参数不生效
--throughput 限制每秒发送的最大的消息数设为 -1 表示不限制
--producer-props 直接指定 Producer 配置格式为 NAMEVALUE例如 bootstrap.server127.0.0.1:9092通过此种方式指定的配置优先级高于 --producer.config
--producer-config 指定 Producer 的配置文件格式参照官方的 config/producer.properties
--print-metrics 在测试结束后打印更详尽的指标默认为 false
--transactional-id 指定事务 ID测试并发事务的性能时需要只有在 --transaction-duration-ms 0 时生效默认值为 performance-producer-default-transactional-id
--transactional-duration-ms 指定事务持续的最长时间超过这段时间后就会调用 commitTransaction 来提交事务只有指定了 0 的值才会开启事务默认值为 0
--record-size 指定每条消息的大小单位是字节和 --payload-file 两个中必须指定一个但不能同时指定
--payload-file 指定消息的来源文件只支持 UTF-8 编码的文本文件文件的消息分隔符通过 --payload-delimeter 指定和 --record-size 两个中必须指定一个但不能同时指定
4.2、消费者测试命令
bin/kafka-consumer-perf-test.sh --bootstrap-server 127.0.0.1:9092 --topic perf_test --messages 1000000 --threads 8 --reporting-interval 1000 --show-detailedtime, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2021-03-25 15:57:59:426, 0, 657.2275, 657.2275, 673001, 673001.0000, 1616659078690, -1616659077690, 0.0000, 0.0000
...***************************输入参数解释***************************
--bootstrap-server 指定 broker 地址必选除非用 --broker-list 代替不建议
--topic 指定消费的 topic必选
--version 输出 Kafka 版本
--consumer.config 指定 Consumer 配置文件
--date-format 指定用于格式化 *.time 的规则默认为 yyyy-MM-dd HH:mm:ss:SSS
--fetch-size 指定一次请求消费的大小默认为 1048576 即 1 MB
--from-latest 如果 Consumer 没有已经建立的 offset则指定从 log 中最新的位点开始消费而不是从最早的位点开始消费
--group 指定 ConsumerGroup ID默认为 perf-consumer-40924
--help 显示使用帮助并退出
--hide-header 指定后不输出 header 信息
--messages 指定消费的消息数量必选
--num-fetch-threads 指定 fetcher 线程的数量
--print-metrics 指定打印 metrics 信息
--reporting-interval 指定打印进度信息的时间间隔默认为 5000 即 5 秒
--show-detailed-stats 指定每隔一段时间由 --reporting-interval 指定输出显示详细的状态信息
--socket-buffer-size 指定 TCP 的 RECV 大小默认为 2097152 即 2 MB
--threads 指定消费的线程数默认为 10
--timeout 指定允许的最大超时时间即每条消息返回的最大时间间隔默认为 10000 即 10 秒***************************输出参数解释***************************
time当前时间格式由 --date-format 指定
threadId线程 ID
data.consumed.in.MB消费到的数据总大小单位为 MB
MB.sec消费 TPS即每秒消费的消息大小
data.consumed.in.nMsg消费到的总消息数
nMsg.sec消费 TPS即每秒消费的消息条数
rebalance.time.ms消费者组重平衡的耗时单位为 ms0 表示没有发生重平衡
fetch.time.msfetch 线程的总耗时单位为 ms
fetch.MB.secfetch 线程每秒钟获取到的消息大小
fetch.nMsg.secfetch 线程每秒钟获取到的消息数量
#若没有指定 --show-detailed则输出信息中的前两项会有所不同如下
start.time, end.time, data.consumed.in.MB, MB.sec, ...
start.time消费开始的时间格式由 --date-format 指定
end.time消费结束的时间格式由 --date-format 指定
5、类执行命令
通过kafka-run-class命令执行指定类
#查看消费者消费进度
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic my-topicmy-topic:0:12345
my-topic:1:23456
my-topic:2:34567
