kafka笔记

kafka

kafka是什么？

　　kafka 是一个分布式消息发布订阅系统，可以处理大量数据，常用在线消息消费，数据可以保存到磁盘里，当消费者去取数据中途发生断开，kafka会做断点，连接成功继续读取，减少数据丢失和冗余

kafka架构

　　Broker
　　　　Kafka集群包含一个或多个服务器，这种服务器被称为broker
　　Topic
　　　　每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
　　Partition
　　　　Parition是物理上的概念，每个Topic包含一个或多个Partition.
　　Producer
　　　　负责发布消息到Kafka broker
　　Consumer
　　　　消息消费者，向Kafka broker读取消息的客户端。
　　Consumer Group
　　　　每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

配置文件

　　Vi server.properies

　　Port 9091默认端口号

　　Broker.id =0 # ID自己定义，唯一

　　#host.name=。。。。 #接收的ip，注释掉接收所有ip

　　Num.network.threads=3 #kafka 处理网络请求线程数一般情况下就是cpu的颗粒数

　　num.io.threads=8 处理io请求线程数

　　socket.send/receive.buffer.bytes=102400 发送接收缓存的大小

　　num.partitions=1 默认一个分区不用改

　　Num.partitions=1 #分区

　　Zookeeper.connect=xxxxx:xxx,yyyy:yyy,zzzz:zzz #zookeeper连接ip

　　zookeeper.connection.timeout.ms=1000000 连接最大超时时间

启动：

　　kafka自带一个zk，但是我们还有用自己的，不用它的

　　先起zookeeper（配置zookeeper请见zookeeper的基本配置）

　　bin/kafka-server-start.sh config/server .properties #不能后台

　　nohup bin/kafka-server-start.sh config/server.properties &

　　内存不足可能起不来ps -ef |grep kafka 查看进程或者看日志

　　kafka-console-consumer.sh
　　这个命令只是简单的将消息输出到标准输出中，通常使用该命令来获取kafka的topic中的数据。
　　使用方法：kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
　　kafka-console-producer.sh
　　这个命令可以将文件或标准输入的内容发送到Kafka集群。
　　使用方法：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

kafka-topics.sh
描述topic的配置
　　bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type topics --entity-name test_topic
设置保留时间
　　如果您需要删除主题中的所有消息，则可以利用保留时间。先将保留时间设置为非常低（1000 ms），等待几秒钟，然后将保留时间恢复为上一个值。默认保留时间为24小时（86400000毫秒）。
　　bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name test_topic --add-config retention.ms=1000
删除topic
　　bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test_topic

kafka-topics.sh
topic信息
　　bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test_topic
添加topic
　　bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic test_topic
列出topic
　　bin/kafka-topics.sh --list --zookeeper localhost:2181

连接测试：生产者消费者演示：3个zookeeper分别对应下面端口个位的1,2,3

生产者：

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

消费者：

　　bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --topic test --from-beginning

总结：作为9092端口代表的leader，可以向3个生产者发送信息，生产者一样可以向leader发送信息，但是不能发送给其他两个follower，当leader宕机或者挂掉，默认有新的leader启动，但是节点没有挂到，而是让其他两个follower的新leader去跑，我kill掉zookeeper2，发现消费者2182报错，其他两个消费者可继续发消息而发送9093成为leader，生产者依然是9092可以使用，但是我后台已经把他kill掉了，这就是zookeeper，此时杀掉，所有服务器个数少于一半zookeeper失效。在其他kafka扮演着消息列的离线收集，多集群，高稳定，低延迟的分布式消息系统

地区性：一般传json格式,生产者发送上海，接收者判断json值是上海还是北京，如果是上海接收，如果是北京不接受