码迷,mamicode.com
首页 > 其他好文 > 详细

Kafka

时间:2019-08-11 12:44:35      阅读:108      评论:0      收藏:0      [点我收藏+]

标签:read   commit   保存   时间   sla   des   守护   方案   进程   

概念

Kafka 是一个快速、可扩展和高可用的基于发布-订阅模式(pub-sub model)的消息系统,用作消息中间件,在系统之间传递消息。其核心概念有:

  • Topic(话题)
  • Producer(生产者)
  • Consumer(消费者)
  • Broker(经纪人)

在 Kafka 中,所有的消息都由 Topic 来组织,Producer 把消息发送到特定的 Topic,Consumer 从特定的 Topic 中读取消息。作为一个分布式系统,Kafka 运行在集群中,集群中的每个节点称之为 Broker。

快速开始

安装

下载 kafka_2.11-2.1.0.tgz,也可以使用命令 wget "http://mirrors.hust.edu.cn/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz" 下载,注意 kafka 中已经包含了 zookeeper。执行以下命令解压并切换工作目录:

tar -xzf kafka_2.11-2.1.0.tgz
cd kafka_2.11-2.1.0

启动

# 启动 zookeeper,默认监听端口 2181
bin/zookeeper-server-start.sh config/zookeeper.properties
# 启动 kafka,默认监听端口 9092
bin/kafka-server-start.sh config/server.properties

# 启动 zookeeper & kafka,以守护进程方式
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
bin/kafka-server-start.sh -daemon config/server.properties

停止

# 停止 zookeeper
bin/zookeeper-server-stop.sh config/zookeeper.properties
# 停止 kafka
ps aux | grep kafka | grep -v grep | awk '{print $2}'
kill [pid]

发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

接收消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

查看 Topic 列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

查看指定 Topic 的信息

bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic a_topic

修改指定 Topic 的 partition

bin/kafka-topics.sh --alter --topic STOP_JOB --zookeeper localhost:2181 --partitions 3

查看所有消费组

# 新版
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server 172.16.0.4:9092 --list
# 旧版
bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --list

查看消费组偏移量

# 新版
bin/kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --describe --group console-consumer-85731
# 旧版
bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --group console-consumer-11967 --describe

重设偏移量(new)

# 指定偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --group ABOT_MASTER --topic JOB_LEGACY --reset-offsets --to-offset 29261788 –execute

./kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --reset-offsets --group ABOT_SLAVE --all-topics  --by-duration PT06H0M0S  --execute

./kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --reset-offsets --group ABOT_MASTER --all-topics  --by-duration PT06H0M0S  --execute

#最新偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --group ABOT_MASTER --topic JOB_LEGACY --reset-offsets --to-latest –execute

bin/kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --reset-offsets --group ABOT_MASTER --all-topics  --to-latest  --execute

bin/kafka-consumer-groups.sh --bootstrap-server 172.16.0.4:9092 --reset-offsets --group ABOT_SLAVE --all-topics  --to-latest  --execute

进一步学习

Kafka Client 消息接收的三种模式

模式 特点 设置
At-most-once(最多一次) 消息最多被消费一次 让 Kafka 在特定时间间隔内自动提交,如网络中断恢复或程序重启可能会使消息未被处理完就被提交了。设置 enable.auto.commit 为 true,设置 auto.commit.interval.ms 为一个较小的时间间隔,不用手动调用 commitSync()
At-least-once(最少一次) 消息至少被消费一次 手动提交,如提交失败,则下次重复推送。设置 enable.auto.commit 为 false,然后调用 commitSync()
Exactly-once(正好一次) 消息有且只有被消费一次 保证消息处理和提交反馈在同一个事务中(ACID,原子性、一致性、隔离性和持久性)。设置 enable.auto.commit 为 false,保存 ConsumeRecord 中的 offset 到数据库,实现 ConsumerRebalanceListener ,监听 Consumer Rebalance 事件,然后使用seek 方法将数据库的 offset 更新到 Kafka。

Consumer Group 与 Partition

  1. 按照如上的算法,所以如果kafka的消费组需要增加组员,最多增加到和partition数量一致,超过的组员只会占用资源,而不起作用。
  2. kafka的partition的个数一定要大于消费组组员的个数,并且partition的个数对于消费组组员取模一定要为0,不然有些消费者会占用资源却不起作用。
    3.如果需要增加消费组的组员个数,那么也需要根据上面的算法,调整partition的个数。

作者:ens
链接:https://juejin.im/post/5baca032e51d450e735e74af
来源:掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

 $ /bin/kafka-topics.sh --alter --topic A_TASK --zookeeper localhost:2181 --partitions 2
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

$ bin/kafka-topics.sh --list --zookeeper localhost:2181

监控

kafka 版本要求 <=1.1.0 ( wget "https://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz" )

可以通过 Morningstar/kafka-offset-monitor 监控 Kafka 偏移量,具体是

  1. 下载 Jar

  2. 运行

    java -Djava.security.auth.login.config=conf/server-client-jaas.conf  -cp KafkaOffsetMonitor-assembly-0.4.6-SNAPSHOT.jar        com.quantifind.kafka.offsetapp.OffsetGetterWeb      --offsetStorage kafka      --kafkaBrokers kafkabroker01:6667,kafkabroker02:6667      --kafkaSecurityProtocol SASL_PLAINTEXT      --zk zkserver01,zkserver02      --port 8081      --refresh 10.seconds      --retain 2.days      --dbName offsetapp_kafka
  3. 通过 http://{{HOST}}:9090 打开 web 页面监控。

排错

kafka 占用磁盘过多

设置 log.retention.hours=6 ,只保存日志 6 小时,然后重启 kafka,这样之前保留的日志会被自动清除。

Connection to node -1 could not be established. Broker may not be available.

  1. 检查连接的 kafka 集群地址是否正确;
  2. 检查 kafka 集群收发消息是否正常;
  3. 检查 kafka 配置文件(server.properties)是否正确。

默认配置如下:

listeners=PLAINTEXT://:9092

如果被修改为错误地址那么将无法连接。

ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

一个线程只能有一个 Consumer,如果多个线程共用一个 Consumer,那么就会出现这个错误。

解决方案就是去解决线程问题,确保只有一个线程调用一个 Consumer。

Kafka 消息不按顺序消费

背景:查看消息日志发现原本按顺序发送的消息 1,2,3 被消费的时候变成 1,3,2 了。

原因:Kafka 不保证全局的消息有序性,但保证同一个 Partition 消息的有序性。

解决方案:发消息时设置同一个 Key 或者直接指定同一个 Partition 即可。

Kafka 日志过多问题

调整日志级别为 info,在 logback.xml 文件中写入

<logger name="org.apache.kafka" level="INFO" />

参考

  1. Quickstart - kafka.apache.org

Kafka

标签:read   commit   保存   时间   sla   des   守护   方案   进程   

原文地址:https://www.cnblogs.com/lshare/p/11334361.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!