专栏首页未读代码的专栏消息队列中间件(三)Kafka 入门指南

消息队列中间件(三)Kafka 入门指南

Kafka 来源

Kafka的前身是由LinkedIn开源的一款产品,2011年初开始开源,加入了 Apache 基金会,2012年从 Apache Incubator 毕业变成了 Apache 顶级开源项目。同时LinkedIn还有许多著名的开源产品。如:

  • 分布式数据同步系统Databus
  • 高性能计算引擎Cubert
  • Java异步处理框架ParSeq
  • Kafka流处理平台

Kafka 介绍

Kafka 用于构建实时数据管道和流应用程序。它具有水平可扩展性,容错性,快速性,并在数千家公司的生产环境中运行。

从官方我们可以知道ApacheKafka一个分布式流媒体平台。这到底是什么意思呢?

流媒体平台有三个关键功能:

  • 发布和订阅记录数据流,类似于消息队列或企业消息传递系统。
  • 有容错能力的可以持久化的存储数据流。
  • 记录发生时可以进行流处理。

Kafka 通常用于两大类应用:

  • 构建可在系统或应用程序之间可靠获取数据的实时流数据管道
  • 构建转换或响应数据流的实时流处理

Kafka 基本概念

  • Producer - 消息和数据的生产者,向 Kafka 的一个 Topic 发布消息的进程/代码/服务。
  • Consumer - 消息和数据的消费者,订阅数据(Topic)并且处理其发布的消息的进程/代码/服务。
  • Consumer Group - 逻辑概念,对于同一个 Topic,会广播不同的 Group,一个Group中,只有一个consumer 可以消费该消息。
  • Broker - 物理概念,Kafka 集群中的每个 Kafka 节点。
  • Topic - 逻辑概念,Kafka消息的类别,对数据进行区分,隔离。
  • Partition - 物理概念,分片,Kafka 下数据存储的基本单元,一个 Topic 数据,会被分散存储到多个Partition,每一个Partition是有序的。
  • Replication - 副本,同一个 Partition 可能会有多个 Replica ,多个 Replica 之间数据是一样的。
  • Replication Leader - 一个 Partition 的多个 Replica 上,需要一个 Leade r负责该 Partition 上与 Produce 和 Consumer 交互
  • ReplicaManager - 负责管理当前的 broker 所有分区和副本的信息,处理 KafkaController 发起的一些请求,副本状态的切换,添加/读取消息等。

概念的延伸

Partition

  • 每一个Topic被切分为多个Partitions
  • 消费者数据要小于等于Partition的数量
  • Broker Group中的每一个Broker保存Topic的一个或多个Partitions
  • Consumer Group中的仅有一个Consumer读取Topic的一个或多个Partions,并且是唯一的Consumer。

Replication

  • 当集群中有Broker挂掉的时候,系统可以主动的使用Replicas提供服务。
  • 系统默认设置每一个Topic的Replication的系数为1,可以在创建Topic的时候单独设置。

Replication特点

  • Replication的基本单位是Topic的Partition。
  • 所有的读和写都从Leader进,Followers只是作为备份。
  • Follower必须能够及时的复制Leader的数据
  • 增加容错性与可扩展性。

Kafka 消息结构

在 Kafka2.0 中的消息结构如下(整理自官网)。

baseOffset: int64 - 用于记录Kafka这个消息所处的偏移位置 batchLength: int32 - 用于记录整个消息的长度 partitionLeaderEpoch: int32 magic: int8 (current magic value is 2) - 一个固定值,用于快速判断是否是Kafka消息 crc: int32 - 用于校验信息的完整性 attributes: int16 - 当前消息的一些属性 bit 0~2: 0: no compression 1: gzip 2: snappy 3: lz4 bit 3: timestampType bit 4: isTransactional (0 means not transactional) bit 5: isControlBatch (0 means not a control batch) bit 6~15: unused lastOffsetDelta: int32 firstTimestamp : int64 maxTimestamp: int64 producerId: int64 producerEpoch: int16 baseSequence: int32 records: length: varint attributes: int8 bit 0~7: unused timestampDelta: varint offsetDelta: varint keyLength: varint key: byte[] valueLen: varint value: byte[] Headers => [Header] headerKeyLength: varint headerKey: String headerValueLength: varint Value: byte[]

关于消息结构的一些释义。

  • Offset -用于记录Kafka这个消息所处的偏移位置
  • Length - 用于记录整个消息的长度
  • CRC32 - 用于校验信息的完整性
  • Magic - 一个固定值,用于快速判断是否是Kafka消息
  • Attributes - 当前消息的一些属性
  • Timestamp - 消息的时间戳
  • Key Length - key的长度
  • Key - Key的具体值
  • Value Length - 值的长度
  • Value - 具体的消息值

Kafka 优点

  1. 分布式 - Kafka是分布式的,多分区,多副本的和多订阅者的,基于Zookeeper调度。
  2. 持久性和扩展性 - Kafka使用分布式提交日志,这意味着消息会尽可能快地保留在磁盘上,因此它是持久的。同时具有一定的容错性,Kafka支持在线的水平扩展,消息的自平衡。
  3. 高性能 - Kafka对于发布和订阅消息都具有高吞吐量。 即使存储了许多TB的消息,它也保持稳定的性能。且延迟低,适用高并发。时间复杂的为o(1)。

Kafka 应用

  1. 用于聚合分布式应用程序中的消息。进行操作监控。
  2. 用于跨组织的从多个服务收集日志,然后提供给多个服务器,解决日志聚合问题。
  3. 用于流处理,如Storm和Spark Streaming,从kafka中读取数据,然后处理在写入kafka供应用使用。

Kafka 安装

安装 Jdk

具体步骤此处不说。

安装 Kafka

直接官方网站下载对应系统的版本解压即可。 由于Kafka对于windows和Unix平台的控制脚本是不同的,因此如果是windows平台,要使用bin\windows\而不是bin/,并将脚本扩展名更改为.bat。以下命令是基于Unix平台的使用。

# 解压
tar -xzf kafka_2.11-2.0.0.tgz
# 启动Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
# 启动Kafka
bin/kafka-server-start.sh config/server.properties
# 或者后台启动
bin/kafka-server-start.sh config/server.properties &

让我们创建一个名为“test”的主题,它只包含一个分区,只有一个副本:

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

如果我们运行list topic命令,我们现在可以看到该主题:

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

或者,您也可以将代理配置为在发布不存在的主题时自动创建主题,而不是手动创建主题。

查看Topic的信息

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic Hello-Kafka

运行生产者,然后在控制台中键入一些消息以发送到服务器。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
This is a message
This is another message`

运行消费者,查看收到的消息

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
> This is a message
> This is another message

Kafka 工程实例

POM 依赖

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.1.0</version></dependency>

生产者

编写生产者 Java 代码。关于 Properties 中的值的意思描述可以在官方文档中找到 http://kafka.apache.org/ 。下面的生产者向 Kafka 推送了10条消息。

import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.Producer;import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;/**
* <p>
* Kafka生产者,发送10个数据
*
* @Author niujinpeng
* @Date 2018/11/16 15:45
*/public class MyProducer {    public static void main(String[] args) {
       Properties props = new Properties();
       props.put("bootstrap.servers", "192.168.110.132:9092");
       props.put("acks", "all");
       props.put("retries", 0);
       props.put("batch.size", 16384);
       props.put("linger.ms", 1);
       props.put("buffer.memory", 33554432);
       props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
       props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");       Producer<String, String> producer = new KafkaProducer<>(props);        for (int i = 0; i < 10; i++) {
           producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i)));
       }
       producer.close();
   }}

消费者

编写消费者 Java 代码。

import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Arrays;import java.util.Properties;/**
* <p>
* Kafka消费者
*
* @Author niujinpeng
* @Date 2018/11/19 15:01
*/public class MyConsumer {    public static void main(String[] args) {       Properties props = new Properties();
       props.put("bootstrap.servers", "192.168.110.132:9092");
       props.put("group.id", "test");
       props.put("enable.auto.commit", "true");
       props.put("auto.commit.interval.ms", "1000");
       props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
       consumer.subscribe(Arrays.asList("test"));        while (true) {
           ConsumerRecords<String, String> records = consumer.poll(100);            for (ConsumerRecord<String, String> record : records) {
               System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
           }
       }
   }}

可以在控制台看到成功运行后的输出,由 offset 可以看到已经消费了10条消息。

 INFO | Kafka version : 2.0.0
INFO | Kafka commitId : 3402a8361b734732
INFO | Cluster ID: 0Xrk5M1CSJet0m1ut3zbiw
INFO | [Consumer clientId=consumer-1, groupId=test] Discovered group coordinator 192.168.110.132:9092 (id: 2147483647 rack: null)
INFO | [Consumer clientId=consumer-1, groupId=test] Revoking previously assigned partitions []
INFO | [Consumer clientId=consumer-1, groupId=test] (Re-)joining group
INFO | [Consumer clientId=consumer-1, groupId=test] Successfully joined group with generation 4
INFO | [Consumer clientId=consumer-1, groupId=test] Setting newly assigned partitions [test-0]
offset = 38, key = 0, value = 0
offset = 39, key = 1, value = 1
offset = 40, key = 2, value = 2
offset = 41, key = 3, value = 3
offset = 42, key = 4, value = 4
offset = 43, key = 5, value = 5
offset = 44, key = 6, value = 6
offset = 45, key = 7, value = 7
offset = 46, key = 8, value = 8
offset = 47, key = 9, value = 9

问题

如果java.net.InetAddress.getCanonicalHostName 取到的是主机名。需要修改 Kafka 的配置文件。

vim server.properties
# x.x.x.x是服务器IP
advertised.listeners=PLAINTEXT://x.x.x.x:9092

本文分享自微信公众号 - 未读代码(weidudaima),作者:雪漫士兵

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-12-12

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Springboot 系列(五)Spring Boot web 开发之静态资源和模版引擎

    Spring Boot 天生的适合 web 应用开发,它可以快速的嵌入 Tomcat, Jetty 或 Netty 用于包含一个 HTTP 服务器。且开发十分简...

    未读代码
  • 关于编译器与解释器

    为了让更多的人能够从本质上理解编译器和解释器的区别,我杜撰了一个小故事 来福与旺财的养牛场 来福和旺财有一个养 牛场。本来养牛不是一件太难的事情,但是偏偏他...

    未读代码
  • 消息队列中间件(二)使用 ActiveMQ

    Active MQ 是由 Apache 出品的一款流行的功能强大的开源消息中间件,它速度快,支持跨语言的客户端,具有易于使用的企业集成模式和许多的高级功能,同时...

    未读代码
  • kafka 主要内容介绍

    根据官网的介绍,ApacheKafka®是一个分布式流媒体平台,它主要有3种功能:

    庞小明
  • 理解Scala的Symbol类型

    joymufeng
  • input文本框输入时正则判断

    奋斗蒙
  • Leetcode 686. 重复叠加字符串匹配

    给定两个字符串 A 和 B, 寻找重复叠加字符串A的最小次数,使得字符串B成为叠加后的字符串A的子串,如果不存在则返回 -1。

    zhipingChen
  • 9分钟了解-Es6中的symbol

    从 ES6 起,JavaScript 的 基础数据类型 变为 6 个:string, number, boolean, null, undefined, sym...

    web前端教室
  • kafka的发行版选择

    今天继续和大家聊一下,kafka的各种发行版。kafka历经数年的发展,从最初纯粹的消息引擎,到近几年开始在流处理平台生态圈发力,衍生出了各种不同特性的版本。

    一条老狗
  • 【视频】Es6新特性-Symbol

    温馨提示:视频请点此观看 视频原文: es6 中的symbol [ˈsɪmbl] symbol 是 ES6 的一个新特性 symbol 是一个 “新” 的...

    web前端教室

扫码关注云+社区

领取腾讯云代金券