Kafka

介绍

官网：http://kafka.apache.org/

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

Apache Kafka与传统消息系统相比，有以下不同：

  • 它被设计为一个分布式系统，易于向外扩展；
  • 它同时为发布和订阅提供高吞吐量；
  • 它支持多订阅者，当失败时能自动平衡消费者；
  • 它将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。

Kafka 安装

下载

https://kafka.apache.org/downloads

mkdir -p /storage/app/kafka && cd storage/app/kafka
wget https://mirrors.gigenet.com/apache/kafka/2.7.0/kafka_2.13-2.7.0.tgz
tar xvf kafka_2.13-2.7.0.tgz

修改配置

vim /storage/app/kafka/kafka_2.13-2.7.0/config/server.properties

broker.id=1
log.dirs=/storage/app/kafka/kafka_2.13-2.7.0/logs

需要注意的是 kafka 依赖 zookeeper, 建议使用外部 zk, 但也支持使用内置的 zk

启动

使用安装包中的脚本启动单节点 Zookeeper 实例：

/storage/app/kafka/kafka_2.13-2.7.0/bin/zookeeper-server-start.sh\
-daemon /storage/app/kafka/kafka_2.13-2.7.0/config/zookeeper.properties

启动 kafka

/storage/app/kafka/kafka_2.13-2.7.0/bin/kafka-server-start.sh\
 /storage/app/kafka/kafka_2.13-2.7.0/config/server.properties

配置 kafka 到 systemd

新建配置 vim /etc/systemd/system/kafka.service

[Unit]
Description=kafka
Documentation=kafka
After=network-online.target
Wants=network-online.target

[Service]
Environment=" "
EnvironmentFile=-/etc/default/%p
ExecStart=/storage/app/kafka/kafka_2.13-2.7.0/bin/kafka-server-start.sh /storage/app/kafka/kafka_2.13-2.7.0/config/server.properties
ExecStop=/bin/kill -HUP $MAINPID
Restart=on-failure
ExecReload=/bin/kill -HUP $MAINPID
Restart=on-failure

KillSignal=SIGINT

[Install]
WantedBy=multi-user.target

重载 systemd

systemctl daemon-reload

启动 kafka

systemctl start kafka.service

查看 kafka 状态

systemctl status kafka.service 

root@jansora-Vostro-3669:/storage/app/kafka/kafka_2.13-2.7.0/config# systemctl status kafka.service 
● kafka.service - kafka
     Loaded: loaded (/etc/systemd/system/kafka.service; disabled; vendor preset: enabled)
     Active: active (running) since Tue 2021-01-19 12:27:40 CST; 4s ago
   Main PID: 1061890 (java)
      Tasks: 74 (limit: 19037)
     Memory: 323.4M
     CGroup: /system.slice/kafka.service
             └─1061890 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineL>

1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,115] INFO [ExpirationReaper-1-AlterAcls]: Starting (kafka.server.DelayedOperationPu>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,144] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChan>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,172] INFO [SocketServer brokerId=1] Starting socket server acceptors and processors>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,199] INFO [SocketServer brokerId=1] Started data-plane acceptor and processor(s) fo>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,199] INFO [SocketServer brokerId=1] Started socket server acceptors and processors >
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,202] INFO Kafka version: 2.7.0 (org.apache.kafka.common.utils.AppInfoParser)
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,202] INFO Kafka commitId: 448719dc99a19793 (org.apache.kafka.common.utils.AppInfoPa>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,202] INFO Kafka startTimeMs: 1611030463200 (org.apache.kafka.common.utils.AppInfoPa>
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,209] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
1月 19 12:27:43 jansora-Vostro-3669 kafka-server-start.sh[1061890]: [2021-01-19 12:27:43,236] INFO [broker-1-to-controller-send-thread]: Recorded new controller, from now o>

演示

发送消息

--broker-list 当前 kafka 服务

--topic kafka 主题

/storage/app/kafka/kafka_2.13-2.7.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

发送消息

>1
>2
>3
>4
>5
>6
>7
>8
>9

接收消息

--bootstrap-server 当前 kafka 服务

--topic kafka 主题

--from-beginning 从开始记录

/storage/app/kafka/kafka_2.13-2.7.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

1
2
3
4
5
6
7
8
9

评论栏