内容简介:近期在做 SOFA 与 SpringCloud 的集成,希望通过一系列的 DEMO 工程去帮助大家更好的使用 SOFA 和 SpringCloud;同时也希望大家一起来参与共建和 star。GitHub传送门:官方网站:https://kafka.apache.org/
近期在做 SOFA 与 SpringCloud 的集成,希望通过一系列的 DEMO 工程去帮助大家更好的使用 SOFA 和 SpringCloud;同时也希望大家一起来参与共建和 star。
GitHub传送门: spring-cloud-sofastack-samples
Kafka 简介
官方网站:https://kafka.apache.org/
功能提供
Apache Kafka™ 是 一个分布式数据流平台,从官方文档的解释来看,其职能大体如下:
- Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system。发布和订阅数据流,与消息队列或企业级消息系统很像。
- Store streams of records in a fault-tolerant durable way。具有很强容灾性的存储数据流
- Process streams of records as they occur。及时的处理数据流。
作为一个后端司机,大多数情况下都是把 Kafka 作为一个分布式消息队列来使用的,分布式消息队列可以提供应用解耦、流量消峰、消息分发等功能,已经是大型互联网服务架构不可缺少的基础设置了。
基本概念
topic 和 partition
Kafka 对数据提供的核心抽象,topic 是发布的数据流的类别或名称。topic 在 Kafka 中,支持多订阅者; 也就是说,topic 可以有零个、一个或多个消费者订阅写到相应 topic 的数据。对应每一个 topic,Kafka 集群会维护像一个如下这样的分区的日志:
每个 Partition 都是一个有序的、不可变的并且不断被附加的记录序列,也就是一个结构化提交日志(commit log)。为了保证唯一标性识 Partition 中的每个数据记录,Partition 中的记录每个都会被分配一个叫做偏移(offset)顺序的ID号。通过一个可配置的保留期,Kafka 集群会保留所有被发布的数据,不管它们是不是已经被消费者处理。例如,如果保留期设置为两天,则在发布记录后的两天内,数据都可以被消费,之后它将被丢弃以释放空间。 Kafka 的性能是不为因为数据量大小而受影响的,因此长时间存储数据并不成问题。
事实上,在每个消费者上保留的唯一元数据是消费者在日志中的偏移位置,这个偏移由消费者控制:通常消费者会在读取记录时线性地提高其偏移值(offset++),但实际上,由于偏移位置由消费者控制,它可以以任何顺序来处理数据记录。 例如,消费者可以重置为较旧的偏移量以重新处理来自过去的数据,或者跳过之前的记录,并从“现在”开始消费。 这种特征的组合意味着 Kafka 消费者非常轻量级,随意的开启和关闭并不会对其他的消费者有大的影响。
日志中的 Partition 有几个目的:
- 保证日志的扩展性,topic 的大小不受单个服务器大小的限制。每个单独的 Partition 大小必须小于托管它的服务器磁盘大小,但 topic 可能有很多 Partition,因此它可以处理任意数量的海量数据。
- 作为并行处理的单位 (知乎-Partition:Kafka可以将主题划分为多个分区(Partition),会根据分区规则选择把消息存储到哪个分区中,只要如果分区规则设置的合理,那么所有的消息将会被均匀的分布到不同的分区中,这样就实现了负载均衡和水平扩展。另外,多个订阅者可以从一个或者多个分区中同时消费数据,以支撑海量数据处理能力)
kafka中的topic为什么要进行分区
原贴: kafka中的topic为什么要进行分区 ,由于不能转载,此处不摘抄原文~
生产者
生产者将数据发布到他们选择的 topic , 生产者负责选择要吧数据分配给 topic 中哪个 Partition。这可以通过循环方式(round-robin)简单地平衡负载,或者可以根据某些语义进行分区(例如基于数据中的某些关键字)来完成。
消费者
消费者们使用消费群组( consumer group )名称来标注自己,几个消费者共享一个 group,每一个发布到 topic 的数据会被传递到每个消费群组( consumer group )中的一个消费者实例。 消费者实例可以在不同的进程中或不同的机器上。
如果所有的消费者实例具有相同的 consumer group,则记录将在所有的消费者实例上有效地负载平衡
如果所有的消费者实例都有不同的 consumer group,那么每个记录将被广播给所有的消费者进程,每个数据都发到了所有的消费者。
上图解释源自 《Kafka 官方文档》 介绍 :
如上图,一个两个服务器节点的Kafka集群, 托管着4个分区(P0-P3),分为两个消费者群. 消费者群A有2个消费者实例,消费者群B有4个. 然而,更常见的是,我们发现主题具有少量的消费者群,每个消费者群代表一个“逻辑订户”。每个组由许多消费者实例组成,保证可扩展性和容错能力。这可以说是“发布-订阅”语义,但用户是一组消费者而不是单个进程。 在Kafka中实现消费的方式,是通过将日志中的分区均分到消费者实例上,以便每个实例在任何时间都是“相应大小的一块”分区的唯一消费者。维护消费者组成员资格的过程,由卡夫卡协议动态处理。 如果新的实例加入组,他们将从组中的其他成员接管一些分区; 如果一个实例消失,其分区将被分发到剩余的实例。 Kafka仅提供单个 分区内 的记录的顺序,而不是主题中的不同分区之间的总顺序。 每个分区 排序 结合按键分区,足以满足大多数应用程序的需求。 但是,如果您需要使用总顺序,则可以通过仅具有一个分区的主题来实现,尽管这仅意味着每个消费者组只有一个消费者进程。
Kafka 作为消息系统
消息系统传统上有两种模式:队列和 发布-订阅 。
- 队列模式中,消费者池可以从服务器读取,每条记录只会被某一个消费者消费
- 允许在多个消费者实例上分配数据处理,但是一旦数据被消费之后,数据就没有了
- 发布订阅模式中,记录将广播给所有消费者
- 允许将数据广播到多个进程,但无法缩放和扩容,因为每个消息都发送给每个订阅用户
本篇只介绍 Kafka 作为消息队列的一些基本概念,更多介绍请参考官方文档。
Kafka 安装
这里来看下如何安装 kafka,下载地址:https://kafka.apache.org/downloads。本篇使用的版本是 kafka_2.12-1.1.1 。
-
获取包文件
> wget http://mirrors.shu.edu.cn/apache/kafka/1.1.1/kafka_2.12-1.1.1.tgz 复制代码
-
解压压缩包
> tar -zxvf kafka_2.12-1.1.1.tgz 复制代码
-
修改配置文件
> cd kafka_2.12-1.1.1/config > vim server.properties 复制代码
我这里主要修改项包括以下几个:
# The id of the broker. This must be set to a unique integer for each broker. broker.id=0 listeners=PLAINTEXT://192.168.0.1:9092 advertised.listeners=PLAINTEXT://192.168.0.1:9092 # zookeeper 地址,可以多个 zookeeper.connect=192.168.0.6:2181 复制代码
Kafka 服务启动需要依赖 Zookeeper ,所以在配置文件中需要指定 Zookeeper 集群地址。Kafka 自己的安装包中解压之后是包括 Zookeeper 的,可以通过以下的方式来启动一个单节点 Zookeeper 实例:
> sh zookeeper-server-start.sh -daemon config/zookeeper.properties 复制代码
这里我是指定了之前部署的一台ZK机器,所以可以直接将ZK地址指到已部署好的地址。Zookeeper 安装可以参考: Linux 下安装 Zookeeper
通过上述操作,下面就可以直接来启动Kafka 服务了:
> sh kafka-server-start.sh config/server.properties 复制代码
SpringBoot 集成 Kafka
构建一个简单的 Kafka Producer 工具依赖
- 依赖引入
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>1.3.5.RELEASE</version><!--$NO-MVN-MAN-VER$--> </dependency> 复制代码
- producer
为了可以把 Kafka 封装已提供给其他模块使用,大家可以将 Kafka 的生产端 工具 类使用 SpringBoot 的自动配置机制进行包装,如下:
@Configuration public class KafkaProducerAutoConfiguration { @Autowired private KafkaTemplate<String, String> kafkaTemplate; @Bean public KafkaSender kafkaSender(){ return new KafkaSender(kafkaTemplate); } } 复制代码
- KafkaSender
public class KafkaSender { private KafkaTemplate<String, String> kafkaTemplate; public KafkaSender(KafkaTemplate<String, String> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } /** * send message */ public void sendMessage(String topic, String message) { kafkaTemplate.send(topic, message); } } 复制代码
- 自动配置
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ io.sofastack.cloud.core.kafka.configuration.KafkaProducerAutoConfiguration 复制代码
工程模块如下: image-20190306151759441.png
案例测试
在测试工程中引入依赖,这个依赖就是上面工程打包来的:
<dependency> <groupId>io.sofastack.cloud</groupId> <artifactId>sofastack-cloud-core-kafka</artifactId> </dependency> 复制代码
- 在 resources 目录下新建 application.properties 配置文件
#============== kafka =================== # 指定kafka 代理地址,可以多个,这里的192.168.0.1是上面Kafka 启动配置文件中对应的 # 注:网上一些帖子中说 Kafka 这里的配置只能是主机名,不支持 ip,没有验证过, # 如果您在验证时出现问题,可以尝试本机绑定下 host spring.kafka.bootstrap-servers= 192.168.0.1:9092 #=============== provider ======================= spring.kafka.producer.retries=0 # 每次批量发送消息的数量 spring.kafka.producer.batch-size=16384 spring.kafka.producer.buffer-memory=33554432 # 指定消息key和消息体的编解码方式 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #=============== consumer ======================= # 指定默认消费者group id spring.kafka.consumer.group-id=test-consumer-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100ms # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.application.name=kafka-test logging.path=./logs 复制代码
- 启动类中模拟发送消息
@SpringBootApplication @PropertySource("classpath:application-kafka.properties") public class ProviderApplication { public static void main(String[] args) { ConfigurableApplicationContext run = SpringApplication.run(ProviderApplication.class, args); // 这里通过容器获取,正常使用情况下,可以直接使用 Autowired 注入 KafkaSender bean = run.getBean(KafkaSender.class); for (int i = 0; i < 3; i++) { //调用消息发送类中的消息发送方法 bean.sendMessage(KafkaContants.TRADE_TOPIC, "send a test message"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } } } 复制代码
- 编写消费者,在 SpringBoot 工程中,消费者实现非常简单
@Component public class KafkaReceiver { // 配置监听的主体,groupId 和配置文件中的保持一致 @KafkaListener(topics = { KafkaContants.TRADE_TOPIC }, groupId = "test-consumer-group") public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); System.out.println(message); } } } 复制代码
启动工程后,可以在控制台看下消费者打印的信息:
这里保持应用正常运行,再通过服务端来手动发送消息,看下是当前消费者能够正确监听到对应的 topic 并消费。
> sh kafka-console-producer.sh --broker-list 192.168.0.1:9092 --topic trading 复制代码
执行上述命令之后,命令行将会等待输入,这里输入先后输入 glmapper 和 sofa :
然后再看下应用程序控制台输入结果如下: image-20190306153452565.png
参考
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。