谁有那种手机网站,wordpress验证登录页面,泗洪做网站公司,上海人才招聘网官网1.简介
Kafka是一种高吞吐量的分布式发布订阅消息系统#xff0c;它可以处理消费者在网站中的所有动作流数据。 这种动作#xff08;网页浏览#xff0c;搜索和其他用户的行动#xff09;是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通…1.简介
Kafka是一种高吞吐量的分布式发布订阅消息系统它可以处理消费者在网站中的所有动作流数据。 这种动作网页浏览搜索和其他用户的行动是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。
Kafka [1] 是一种高吞吐量 [2] 的分布式发布订阅消息系统有如下特性
通过O(1)的磁盘数据结构提供消息的持久化这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。
高吞吐量 [2] 即使是非常普通的硬件Kafka也可以支持每秒数百万 [2] 的消息。 1.1 使用消息队列好处 1)、提高扩展性因为消息队列解耦了处理过程有新增需求时只要另外增加处理过程即可。
2)、提高峰值处理能力在访问量剧增的情况下应用仍然需要继续发挥作用。使用消息队列能够使关键组件顶住突发的访问压力而不会因为突发的超负荷请求而完全崩溃
3、提高系统的可恢复性消息队列降低了进程间的耦合度所以即使一个处理消息的进程挂掉加入队列中的消息仍然可以在系统回复后被处理1.2 kafka的结构 Producer消息生产者产生的消息将会被发送到某个topicConsumer消息消费者消费的消息内容来自某个topicTopic消息根据topic进行归类topic其本质是一个目录即将同一主题消息归类到同一个目录Broker每一个kafka实例或者说每台kafka服务器节点就是一个broker一个broker可以有多个topic
2.下载
获取我的baidu网盘中ELK kafka.apache.org 官方网址 由于我下载的filebeat是7.9.3与这个版本filebeat配合的kafka版本为0.11--2.2.2 https://www.elastic.co/guide/en/beats/filebeat/7.9/kafka-output.html 这里有说明。
Compatibility
This output works with all Kafka versions in between 0.11 and 2.2.2. Older versions might work as well, but are
not supported.
3.安装
tar zxvf kafka_2.11-2.0.1.tgz -C /usr/local ls cd /usr/local/ mv kafka_2.11-2.0.1 kafka ./bin 执行文件 ./config 配置文件主配置文件为 server.properties
4.配置
vi server.properties
broker.id0 集群内的唯一标识
listenersPLAINTEXT://ip[或者主机名]:9092 #配置监听端口 9092位kafka默认端口
log.dirs/usr/local/kafka/logs #日志及传输的数据存放目录可以通过“”分隔存放到多个目录如
log.dirs/data1,/data2
num.partitions6 #定义新的topic有多少个分区
num.retention.hours60 消息日志保留的时间单位是小时
#num.retention.minutes 或者num.retention.ms可以多个参数同时设置时间最小生效
log.segment.bytes1073741824 每个段文件的大小1G为默认值topic--partition--segement
zookeeper.connect10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
auto.create.topics.enabletrue #可以自动创建topics
delete.topic.enabletrue #可以自动删除空闲的topics是逻辑的删除非物理删除。
zookeeper.connect.timeout.ms6000
mkdir /data1 mkdir /data2
5.启动kafka
cd /usr/local/kafka
./bin/kafka-server-start.sh ./config/server.properties -d #-d 后台执行
或者nohup ./kafka-server-start.sh ../config/server.properties
启动时可能会报错注意关闭服务器的防火墙
jps查看
[rootlocalhost kafka]# jps
23696 Kafka
9271 QuorumPeerMain
24077 Jps
kafka提供了多个命令来查看创建修改删除topic信息也可以通过命令来测试日志消息 命令存放路径为usr/local/kafka/bin
[rootlocalhost bin]# pwd /usr/local/kafka/bin
[rootlocalhost bin]# ls
connect-distributed.sh kafka-consumer-groups.sh kafka-preferred-replica-election.sh kafka-streams-
application-reset.sh zookeeper-server-start.sh
connect-standalone.sh kafka-consumer-perf-test.sh kafka-producer-perf-test.sh kafka-topics.sh zookeeper-server-stop.sh
kafka-acls.sh kafka-delegation-tokens.sh kafka-reassign-partitions.sh kafka-
verifiable-consumer.sh zookeeper-shell.sh
kafka-broker-api-versions.sh kafka-delete-records.sh kafka-replica-verification.sh kafka-
verifiable-producer.sh
kafka-configs.sh kafka-dump-log.sh kafka-run-class.sh trogdor.sh
kafka-console-consumer.sh kafka-log-dirs.sh kafka-server-start.sh windows
kafka-console-producer.sh kafka-mirror-maker.sh kafka-server-stop.sh zookeeper-
security-migration.sh
例子kafka的任何操作都需要zookeeper 查看有多少个topic
./kafka-topics.sh --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --list
创建topic
./kafka-topics.sh --create --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --replication-factor
1 --partitions 3 --topic mytopic
实例
[rootlocalhost bin]# ./kafka-topics.sh --create --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--replication-factor 1 --partitions 3 --topic mytopic
Created topic mytopic.
[rootlocalhost bin]# ./kafka-topics.sh --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --list mytopic
[rootlocalhost bin]#
看看topic的具体信息
./kafka-topics.sh --describe --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --topic mytopic[rootlocalhost bin]# ./kafka-topics.sh --describe --zookeeper
10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --topic mytopic
Topic:mytopic PartitionCount:3 ReplicationFactor:1 Configs:Topic: mytopic Partition: 0 Leader: 1 Replicas: 1 Isr: 1Topic: mytopic Partition: 1 Leader: 2 Replicas: 2 Isr: 2Topic: mytopic Partition: 2 Leader: 3 Replicas: 3 Isr: 3
消费信息 交互式的生成消息到topic mytopic中
./kafka-console-producer.sh --broker-list 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic mytopic ctrl c 结束查看消费消息
./kafka-console-consumer.sh --bootstrap-server 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic
mytopic./kafka-console-consumer.sh --bootstrap-server 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic
mytopic--from-beginning #从头开始消费
删除topic
[rootlocalhost bin]# ./kafka-topics.sh --delete --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--topic mytopic[rootlocalhost bin]# ./kafka-topics.sh --delete --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--topic mytopic
Topic mytopic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
