Kafka简介及安装配置

Kafka简介及安装配置

1、简介

    Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，使用Scala语言编写，以后成为Apache项目的一部分，目前是Apache的一个顶级项目。

    Kafka是一个分布式的、可分区的、可复制的消息系统。它提供了普通消息系统的功能，但具备本身独特的设计。

    它提供了相似于JMS（Java消息队列规范）的特性，可是在设计实现上彻底不一样，此外它并非JMS规范的实现。

    kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群由多个kafka实例组成，每一个实例(server)称为broker。

    不管是kafka集群，仍是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。

    总结：

    Kafka是分布式消息队列，按topic分类存放数据，拥有Producer、Consumer、Broker三个角色，使用zookeeper作为集群的协调工具。

一、kafka的特色

1．高吞吐量

    理论上Kafka每秒能够生产约25万消息（50MB），每秒处理55万消息（110MB），生产环境中在这个速度上下浮动，这个速度就至关于硬盘IO的速度。

2．持久化数据存储

    可进行持久化操做。将消息持久化到磁盘，所以可用于批量消费，例如ETL，以及实时应用程序。经过将数据持久化到硬盘以及replication防止数据丢失。

3．分布式系统易于扩展

    全部的producer、broker和consumer均可以有多个，均为分布式的。无需停机便可扩展机器。

4．客户端维护状态

    消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。

2、基本概念

    Kafka将消息以topic为单位进行概括。topic之间的数据是相互隔离的。

    将向Kafka topic发布消息的程序称为producers。

    将预订topics并消费消息的程序称为consumer。

    Kafka以集群的方式运行，能够由一个或多个服务组成，每一个服务叫作一个broker。

    producers经过网络将消息发送到Kafka集群，集群向消费者提供消息。

    客户端和服务端经过TCP协议通讯。Kafka提供了Java客户端，而且对多种语言都提供了支持。

一、Topic

一个topic是对一组消息的概括。

1．分区

Kafka针对对每一个topic的日志进行了分区。

kafka中的分区是负载均衡和失败恢复的基本单位。

 

2．offset

    每一个分区都由一系列有序的、不可变的消息组成，这些消息被连续的追加到分区中。分区中的每一个消息都有一个连续的序列号叫作offset，用来在分区中惟一标识一个消息。在一个可配置的时间段内，Kafka集群保留全部发布的消息，无论这些消息有没有被消费。

    好比，若是消息的保存策略被设置为2天，那么在一个消息被发布的两天时间内，它都是能够被消费的。过时以后它将被丢弃以释放空间。Kafka的性能是和数据量无关的常量级的，因此保留太多的数据并非问题。

3．分区副本

    每一个分区在Kafka集群的若干服务中都有副本，这样这些持有副本的服务能够共同处理数据和请求，副本数量是能够配置的。副本使Kafka具有了容错能力。

    每一个分区都由一个服务器做为“leader”，零或若干服务器做为“followers”，leader负责处理消息的读和写，followers则去同步leader的数据，并对外提供读操做。若是leader down了，followers中的一台则会自动成为leader。

    集群中的每一个服务器都会同时扮演两个角色：做为它所持有的一部分分区的leader，同时做为其余分区的followers，这样集群就据有较好的负载均衡。

    将日志分区能够达到如下目的：

    首先这使得每一个日志的数量不会太大，能够在单个服务上保存。另外每一个分区能够单独发布和消费，为并发操做topic提供了一种可能。

    分区是负载均衡失败恢复分布式数据存储的基本单元。

二、Producers

    Producer将消息发布到它指定的topic中，并负责决定发布到哪一个分区。一般简单的由负载均衡机制随机选择分区，但也能够经过特定的分区函数选择分区。使用的更多的是第二种。

三、Consumers

    实际上每一个consumer惟一须要维护的数据是消息在日志中的位置，也就是offset。这个offset由consumer来维护，通常状况下随着consumer不断的读取消息，这offset的值会不断增长，但其实consumer能够以任意的顺序读取消息，好比它能够将offset设置成为一个旧的值来重读以前的消息。

    以上特色的结合，使Kafka consumers很是的轻量级，它们能够在不对集群和其余consumer形成影响的状况下读取消息。你可使用命令行来"tail"消息而不会对其余正在消费消息的consumer形成影响。

    消费消息一般有两种模式：队列模式（queuing）和发布-订阅模式(publish-subscribe)。

1．队列模式

    队列模式中，多个consumers能够同时从服务端读取消息，每一个消息只被其中一个consumer读到。

    通俗的来说，consumers之间是竞争关系，都在从borker中抢数据，而这个数据只有一份，谁抢到就是谁的。

2．发布订阅模式

    发布-订阅模式中消息被广播到全部的consumer中。这种模式之下，每一个consumers都能获得相同的消息数据。

3．consumer group

    Consumers能够加入一个consumer group，此组就是用来实现以上两种模式的。

1>组内

    若是全部的consumer都在一个组中，这就成为了传统的队列模式，在各consumer中实现负载均衡。

    组内的Consumer是处在队列模式下，共同竞争一个topic内的消息，topic中的消息将被分发到组中的一个成员中，同一条消息只发往其中的一个消费者。同一组中的consumer能够在不一样的程序中，也能够在不一样的机器上。

2>组间

    若是全部的consumer都不在不一样的组中，这就成为了发布-订阅模式，全部的消息都被分发到全部的consumer中。

    而若是有多个Consumer group来消费相同的Topic中的消息，则组和组之间使用的就是发布订阅模式，是一个共享数据的状态。每个组均可以获取到这个主题中的全部消息。

3>应用

    常见的应用方式是，每一个topic都有若干数量的consumer组来消费，每一个组都是一个逻辑上的“订阅者”，为了容错和更好的稳定性，每一个组都由若干consumer组成，在组内竞争实现负载均衡。实现了组内竞争负载均衡，组间共享互不影响，这其实就是一个发布-订阅模式，只不过订阅者是个组而不是单个consumer。 

四、与传统消息队列对比

    相比传统的消息系统，Kafka能够很好的保证有序性。

    传统的队列在服务器上保存有序的消息，若是多个consumers同时从这个服务器消费消息，服务器就会以消息存储的顺序向consumer分发消息。虽然服务器按顺序发布消息，可是消息是被异步的分发到各consumer上，因此当消息到达时可能已经失去了原来的顺序，这意味着并发消费将致使顺序错乱。为了不故障，这样的消息系统一般使用“专用consumer”的概念，其实就是只容许一个消费者消费消息，固然这就意味着失去了并发性。

    在这方面Kafka作的更好，经过分区的概念，Kafka能够在多个consumer组并发的状况下提供较好的有序性和负载均衡。将每一个分区分只分发给一个consumer组，这样一个分区就只被这个组的一个consumer消费，就能够顺序的消费这个分区的消息。由于有多个分区，依然能够在多个consumer组之间进行负载均衡。注意consumer组的数量不能多于分区的数量，也就是有多少分区就容许多少并发消费。

    Kafka只能保证一个分区以内消息的有序性，在不一样的分区之间是不能够的，这已经能够知足大部分应用的需求。若是须要topic中全部消息的有序性，那就只能让这个topic只有一个分区，固然也就只有一个consumer组消费它。

五、选择Kafka的理由

    为何大数据环境下的消息队列常选择kafka？

    分布式存储数据，提供了更好的性能、可靠性、可扩展能力。

    利用磁盘存储数据，且按照主题、分区来分布式存放数据，持久化存储，提供海量数据存储能力。

    采用磁盘存储数据，连续进行读写保证性能，性能和磁盘的性能相关和数据量的大小无关。

3、安装配置

一、下载安装

    下载Kafka安装包，上传到Linux服务器。

    解压：

tar -zxvf kafka_2.9.2-0.8.1.1.tgz

    解压完成就至关于安装完毕，不过还要进行响应的配置。

二、配置

1．伪分布式

1>server.properties

    修改server.properties文件。

log.dirs=/tmp/kafka-logs

    此选项配置的是Kafka的数据存储位置，须要更改。

2>zookeeper.properties

    修改zookeeper.properties配置文件。此文件是Kafka内置的Zookeeper的配置文件。Kafka为了保证软件的独立性，本身内置了一个Zookeeper，因此使用为分布式的状况下，不用专门安装Zookeeper。

    以下项配置的是zookeeper的数据存储位置，默认在/tmp中，须要修改。

dataDir=/tmp/zookeeper

3>启动kafka

    启动zookeeper：

bin/zookeeper-server-start.sh config/zookeeper.properties &

启动kafka：

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

2．彻底分布式

1>server.properties

    在config目录下，修改server.properties，在文件中修改以下参数：

broker.id=0 #当前server编号
port=9092 #使用的端口
log.dirs=/tmp/kafka-logs-1 #日志存储目录
zookeeper.connect=yun01:2181

    集群中broker.id要具备惟一性。

    日志存放目录须要更改成规划目录，不能使用默认的/tmp目录。

    Zookeeper须要配置Zookeeper集群中全部服务器的ip或主机名：端口。

2>启动

①启动zookeeper

    在各个机器上执行以下启动命令：

zkServer.sh start

②启动kafka

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

    Kafka服务默认启动会占用控制台，因此能够后台运行。

三、测试

1．建立topic

    建立一个拥有3个副本的topic:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic topicname

2．查看

1>查看主题

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

2>查看节点

    查看每一个节点的信息：

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

3．生产消息

    向topic发送消息：

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

4．消费消息

    消费消息：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic topicname

5．实验：容错性

    制造宕机，查看Kafka的容错性。

kill -9 7564
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topicname
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic topicname

    启动kafka:

    启动zookeeper：

zkServer.sh start

启动kafka：

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

4、使用kafka

一、sell操做

1．建立topic

bin/kafka-topics.sh --create --zookeeper localhost:9092 --replication-factor 1 --partitions 1 --topic test

2．查看topic

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

3．生产消息

使用命令行producer从文件中或者从标准输入中读取消息并发送到服务端：

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

4．消费消息

启动命令行consumer读取消息并输出到标准输出：

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

二、JavaAPI操做

1．搭建开发环境

    建立java工程，导入kafka相关包，jar包存在于Kafka安装包的libs目录中，拷贝时注意，里面不仅有jar包，还含有其余类型的文件，只拷贝jar包便可。

2．代码实现

1>消费者

/**
接收数据
*/
@Test
public void ConsumerReceive() throws Exception{
  Properties properties = new Properties();  
  properties.put("zookeeper.connect", "yun01:2181,yun02:2181,yun03:2181");//声明zk  
  // 必需要使用别的组名称， 若是生产者和消费者都在同一组，则不能访问同一组内的topic数据  
  properties.put("group.id", "group2xx");
  properties.put("auto.offset.reset", "smallest");
  // properties.put("zookeeper.session.timeout.ms", "400");
  // properties.put("zookeeper.sync.time.ms", "200");
  // properties.put("auto.commit.interval.ms", "1000");
  // properties.put("serializer.class", "kafka.serializer.StringEncoder");
  ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
  Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
  topicCountMap.put("my-replicated-topic", 1); // 一次从主题中获取一个数据  
  Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
  // 获取每次接收到的这个数据 
  KafkaStream<byte[], byte[]> stream = messageStreams.get("my-replicated-topic").get(0); 
  ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
  while(iterator.hasNext()){
    System.out.println("receive：" + new String(iterator.next().message()));
  }
}

2>生产者

/**
发送数据
*/
@Test
public void ProducerSend(){
  Properties props = new Properties();
  props.put("serializer.class", "kafka.serializer.StringEncoder");
  props.put("metadata.broker.list", "192.168.242.101:9092");
  Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(props ));
  producer.send(new KeyedMessage<Integer, String>("my-replicated-topic","message~xxx123asdf"));
  producer.close();
}

下一篇：Kafka的存储机制以及可靠性