Kafka基本使用介绍

转自： https://blog.csdn.net/wing_93/article/details/78513782

 

一、Kafka概述

Kafka® is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

这是Kafka官网里的一段介绍，从这段话，我们可以知道，kafka在构建实时数据管道和streaming应用中被常使用，它可以横向扩展，容错，快速地运行在数以千计的公司产品上。

它可以像消息系统一样读写数据流，并且可以在实时业务的场景中写可靠的流处理应用，并且能安全地存储数据流到分布式、多副本、容错的集群中。所以通俗点理解，可以说Kafka就是一个消息中间件。

Kafka场景比喻

接下来我大概比喻下Kafka的使用场景

消息中间件：生产者和消费者

妈妈：生产者
你：消费者
馒头：数据流、消息
正常情况下： 生产一个  消费一个
其他情况：  
1）一直生产，你吃到某一个馒头时，你卡主(机器故障)， 馒头就丢失了
2）一直生产，做馒头速度快，你吃来不及，馒头也就丢失了
为了放着其他生产情况的出现，我们可以拿个碗/篮子，馒头做好以后先放到篮子里，你要吃的时候去篮子里面取出来吃，而这篮子/框就可以为：Kafka。当篮子满了，馒头就装不下了，咋办？ 多准备几个篮子 === Kafka的扩容

二、Kafka的架构和核心概念

这是张我在Kafka官网上截的图，我大概可以把Kafka的主要结构分为以下几点：

producer：生产者，就是生产馒头(老妈)
consumer：消费者，就是吃馒头的(你)
broker：篮子
topic：主题，给馒头带一个标签，topica的馒头是给你吃的，topicb的馒头是给你弟弟吃

另外，我们在看下官网上的这段解释：

 

First a few concepts:

• Kafka is run as a cluster on one or more servers.
• The Kafka cluster stores streams of records in categories called topics.
• Each record consists of a key, a value, and a timestamp.

可以看出来：

1.Kafka可以作为集群运行在一台或者多个服务器上面；

2.Kafka集群可以分类地存储记录流，以打标签的方式，就是采用topics，每个broker可以打个topic，这样能保证消费者可以根据topic选择性消费；

3.每个记录由Key、Value、timestamp构成。

Kafka四个核心的API

1.ProducerAPI：允许一个应用向一个或多个topic里发布记录流；

2.ConsumerAPI：允许一个应用订阅一个或多个topics，处理topic里的数据流，就相当于消费；

3.StreamAPI：允许应用扮演流处理的作用，从一个或多个topic里消费数据流，然后产生输出流数据到其他一个或多个topic里，对输入流数据有效传输到输出口；

4.ConnectorAPI：允许运行和构建一个可重复利用的生产者和消费者，能将kafka的topic与其他存在的应用和数据库设备相连接，比如链接一个实时数据库，可以捕捉到每张表的变化。

这四个API，主要应用在IDEA上对应用程序的开发中，通过代码的形式管理Kafka。在第四部分将会对前两个API写个简单DEMO演示。

三、Kafka的快速使用

Kafka使用到了zookeeper，所以首先你得安装zookeeper再安装kafka。

1.单节点的broker部署

首先我们需要修改$KAFKA_HOME/config/server.properties这个配置文件，主要以下几处需要修改：
broker.id=0，每个broker的ID需要唯一
listeners：监听的端口（此处笔者设置的是默认端口9092）
host.name：当前机器
log.dirs：存储日志的文件夹

num.partitions：分区的数量
zookeeper.connect：zookeeper的地址（默认为localhost：2181）

这几处根据你自身需要进行配置，然后启动步骤如下：

1）开启zookeeper，此处需要注意的是zookeeper的conf目录下的zoo.cfg配置文件，主要修改的也是日志存储目录那块。

2）启动Kafka，命令为：kafka-server-start.sh $KAFKA_HOME/config/server.properties

3）创建topic，需要指定zookeeper，命令为：kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic hello_topic。 注意指定zookeeper，后面几个属性可以根据你实际情况进行定义。另外查看所有topic的命令为：
kafka-topics.sh --list --zookeeper hadoop000:2181

4）发送消息，需要指定broker，命令为：kafka-console-producer.sh --broker-list hadoop000:9092 --topic hello_topic

5）消费消息，需要指定zookeeper，命令为：kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic --from-beginning。意思就是指定zookeeper上的topic进行消费，from-beginning的设置，可以查看之前的消息。

2.单节点，多broker

主要是增加多个server.properties文件，一个配置文件就相当于一个broker，我就设置三个broker：

 

1. server-1.properties

2. log.dirs=/home/hadoop/app/tmp/kafka-logs-1

3. listeners=PLAINTEXT://:9093

4. broker.id=1

5.  

6. server-2.properties

7. log.dirs=/home/hadoop/app/tmp/kafka-logs-2

8. listeners=PLAINTEXT://:9094

9. broker.id=2

10.  

11. server-3.properties

12. log.dirs=/home/hadoop/app/tmp/kafka-logs-3

13. listeners=PLAINTEXT://:9095

14. broker.id=3

然后依次开启，命令如下：

 

1. kafka-server-start.sh -daemon $KAFKA_HOME/config/server-1.properties &

2. kafka-server-start.sh -daemon $KAFKA_HOME/config/server-2.properties &

3. kafka-server-start.sh -daemon $KAFKA_HOME/config/server-3.properties &

4.  

 接下来就跟上面的步骤一样：

 

1. kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

2.  

3. kafka-console-producer.sh --broker-list hadoop000:9093,hadoop000:9094,hadoop000:9095 --topic my-replicated-topic

4. kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic my-replicated-topic

 查看 topic的详细信息：

 

kafka-topics.sh --describe --zookeeper hadoop000:2181 --topic my-replicated-topic

要注意的是，副本中会有个leader，而多副本也实现了kafka的容错性，挂掉一个副本后，会自动在剩下副本里选出一个leader来同步操作。

 

根据上面步骤操作，我们在producer窗口输入，在consumer消费窗口看到相应输出。

四、Producer和Consumer API的使用

接下来展示一个简单的Demo，在生产端简单创建个线程进行循环输出，然后用消费者端对输出的内容进行展示，也就是消费。

配置文件

1. /**

2. * Kafka常用配置文件

3. */

4. public class KafkaProperties {

5.  

6. public static final String ZK = "192.168.199.111:2181";

7.  

8. public static final String TOPIC = "hello_topic";

9.  

10. public static final String BROKER_LIST = "192.168.199.111:9092";

11.  

12. public static final String GROUP_ID = "test_group1";

13.  

14. }

 Producer API DEMO

1. import kafka.javaapi.producer.Producer;

2. import kafka.producer.KeyedMessage;

3. import kafka.producer.ProducerConfig;

4.  

5. import java.util.Properties;

6.  

7. /**

8. * Kafka生产者

9. */

10. public class KafkaProducer extends Thread{

11.  

12. private String topic;

13.  

14. private Producer<Integer, String> producer;

15.  

16. public KafkaProducer(String topic) {

17. this.topic = topic;

18.  

19. Properties properties = new Properties();

20.  

21. properties.put("metadata.broker.list",KafkaProperties.BROKER_LIST);

22. properties.put("serializer.class","kafka.serializer.StringEncoder");

23. properties.put("request.required.acks","1");

24.  

25. producer = new Producer<Integer, String>(new ProducerConfig(properties));

26. }

27.  
28.  

29. @Override

30. public void run() {

31.  

32. int messageNo = 1;

33.  

34. while(true) {

35. String message = "message_" + messageNo;

36. producer.send(new KeyedMessage<Integer, String>(topic, message));

37. System.out.println("Sent: " + message);

38.  

39. messageNo ++ ;

40.  

41. try{

42. Thread.sleep(2000);

43. } catch (Exception e){

44. e.printStackTrace();

45. }

46. }

47.  

48. }

49. }

Consumer API DEMO

1. import kafka.consumer.Consumer;

2. import kafka.consumer.ConsumerConfig;

3. import kafka.consumer.ConsumerIterator;

4. import kafka.consumer.KafkaStream;

5. import kafka.javaapi.consumer.ConsumerConnector;

6.  

7. import java.util.HashMap;

8. import java.util.List;

9. import java.util.Map;

10. import java.util.Properties;

11.  

12. /**

13. * Kafka消费者

14. */

15. public class KafkaConsumer extends Thread{

16.  

17. private String topic;

18.  

19. public KafkaConsumer(String topic) {

20. this.topic = topic;

21. }

22.  
23.  

24. private ConsumerConnector createConnector(){

25. Properties properties = new Properties();

26. properties.put("zookeeper.connect", KafkaProperties.ZK);

27. properties.put("group.id",KafkaProperties.GROUP_ID);

28. return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));

29. }

30.  

31. @Override

32. public void run() {

33. ConsumerConnector consumer = createConnector();

34.  

35. Map<String, Integer> topicCountMap = new HashMap<String, Integer>();

36. topicCountMap.put(topic, 1);

37. // topicCountMap.put(topic2, 1);

38. // topicCountMap.put(topic3, 1);

39.  

40. // String: topic

41. // List<KafkaStream<byte[], byte[]>> 对应的数据流

42. Map<String, List<KafkaStream<byte[], byte[]>>> messageStream = consumer.createMessageStreams(topicCountMap);

43.  

44. KafkaStream<byte[], byte[]> stream = messageStream.get(topic).get(0); //获取我们每次接收到的暑假

45.  

46. ConsumerIterator<byte[], byte[]> iterator = stream.iterator();

47.  
48.  

49. while (iterator.hasNext()) {

50. String message = new String(iterator.next().message());

51. System.out.println("rec: " + message);

52. }

53. }

54. }

最后在main函数对这两个类调用即可，结果如下：

 

 

--------------------- 本文来自 疯兔子大叔 的CSDN 博客 ，全文地址请点击：https://blog.csdn.net/wing_93/article/details/78513782?utm_source=copy