Kafka 基本原理

简介

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发,之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的,分区的和可复制的提交日志服务。

Kafka架构

它的架构包括以下组件:

  • 话题(Topic):是特定类型的消息流。消息是字节的有效负载(Payload),话题是消息的分类名或种子(Feed)名。
  • 生产者(Producer):是能够发布消息到话题的任何对象。
  • 服务代理(Broker):已发布的消息保存在一组服务器中,它们被称为代理(Broker)或Kafka集群。
  • 消费者(Consumer):可以订阅一个或多个话题,并从Broker拉数据,从而消费这些已发布的消息。

Kafka存储策略

1)kafka以topic来进行消息管理,每个topic包含多个partition,每个partition对应一个逻辑log,有多个segment组成。

2)每个segment中存储多条消息(见下图),消息id由其逻辑位置决定,即从消息id可直接定位到消息的存储位置,避免id到位置的额外映射。

3)每个part在内存中对应一个index,记录每个segment中的第一条消息偏移。

4)发布者发到某个topic的消息会被均匀的分布到多个partition上(或根据用户指定的路由规则进行分布),broker收到发布消息往对应partition的最后一个segment上添加该消息,当某个segment上的消息条数达到配置值或消息发布时间超过阈值时,segment上的消息会被flush到磁盘,只有flush到磁盘上的消息订阅者才能订阅到,segment达到一定的大小后将不会再往该segment写数据,broker会创建新的segment。

Kafka删除策略

1)N天前的删除。

2)保留最近的MGB数据。

Kafka broker

与其它消息系统不同,Kafka broker是无状态的。这意味着消费者必须维护已消费的状态信息。这些信息由消费者自己维护,broker完全不管(有offset managerbroker管理)。

  • 从代理删除消息变得很棘手,因为代理并不知道消费者是否已经使用了该消息。Kafka创新性地解决了这个问题,它将一个简单的基于时间的SLA应用于保留策略。当消息在代理中超过一定时间后,将会被自动删除。
  • 这种创新设计有很大的好处,消费者可以故意倒回到老的偏移量再次消费数据。这违反了队列的常见约定,但被证明是许多消费者的基本特征。

以下摘抄自kafka官方文档:

Kafka Design

目标

1) 高吞吐量来支持高容量的事件流处理

2) 支持从离线系统加载数据

3) 低延迟的消息系统

持久化

1) 依赖文件系统,持久化到本地

2) 数据持久化到log

效率

1) 解决”small IO problem“:

使用”message set“组合消息。

server使用”chunks of messages“写到log。

consumer一次获取大的消息块。

2)解决”byte copying“:

在producer、broker和consumer之间使用统一的binary message format。

使用系统的pagecache。

使用sendfile传输log,避免拷贝。

端到端的批量压缩(End-to-end Batch Compression)

Kafka支持GZIP和Snappy压缩协议。

The Producer

负载均衡

1)producer可以自定义发送到哪个partition的路由规则。默认路由规则:hash(key)%numPartitions,如果key为null则随机选择一个partition。

2)自定义路由:如果key是一个user id,可以把同一个user的消息发送到同一个partition,这时consumer就可以从同一个partition读取同一个user的消息。

异步批量发送

批量发送:配置不多于固定消息数目一起发送并且等待时间小于一个固定延迟的数据。

The Consumer

consumer控制消息的读取。

Push vs Pull

1)producer push data to broker,consumer pull data from broker

2)consumer pull的优点:consumer自己控制消息的读取速度和数量。

3)consumer pull的缺点:如果broker没有数据,则可能要pull多次忙等待,Kafka可以配置consumer long pull一直等到有数据。

Consumer Position

1)大部分消息系统由broker记录哪些消息被消费了,但Kafka不是。

2)Kafka由consumer控制消息的消费,consumer甚至可以回到一个old offset的位置再次消费消息。

Message Delivery Semantics

三种:

At most once—Messages may be lost but are never redelivered.

At least once—Messages are never lost but may be redelivered.

Exactly once—this is what people actually want, each message is delivered once and only once.

Producer:有个”acks“配置可以控制接收的leader的在什么情况下就回应producer消息写入成功。

Consumer:

* 读取消息,写log,处理消息。如果处理消息失败,log已经写入,则无法再次处理失败的消息,对应”At most once“。

* 读取消息,处理消息,写log。如果消息处理成功,写log失败,则消息会被处理两次,对应”At least once“。

* 读取消息,同时处理消息并把result和log同时写入。这样保证result和log同时更新或同时失败,对应”Exactly once“。

Kafka默认保证at-least-once delivery,容许用户实现at-most-once语义,exactly-once的实现取决于目的存储系统,kafka提供了读取offset,实现也没有问题。

复制(Replication)

1)一个partition的复制个数(replication factor)包括这个partition的leader本身。

2)所有对partition的读和写都通过leader。

3)Followers通过pull获取leader上log(message和offset)

4)如果一个follower挂掉、卡住或者同步太慢,leader会把这个follower从”in sync replicas“(ISR)列表中删除。

5)当所有的”in sync replicas“的follower把一个消息写入到自己的log中时,这个消息才被认为是”committed“的。

6)如果针对某个partition的所有复制节点都挂了,Kafka选择最先复活的那个节点作为leader(这个节点不一定在ISR里)。

日志压缩(Log Compaction)

1)针对一个topic的partition,压缩使得Kafka至少知道每个key对应的最后一个值。

2)压缩不会重排序消息。

3)消息的offset是不会变的。

4)消息的offset是顺序的。

Distribution

Consumer Offset Tracking

1)High-level consumer记录每个partition所消费的maximum offset,并定期commit到offset manager(broker)。

2)Simple consumer需要手动管理offset。现在的Simple consumer Java API只支持commit offset到zookeeper。

Consumers and Consumer Groups

1)consumer注册到zookeeper

2)属于同一个group的consumer(group id一样)平均分配partition,每个partition只会被一个consumer消费。

3)当broker或同一个group的其他consumer的状态发生变化的时候,consumer rebalance就会发生。

Zookeeper协调控制

1)管理broker与consumer的动态加入与离开。

2)触发负载均衡,当broker或consumer加入或离开时会触发负载均衡算法,使得一个consumer group内的多个consumer的订阅负载平衡。

3)维护消费关系及每个partition的消费信息。

生产者代码示例:

import java.util.*;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

public class TestProducer {

public static void main(String[] args) {

long events = Long.parseLong(args[0]);

Random rnd = new Random();

Properties props = new Properties();

props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");

props.put("serializer.class", "kafka.serializer.StringEncoder");

props.put("partitioner.class", "example.producer.SimplePartitioner");

props.put("request.required.acks", "1");

ProducerConfig config = new ProducerConfig(props);

Producer<String, String> producer = new Producer<String, String>(config);

for (long nEvents = 0; nEvents < events; nEvents++) {

long runtime = new Date().getTime();

String ip = “192.168.2.” + rnd.nextInt(255);

String msg = runtime + “,www.example.com,” + ip;

KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);

producer.send(data);

}

producer.close();

}

}

Partitioning Code:

import kafka.producer.Partitioner;

import kafka.utils.VerifiableProperties;

public class SimplePartitioner implements Partitioner {

public SimplePartitioner (VerifiableProperties props) {

}

public int partition(Object key, int a_numPartitions) {

int partition = 0;

String stringKey = (String) key;

int offset = stringKey.lastIndexOf(‘.‘);

if (offset > 0) {

partition = Integer.parseInt( stringKey.substring(offset+1)) % a_numPartitions;

}

return partition;

}

}

消费者代码示例:

import kafka.consumer.ConsumerConfig;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Properties;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class ConsumerGroupExample {

private final ConsumerConnector consumer;

private final String topic;

private  ExecutorService executor;

public ConsumerGroupExample(String a_zookeeper, String a_groupId, String a_topic) {

consumer = kafka.consumer.Consumer.createJavaConsumerConnector(

createConsumerConfig(a_zookeeper, a_groupId));

this.topic = a_topic;

}

public void shutdown() {

if (consumer != null) consumer.shutdown();

if (executor != null) executor.shutdown();

try {

if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {

System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");

}

} catch (InterruptedException e) {

System.out.println("Interrupted during shutdown, exiting uncleanly");

}

}

public void run(int a_numThreads) {

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

topicCountMap.put(topic, new Integer(a_numThreads));

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

List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

// now launch all the threads

//

executor = Executors.newFixedThreadPool(a_numThreads);

// now create an object to consume the messages

//

int threadNumber = 0;

for (final KafkaStream stream : streams) {

executor.submit(new ConsumerTest(stream, threadNumber));

threadNumber++;

}

}

private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {

Properties props = new Properties();

props.put("zookeeper.connect", a_zookeeper);

props.put("group.id", a_groupId);

props.put("zookeeper.session.timeout.ms", "400");

props.put("zookeeper.sync.time.ms", "200");

props.put("auto.commit.interval.ms", "1000");

return new ConsumerConfig(props);

}

public static void main(String[] args) {

String zooKeeper = args[0];

String groupId = args[1];

String topic = args[2];

int threads = Integer.parseInt(args[3]);

ConsumerGroupExample example = new ConsumerGroupExample(zooKeeper, groupId, topic);

example.run(threads);

try {

Thread.sleep(10000);

} catch (InterruptedException ie) {

}

example.shutdown();

}

}

import kafka.consumer.ConsumerIterator;

import kafka.consumer.KafkaStream;

public class ConsumerTest implements Runnable {

private KafkaStream m_stream;

private int m_threadNumber;

public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {

m_threadNumber = a_threadNumber;

m_stream = a_stream;

}

public void run() {

ConsumerIterator<byte[], byte[]> it = m_stream.iterator();

while (it.hasNext())

System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));

System.out.println("Shutting down Thread: " + m_threadNumber);

}

}

开发环境搭建:

https://cwiki.apache.org/confluence/display/KAFKA/Developer+Setup

一些example:

https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example

参考:

  • https://kafka.apache.org/documentation.html
  • https://cwiki.apache.org/confluence/display/KAFKA/Index
时间: 2024-11-05 17:30:37

Kafka 基本原理的相关文章

Kafka基本原理

Kafka基本原理 简介 Apache Kafka是分布式发布-订阅消息系统.它最初由LinkedIn公司开发,之后成为Apache项目的一部分.Kafka是一种快速.可扩展的.设计内在就是分布式的,分区的和可复制的提交日志服务. Kafka架构 它的架构包括以下组件: 话题(Topic):是特定类型的消息流.消息是字节的有效负载(Payload),话题是消息的分类名或种子(Feed)名. 生产者(Producer):是能够发布消息到话题的任何对象. 服务代理(Broker):已发布的消息保存在

kafka基本原理学习

下载安装地址:http://kafka.apache.org/downloads.html 原文链接 http://www.jasongj.com/2015/01/02/Kafka深度解析 Kafka主要术语直观解释 BrokerKafka集群包含一个或多个服务器,这种服务器被称为broker Topic每条发布到Kafka集群的消息都有一个类别,这个类别被称为topic.(物理上不同topic的消息分开存储,逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需指定消息的top

kafka 基本原理和概念

Kafka系统的角色 Broker :一台kafka服务器就是一个broker.一个集群由多个broker组成.一个broker可以容纳多个topic topic: 可以理解为一个MQ消息队列的名字 Partition:为了实现扩展性,一个非常大的topic可以分布到多个 broker(即服务器)上,一个topic可以分为多个partition,每个partition是一个有序的队列.partition中的每条消息 都会被分配一个有序的id(offset).kafka只保证按一个partitio

消息服务百科全书——Kafka基本原理介绍

架构 1.1 总体架构 因为Kafka内在就是分布式的,一个Kafka集群通常包括多个代理. 为了均衡负载,将话题分成多个分区,每个代理存储一或多个分区.多个生产者和消费者能够同时生产和获取消息. 一个典型的Kafka请添加链接描述集群中包含若干Producer(可以是web前端产生的Page View,或者是服务器日志,系统CPU.Memory等),若干broker(Kafka支持水平扩展,一般broker数量越多,集群吞吐率越高),若干Consumer Group,以及一个Zookeeper

【Apache Kafka】 Kafka简介及其基本原理

??对于大数据,我们要考虑的问题有很多,首先海量数据如何收集(如Flume),然后对于收集到的数据如何存储(典型的分布式文件系统HDFS.分布式数据库HBase.NoSQL数据库Redis),其次存储的数据不是存起来就没事了,要通过计算从中获取有用的信息,这就涉及到计算模型(典型的离线计算MapReduce.流式实时计算Storm.Spark),或者要从数据中挖掘信息,还需要相应的机器学习算法.在这些之上,还有一些各种各样的查询分析数据的工具(如Hive.Pig等).除此之外,要构建分布式应用还

kafka深入研究之路(2) kafka简介与专业术语解释说明

目录:1.kafka简介 什么是kafka? 设计目标是什么?2.kafka的优缺点3.kafka中专业术语解释说明 官方网站: http://kafka.apache.org/introkafka中文教程 http://orchome.com/kafka/index 1/ kafka 简介Kafka是最初由Linkedin公司开发,是一个分布式.分区的.多副本的.多订阅者,基于zookeeper协调的分布式日志系统(也可以当做MQ系统),常见可以用于web/nginx日志.访问日志,消息服务等

RabbitMQ,Apache的ActiveMQ,阿里RocketMQ,Kafka,ZeroMQ,MetaMQ,Redis也可实现消息队列,RabbitMQ的应用场景以及基本原理介绍,RabbitMQ基础知识详解,RabbitMQ布曙

消息队列及常见消息队列介绍 2017-10-10 09:35操作系统/客户端/人脸识别 一.消息队列(MQ)概述 消息队列(Message Queue),是分布式系统中重要的组件,其通用的使用场景可以简单地描述为: 当不需要立即获得结果,但是并发量又需要进行控制的时候,差不多就是需要使用消息队列的时候. 消息队列主要解决了应用耦合.异步处理.流量削锋等问题. 当前使用较多的消息队列有RabbitMQ.RocketMQ.ActiveMQ.Kafka.ZeroMQ.MetaMq等,而部分数据库如Re

Kafka简介、基本原理、执行流程与使用场景

一.简介 Apache Kafka是分布式发布-订阅消息系统,在 kafka官网上对 kafka 的定义:一个分布式发布-订阅消息传递系统. 它最初由LinkedIn公司开发,Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目.Kafka是一种快速.可扩展的.设计内在就是分布式的,分区的和可复制的提交日志服务. 几种分布式系统消息系统的对比: ##二.Kafka基本架构它的架构包括以下组件: 话题(Topic):是特定类型的消息流.消息是字节的有效负载(Payload),话

腾讯资深架构师给你讲解 kafka的基本原理,带你实战实践

前言 Kafka是最初由Linkedin公司开发,是一个分布式.支持分区的(partition).多副本的(replica),基于zookeeper协调的分布式消息系统,它的最大的特性就是可以实时的处理大量数据以满足各种需求场景:比如基于hadoop的批处理系统.低延迟的实时系统.storm/Spark流式处理引擎,web/nginx日志.访问日志,消息服务等等,用scala语言编写,Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目. kafka入门与实践 第一章 kaf