Ubuntu16.04安装Kafka集群

fswdnr

　　下载
　　http://kafka.apache.org/downloads.html
　　http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz
　　root@master:/usr/local/kafka_2.11-0.11.0.0/config# vim server.properties
　　broker.id=2            每个节点不同
　　log.retention.hours=168
　　message.max.byte=5242880
　　default.replication.factor=2
　　replica.fetch.max.bytes=5242880
　　zookeeper.connect=master:2181,slave1:2181,slave2:2181
　　复制到其它节点
　　注意要事先在zk中创建/kafka节点，否则会报出错误：java.lang.IllegalArgumentException: Path length must be > 0
　　root@master:/usr/local/zookeeper-3.4.9# bin/zkCli.sh -server master

[zk: master(CONNECTED) 7] create /kafka ''

　　Created /kafka

[zk: master(CONNECTED) 8] ls /

[cluster, controller, controller_epoch, brokers, zookeeper, kafka, admin, isr_change_notification, consumers, latest_producer_id_block, config]

[zk: master(CONNECTED) 9] ls /kafka

[]

　　以守护进程方式启动kafka
　　root@master:/usr/local/kafka_2.11-0.11.0.0# nohup bin/kafka-server-start.sh config/server.properties &
　　创建topic：
　　root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh  --create  --zookeeper  master:2181  --replication-factor 1  --partitions  1  --topic test
　　Created topic "test".
　　列出所有topic：
　　root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --list --zookeeper master:2181
　　test
　　发送消息
　　root@master:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-producer.sh --broker-list master:9092 --topic test
　　>this is a message
　　>this is ant^H message
　　消费消息
　　root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test --from-beginning
　　Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
　　this is a message
　　this is an message
　　查看集群状态信息
　　root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --describe --zookeeper slave1:2181 --topic my-replicated-topic
　　Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:
　　Topic: my-replicated-topic    Partition: 0    Leader: 3    Replicas: 1,3,2    Isr: 3,2
　　安装kafka-manager
　　root@master:/usr/local/kafka_2.11-0.11.0.0# git clone https://github.com/yahoo/kafka-manager
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# cd kafka-manager/
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# ./sbt clean dist

[success] Total time: 3453 s, completed Aug 7, 2017 8:48:15 PM

　　打包好的文件存在
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager/target/universal# ls
　　kafka-manager-1.3.3.12.zip  tmp
　　修改kafka-manager配置文件
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# vim conf/application.conf
　　kafka-manager.zkhosts="192.168.117.243:2181,192.168.117.45:2181,192.168.117.242:2181"
　　启动kafka-manager
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf
　　推荐启动方式：
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=7778 &
　　登录kafka manager:
　　http://192.168.117.243:7778/
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# netstat -antlup | grep 7778
　　tcp6       0      0 :::7778                 :::*                    LISTEN      100620/java
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf
　　This application is already running (Or delete /usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12/RUNNING_PID file).
　　停止kafka-manager
　　root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# rm RUNNING_PID
　　root@master:/usr/local/kafka_2.11-0.11.0.0# cd kafka-manager-1.0-SNAPSHOT/
　　生产服务器配置
　　# Replication configurations
　　num.replica.fetchers=4
　　replica.fetch.max.bytes=1048576
　　replica.fetch.wait.max.ms=500
　　replica.high.watermark.checkpoint.interval.ms=5000
　　replica.socket.timeout.ms=30000
　　replica.socket.receive.buffer.bytes=65536
　　replica.lag.time.max.ms=10000
　　replica.lag.max.messages=4000
　　controller.socket.timeout.ms=30000
　　controller.message.queue.size=10
　　# Log configuration
　　num.partitions=8
　　message.max.bytes=1000000
　　auto.create.topics.enable=true
　　log.index.interval.bytes=4096
　　log.index.size.max.bytes=10485760
　　log.retention.hours=168
　　log.flush.interval.ms=10000
　　log.flush.interval.messages=20000
　　log.flush.scheduler.interval.ms=2000
　　log.roll.hours=168
　　log.retention.check.interval.ms=300000
　　log.segment.bytes=1073741824
　　# ZK configuration
　　zookeeper.connection.timeout.ms=6000
　　zookeeper.sync.time.ms=2000
　　# Socket server configuration
　　num.io.threads=8
　　num.network.threads=8
　　socket.request.max.bytes=104857600
　　socket.receive.buffer.bytes=1048576
　　socket.send.buffer.bytes=1048576
　　queued.max.requests=16
　　fetch.purgatory.purge.interval.requests=100
　　producer.purgatory.purge.interval.requests=100
　　Kafka是一种高吞吐量的分布式发布订阅的消息队列系统，原本开发自LinkedIn，用作LinkedIn的活动流（ActivityStream）和运营数据处理管道（Pipeline）的基础。现在它已被多家不同类型的公司作为多种类型的数据管道和消息系统使用。
1 Kafka消息队列简介
1.1 基本术语

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

1.2 消息队列
1.2.1 基本特性

• 　　可扩展
  
• 　　在不需要下线的情况下进行扩容
  
• 　　数据流分区(partition)存储在多个机器上
  
• 　　高性能
  
• 　　单个broker就能服务上千客户端
  
• 　　单个broker每秒种读/写可达每秒几百兆字节
  
• 　　多个brokers组成的集群将达到非常强的吞吐能力
  
• 　　性能稳定，无论数据多大
  
• 　　Kafka在底层摒弃了Java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合Zero-Copy的特性极大地改善了IO性能。
  
• 　　持久存储
  
• 　　存储在磁盘上
  
• 　　冗余备份到其他服务器上以防止丢失
  

1.2.2 消息格式

• 　　一个topic对应一种消息格式，因此消息用topic分类
  
• 　　一个topic代表的消息有1个或者多个patition(s)组成
  
• 　　一个partition中
  
• 　　一个partition应该存放在一到多个server上
  

• 　　一个server为leader
  
• 　　其他servers为followers
  
• 　　leader需要接受读写请求
  
• 　　followers仅作冗余备份
  
• 　　leader出现故障，会自动选举一个follower作为leader，保证服务不中断
  
• 　　每个server都可能扮演一些partitions的leader和其它partitions的follower角色，这样整个集群就会达到负载均衡的效果
  

• 　　如果只有一个server，就没有冗余备份，是单机而不是集群
  
• 　　如果有多个server
  
• 　　
  
  

消息按顺序存放消息顺序不可变只能追加消息，不能插入每个消息都有一个offset，用作消息ID, 在一个partition中唯一offset有consumer保存和管理，因此读取顺序实际上是完全有consumer决定的，不一定是线性的消息有超时日期，过期则删除1.2.3 生产者 producer

• 　　producer将消息写入kafka
  
• 　　写入要指定topic和partition
  
• 　　消息如何分到不同的partition，算法由producer指定
  

1.2.4 消费者 consumer

• 　　consumer读取消息并作处理
  
• 　　consumer group
  
• 　　
  
  
  
  
  • 　　可以按照partition的数目进行并发处理
    
  • 　　每个partition都只有一个consumer读取，因而保证了消息被处理的顺序是按照partition的存放顺序进行，注意这个顺序受到producer存放消息的算法影响
    
  • 　　这个概念的引入为了支持两种场景：每条消息分发一个消费者，每条消息广播给所有消费者
    
  • 　　多个consumer group订阅一个topic，该topci的消息广播给所有consumer group
    
  • 　　一条消息发送到一个consumer group后，只能由该group的一个consumer接收和使用
    
  • 　　一个group中的每个consumer对应一个partition可以带来如下好处
    
  • 　　
    
    
  
  
• 　　一个Consumer可以有多个线程进行消费，线程数应不多于topic的partition数，因为对于一个包含一或多消费线程的consumer group来说，一个partition只能分给其中的一个消费线程消费，且让尽可能多的线程能分配到partition（不过实际上真正去消费的线程及线程数还是由线程池的调度机制来决定）。这样如果线程数比partition数多，那么单射分配也会有多出的线程，它们就不会消费到任何一个partition的数据而空转耗资源 。
  
• 　　如果consumer从多个partition读到数据，不保证数据间的顺序性，kafka只保证在一个partition上数据是有序的，但多个partition，根据你读的顺序会有不同
  
• 　　增减consumer，broker，partition会导致rebalance，所以rebalance后consumer对应的partition会发生变化