首先，我们先看下图，这是一张生产消息到kafka，从kafka消费消息的结构图。
当然， 这张图很简单，拿这张图的目的是从中可以得到的跟本节文章有关的消息，有以下两个：
1，kafka中的消息不是kafka主动去拉去的，而必须有生产者往kafka写消息。
2，kafka是不会主动往消费者发布消息的，而必须有消费者主动从kafka拉取消息。
spark Streaming结合kafka
Spark Streaming现在在企业中流处理也是用的比较广泛，但是大家都知道其不是真正的实时处理，而是微批处理。
在spark 1.3以前，SPark Streaming与kafka的结合是基于Receiver方式，顾名思义，我们要启动1+个Receiver去从kafka里面拉去数据，拉去的数据会每隔200ms生成一个block，然后在job生成的时候，取出该job处理时间范围内所有的block，生成blockrdd，然后进入Spark core处理。
自Spark1.3以后，增加了direct Stream API，这种呢，主要特点是去掉了Receiver，在生成job，去取rdd的时候，计算每个partition要取数据的offset范围，然后生成一个kafkardd，该rdd特点是与kafka的分区是一一对应的。
有上面的特点可以看出，Spark Streaming是要生成rdd，然后进行处理的，rdd数据集我们可以理解为静态的，然每个批次，都会生成一个rdd，该过程就体现了批处理的特性，由于数据集时间段小，数据小，所以又称微批处理，那么就说明不是真正的实时处理。
还有一点，spark Streaming与kafka的结合是不会发现kafka动态增加的topic或者partition。
Spark的详细教程，请关注浪尖公众号，查看历史推文。
Spark Streaming与kafka结合源码讲解，请加入知识星球，获取。
flink结合kafka
大家都知道flink是真正的实时处理，他是基于事件触发的机制进行处理，而不是像spark Streaming每隔若干时间段，生成微批数据，然后进行处理。那么这个时候就有了个疑问，在前面kafka小节中，我们说到了kafka是不会主动往消费者里面吐数据的，需要消费者主动去拉去数据来处理。那么flink是如何做到基于事件实时处理kafka的数据呢？在这里浪尖带着大家看一下源码，flink1.5.0为例。
1，flink与kafka结合的demo。
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.getConfig.disableSysoutLogging
env.getConfig.setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000))
// create a checkpoint every 5 seconds
env.enableCheckpointing(5000)
// make parameters available in the web interface
env.getConfig.setGlobalJobParameters(params)
// create a Kafka streaming source consumer for Kafka 0.10.x
val kafkaConsumer = new FlinkKafkaConsumer010(
params.getRequired(“input-topic”),
new SimpleStringSchema,
params.getProperties)
val messageStream = env
.addSource(kafkaConsumer)
.map(in => prefix + in)
// create a Kafka producer for Kafka 0.10.x
val kafkaProducer = new FlinkKafkaProducer010(
params.getRequired(“output-topic”),
new SimpleStringSchema,
params.getProperties)
// write data into Kafka
messageStream.addSink(kafkaProducer)
env.execute(“Kafka 0.10 Example”)
从上面的demo可以看出，数据源的入口就是FlinkKafkaConsumer010，当然这里面只是简单的构建了一个对象，并进行了一些配置的初始化，真正source的启动是在其run方法中run方法的调用过程在这里不讲解，后面会出教程讲解。
首先看一下类的继承关系
public class FlinkKafkaConsumer010 extends FlinkKafkaConsumer09
public class FlinkKafkaConsumer09 extends FlinkKafkaConsumerBase
其中，run方法就在FlinkKafkaConsumerBase里，当然其中open方法里面对kafka相关内容进行里初始化。
从输入到计算到输出完整的计算链条的调用过程，后面浪尖会出文章介绍。在这里只关心flink如何从主动消费数据，然后变成事件处理机制的过程。
由于其FlinkKafkaConsumerBase的run比较长，我这里只看重要的部分，首先是会创建Kafka09Fetcher。
this.kafkaFetcher = createFetcher(
sourceContext,
subscribedPartitionsToStartOffsets,
periodicWatermarkAssigner,
punctuatedWatermarkAssigner,
(StreamingRuntimeContext) getRuntimeContext(),
offsetCommitMode,
getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
useMetrics);
接着下面有段神器，flink严重优越于Spark Streaming的，代码如下：
final AtomicReference discoveryLoopErrorRef = new AtomicReference<>();
this.discoveryLoopThread = new Thread(new Runnable() {
@Override
public void run() {
try {
// --------------------- partition discovery loop ---------------------
List discoveredPartitions;
// throughout the loop, we always eagerly check if we are still running before
// performing the next operation, so that we can escape the loop as soon as possible
while (running) {
if (LOG.isDebugEnabled()) {
LOG.debug(“Consumer subtask {} is trying to discover new partitions …”, getRuntimeContext().getIndexOfThisSubtask());
}
try {
discoveredPartitions = partitionDiscoverer.discoverPartitions();
} catch (AbstractPartitionDiscoverer.WakeupException | AbstractPartitionDiscoverer.ClosedException e) {
// the partition discoverer may have been closed or woken up before or during the discovery;
// this would only happen if the consumer was canceled; simply escape the loop
break;
}
// no need to add the discovered partitions if we were closed during the meantime
if (running && !discoveredPartitions.isEmpty()) {
kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
}
// do not waste any time sleeping if we’re not running anymore
if (running && discoveryIntervalMillis != 0) {
try {
Thread.sleep(discoveryIntervalMillis);
} catch (InterruptedException iex) {
// may be interrupted if the consumer was canceled midway; simply escape the loop
break;
}
}
}
} catch (Exception e) {
discoveryLoopErrorRef.set(e);
} finally {
// calling cancel will also let the fetcher loop escape
// (if not running, cancel() was already called)
if (running) {
cancel();
}
}
}
}, "Kafka Partition Discovery for " + getRuntimeContext().getTaskNameWithSubtasks());
它定义了一个线程池对象，去动态发现kafka新增的topic（支持正则形式指定消费的topic），或者动态发现kafka新增的分区。
接着肯定是启动动态发现分区或者topic线程，并且启动kafkaFetcher。
discoveryLoopThread.start();
kafkaFetcher.runFetchLoop();
// --------------------------------------------------------------------
自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。
深知大多数Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！
因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，真正体系化！
由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！
如果你觉得这些内容对你有帮助，可以扫码获取！！（备注Java获取）
最后
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！
码获取！！（备注Java获取）**
最后
[外链图片转存中…(img-MBKKfufj-1711968643609)]文章来源：https://www.toymoban.com/news/detail-854696.html
[外链图片转存中…(img-tuKAnOdP-1711968643611)]
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！文章来源地址https://www.toymoban.com/news/detail-854696.html
到了这里，关于Flink与Spark Streaming在与kafka结合的区别！的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
