Kafka流处理平台总结-元一软件

第一章课程介绍1-1 课程介绍课程介绍Kafka概念解析Kafka结构设计Kafka场景应用Kafka高级特性第二章概念解析2-1 发展背景LinkedIn 开源Databus 分布式数据同步系统Cubert 高性能计算引擎ParSeq Java异步处理框架Kafka 分布式发布订阅消息系统流处理平台Kafka发展历程LinkedIn 开发2011年初开源加入Apache基金会2012年从Apache Incubator毕业Apache顶级开源项目Kafka的特性可以发布和订阅且记录数据的流类似于消息队列数据流存储的平台具备容错能力在数据产生时就可以进行处理Kafka通常被用于构建实时数据流管道构建实时数据流处理Kafka是什么面向于数据流的生产、转换、存储、消费整体的流处理平台Kafka不仅仅是一个消息队列2-2 基本概念Producer数据生产者消息和数据的生产者向Kafka的一个topic发布消息的进程或代码或服务Consumer数据消费者消息和数据的消费者向Kafka订阅数据topic并且处理其发布的消息的进程或代码或服务Consumer Group消费者组对于同一个topic会广播给不同的Group一个Group中只有一个Consumer可以消费该消息Broker服务节点Kafka集群中的每个Kafka节点Topic主题Kafka消息的类别对数据进行区分、隔离Partition分区Kafka中数据存储的基本单元一个topic数据会被分散存储到多个Partition一个Partition只会存在一个Broker上每个Partition是有序的Replication分区的副本同一个Partition可能会有多个Replication多个Replication之间数据是一样的Replication Leader副本的老大一个Partition的多个Replication上需要一个Leader负责该Partition上与Producer和Consumer交互Replication Manager副本的管理者负责管理当前Broker所有分区和副本的信息处理KafkaController发起的一些请求副本状态的切换添加、读取消息等2-3 概念延伸Partition分区每一个Topic被切分为多个Partition消费者数目少于或等于Partition的数目Broker Group中的每一个Broker保存Topic的一个或多个PartitionConsumer Group中的仅有一个Consumer读取Topic的一个或多个Partition并且是惟一的ConsumerReplication分区的副本当集群中有Broker挂掉的情况系统可以主动地使Replication提供服务系统默认设置每一个Topic的Replication系数为1可以在创建Topic时单独设置Replication的基本单位是Topic的Partition所有的读和写都从Replication Leader进行Replication Followers只是作为备份Replication Followers必须能够及时复制Replication Leader的数据增加容错性与可扩展性第三章结构设计3-1 基本结构Kafka功能结构Kafka数据流势Kafka消息结构Offset当前消息所处于的偏移Length消息的长度CRC32校验字段用于校验当前信息的完整性Magic很多分布式系统都会设计该字段固定的数字用于快速判定当前信息是否为Kafka消息attributes可选字段消息的属性Timestamp时间戳Key LengthKey的长度KeyKeyValue LengthValue的长度ValueValue3-2 功能特点Kafka特点分布式多分区多副本多订阅者基于Zookeeper调度Kafka特点高性能高吞吐量低延迟高并发时间复杂度为O(1)Kafka特点持久性与扩展性数据可持久化容错性支持在线水平扩展消息自动平衡第四章场景应用4-1 应用场景Kafka应用场景消息队列行为跟踪元信息监控日志收集流处理事件源持久性日志commit log4-2 应用案例Kafka简单案例部署启动简单生产者简单消费者学习笔记1.下载与安装 Zookeeper下载https://zookeeper.apache.org/releases.html#download Kafka下载http://kafka.apache.org/downloads 安装解压、配置环境变量 2.Zookeeper启动 解压tar -zxf zookeeper-3.4.12.tar.gz 目录cd zookeeper-3.4.12/bin 启动./zkServer.sh start /home/zc/server/kafka_2.12-2.0.0/config/zookeeper.properties 3.Kafka启动 解压tar -zxf kafka_2.12-2.0.0.tgz 目录cd kafka_2.12-2.0.0 启动sudo bin/kafka-server-start.sh config/server.properties 4.使用控制台操作生产者与消费者 创建Topicsudo ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic myimooc-kafka-topic 查看Topicsudo ./bin/kafka-topics.sh --list --zookeeper localhost:2181 启动生产者sudo ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic myimooc-kafka-topic 启动消费者sudo ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myimooc-kafka-topic --from-beginning 生产消息first message 生产消息second message4-3 代码案例创建49-kafka-example的maven工程pom如下?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd parent artifactId49-kafka/artifactId groupIdcom.myimooc/groupId version1.0-SNAPSHOT/version /parent modelVersion4.0.0/modelVersion artifactId49-kafka-example/artifactId properties spring.boot.version2.0.4.RELEASE/spring.boot.version /properties dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-parent/artifactId version${spring.boot.version}/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.kafka/groupId artifactIdspring-kafka/artifactId /dependency dependency groupIdcom.alibaba/groupId artifactIdfastjson/artifactId version1.2.36/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project1.编写MessageEntitypackage com.myimooc.kafka.example.common; import java.util.Objects; /** * br * 标题: 消息实体br * 描述: 消息实体br * 时间: 2018/09/09br * * author zc */ public class MessageEntity { /** * 标题 */ private String title; /** * 内容 */ private String body; Override public String toString() { return MessageEntity{ title title \ , body body \ }; } Override public boolean equals(Object o) { if (this o) { return true; } if (o null || getClass() ! o.getClass()) { return false; } MessageEntity that (MessageEntity) o; return Objects.equals(title, that.title) Objects.equals(body, that.body); } Override public int hashCode() { return Objects.hash(title, body); } public String getTitle() { return title; } public void setTitle(String title) { this.title title; } public String getBody() { return body; } public void setBody(String body) { this.body body; } }2.编写SimpleProducerpackage com.myimooc.kafka.example.producer; import com.alibaba.fastjson.JSON; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; /** * br * 标题: 生产者br * 描述: 生产者br * 时间: 2018/09/09br * * author zc */ Component public class SimpleProducerT { private Logger logger LoggerFactory.getLogger(getClass()); Autowired private KafkaTemplateString, Object kafkaTemplate; public void send(String topic, String key, Object entity) { logger.info(发送消息入参{}, entity); ProducerRecordString, Object record new ProducerRecord( topic, key, JSON.toJSONString(entity) ); long startTime System.currentTimeMillis(); ListenableFutureSendResultString, Object future this.kafkaTemplate.send(record); future.addCallback(new ListenableFutureCallbackSendResultString, Object() { Override public void onFailure(Throwable ex) { logger.error(消息发送失败{}, ex); } Override public void onSuccess(SendResultString, Object result) { long elapsedTime System.currentTimeMillis() - startTime; RecordMetadata metadata result.getRecordMetadata(); StringBuilder record new StringBuilder(128); record.append(message() .append(key ).append(key).append(,) .append(message ).append(entity).append()) .append(send to partition().append(metadata.partition()).append()) .append(with offset().append(metadata.offset()).append()) .append(in ).append(elapsedTime).append( ms); logger.info(消息发送成功{}, record.toString()); } }); } }3.编写SimpleConsumerpackage com.myimooc.kafka.example.consumer; import com.alibaba.fastjson.JSONObject; import com.myimooc.kafka.example.common.MessageEntity; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.messaging.handler.annotation.Header; import org.springframework.stereotype.Component; import java.util.Optional; /** * br * 标题: 消费者br * 描述: 消费者br * 时间: 2018/09/09br * * author zc */ Component public class SimpleConsumer { private Logger logger LoggerFactory.getLogger(getClass()); KafkaListener(topics ${kafka.topic.default}) public void listen(ConsumerRecord?, ? record, Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { //判断是否NULL Optional? kafkaMessage Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { //获取消息 Object message kafkaMessage.get(); MessageEntity messageEntity JSONObject.parseObject(message.toString(), MessageEntity.class); logger.info(接收消息Topic{}, topic); logger.info(接收消息Record{}, record); logger.info(接收消息Message{}, messageEntity); } } }4.编写Responsepackage com.myimooc.kafka.example.common; import java.io.Serializable; /** * br * 标题: REST请求统一响应对象br * 描述: REST请求统一响应对象br * 时间: 2018/09/09br * * author zc */ public class Response implements Serializable { private static final long serialVersionUID -972246069648445912L; /** * 响应编码 */ private int code; /** * 响应消息 */ private String message; public Response() { } public Response(int code, String message) { this.code code; this.message message; } Override public String toString() { return Response{ code code , message message \ }; } public int getCode() { return code; } public void setCode(int code) { this.code code; } public String getMessage() { return message; } public void setMessage(String message) { this.message message; } }5.编写ErrorCodepackage com.myimooc.kafka.example.common; /** * br * 标题: 错误编码br * 描述: 错误编码br * 时间: 2018/09/09br * * author zc */ public class ErrorCode { /** * 成功 */ public final static int SUCCESS 200; /** * 失败 */ public final static int EXCEPTION 500; }6.编写ProducerControllerpackage com.myimooc.kafka.example.controller; import com.alibaba.fastjson.JSON; import com.myimooc.kafka.example.common.ErrorCode; import com.myimooc.kafka.example.common.MessageEntity; import com.myimooc.kafka.example.common.Response; import com.myimooc.kafka.example.producer.SimpleProducer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.web.bind.annotation.*; /** * br * 标题: 生产者Controllerbr * 描述: 生产者Controllerbr * 时间: 2018/09/09br * * author zc */ RestController RequestMapping(/producer) public class ProducerController { private Logger logger LoggerFactory.getLogger(getClass()); Autowired private SimpleProducer simpleProducer; Value(${kafka.topic.default}) private String topic; private static final String KEY key; PostMapping(/send) public Response sendKafka(RequestBody MessageEntity message) { try { logger.info(kafka的消息{}, JSON.toJSONString(message)); this.simpleProducer.send(topic, KEY, message); logger.info(kafka消息发送成功); return new Response(ErrorCode.SUCCESS,kafka消息发送成功); } catch (Exception ex) { logger.error(kafka消息发送失败, ex); return new Response(ErrorCode.EXCEPTION,kafka消息发送失败); } } }7.编写application.properties##----------kafka配置 ## TOPIC kafka.topic.defaultmyimooc-kafka-topic # kafka地址 spring.kafka.bootstrap-servers192.168.0.105:9092 # 生产者配置 spring.kafka.producer.retries0 # 批量发送消息的数量 spring.kafka.producer.batch-size4096 # 缓存容量 spring.kafka.producer.buffer-memory40960 # 指定消息key和消息体的编解码方式 spring.kafka.producer.key-serializerorg.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializerorg.apache.kafka.common.serialization.StringSerializer # 消费者配置 spring.kafka.consumer.group-idmyimooc spring.kafka.consumer.auto-commit-interval100 spring.kafka.consumer.auto-offset-resetlatest spring.kafka.consumer.enable-auto-committrue # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializerorg.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializerorg.apache.kafka.common.serialization.StringDeserializer # 指定listener 容器中的线程数用于提高并发量 spring.kafka.listener.concurrency38.编写ExampleApplicationpackage com.myimooc.kafka.example; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.kafka.annotation.EnableKafka; /** * br * 标题: 启动类br * 描述: 启动类br * 时间: 2018/09/09br * * author zc */ SpringBootApplication EnableKafka public class ExampleApplication { public static void main(String[] args) { SpringApplication.run(ExampleApplication.class, args); } }第五章高级特性5-1 消息事务为什么要支持事务满足“读取-处理-写入”模式流处理需求的不断增强不准确的数据处理的容忍度不断降低数据传输的事务定义最多一次消息不会被重复发送最多被传输一次但也有可能一次不传输最少一次消息不会被漏发送最少被传输一次但也有可能被重复传输精确的一次Exactly once不会漏传输也不会重复传输每个消息都被传输一次且仅仅被传输一次这是大家所期望的事务保证内部重试问题Procedure幂等处理多分区原子写入避免僵尸实例每个事务Procedure分配一个 transactionl. id在进程重新启动时能够识别相同的Procedure实例Kafka增加了一个与transactionl.id相关的epoch存储每个transactionl.id内部元数据一旦epoch被触发任务具有相同的transactionl.id和更旧的epoch的Producer被视为僵尸Kafka会拒绝来自这些Producer的后续事务性写入5-2 零拷贝零拷贝简介网络传输持久性日志块Java Nio channel.transforTo()方法Linux sendfile系统调用文件传输到网络的公共数据路径第一次拷贝操作系统将数据从磁盘读入到内核空间的页缓存第二次拷贝应用程序将数据从内核空间读入到用户空间缓存中第三次拷贝应用程序将数据写回到内核空间到socket缓存中第四次拷贝操作系统将数据从socket缓冲区复制到网卡缓冲区以便将数据经网络发出零拷贝过程指内核空间和用户空间的交互拷贝次数为零第一次拷贝操作系统将数据从磁盘读入到内核空间的页缓存将数据的位置和长度的信息的描述符增加至内核空间socket缓存区第二次拷贝操作系统将数据从内核拷贝到网卡缓冲区以便将数据经网络发出文件传输到网络的公共数据路径演变