首先，我們先看下圖，這是一張生產(chǎn)消息到kafka，從kafka消費(fèi)消息的結(jié)構(gòu)圖。
當(dāng)然， 這張圖很簡(jiǎn)單，拿這張圖的目的是從中可以得到的跟本節(jié)文章有關(guān)的消息，有以下兩個(gè)：
1，kafka中的消息不是kafka主動(dòng)去拉去的，而必須有生產(chǎn)者往kafka寫消息。
2，kafka是不會(huì)主動(dòng)往消費(fèi)者發(fā)布消息的，而必須有消費(fèi)者主動(dòng)從kafka拉取消息。
spark Streaming結(jié)合kafka
Spark Streaming現(xiàn)在在企業(yè)中流處理也是用的比較廣泛，但是大家都知道其不是真正的實(shí)時(shí)處理，而是微批處理。
在spark 1.3以前，SPark Streaming與kafka的結(jié)合是基于Receiver方式，顧名思義，我們要啟動(dòng)1+個(gè)Receiver去從kafka里面拉去數(shù)據(jù)，拉去的數(shù)據(jù)會(huì)每隔200ms生成一個(gè)block，然后在job生成的時(shí)候，取出該job處理時(shí)間范圍內(nèi)所有的block，生成blockrdd，然后進(jìn)入Spark core處理。
自Spark1.3以后，增加了direct Stream API，這種呢，主要特點(diǎn)是去掉了Receiver，在生成job，去取rdd的時(shí)候，計(jì)算每個(gè)partition要取數(shù)據(jù)的offset范圍，然后生成一個(gè)kafkardd，該rdd特點(diǎn)是與kafka的分區(qū)是一一對(duì)應(yīng)的。
有上面的特點(diǎn)可以看出，Spark Streaming是要生成rdd，然后進(jìn)行處理的，rdd數(shù)據(jù)集我們可以理解為靜態(tài)的，然每個(gè)批次，都會(huì)生成一個(gè)rdd，該過程就體現(xiàn)了批處理的特性，由于數(shù)據(jù)集時(shí)間段小，數(shù)據(jù)小，所以又稱微批處理，那么就說明不是真正的實(shí)時(shí)處理。
還有一點(diǎn)，spark Streaming與kafka的結(jié)合是不會(huì)發(fā)現(xiàn)kafka動(dòng)態(tài)增加的topic或者partition。
Spark的詳細(xì)教程，請(qǐng)關(guān)注浪尖公眾號(hào)，查看歷史推文。
Spark Streaming與kafka結(jié)合源碼講解，請(qǐng)加入知識(shí)星球，獲取。
flink結(jié)合kafka
大家都知道flink是真正的實(shí)時(shí)處理，他是基于事件觸發(fā)的機(jī)制進(jìn)行處理，而不是像spark Streaming每隔若干時(shí)間段，生成微批數(shù)據(jù)，然后進(jìn)行處理。那么這個(gè)時(shí)候就有了個(gè)疑問，在前面kafka小節(jié)中，我們說到了kafka是不會(huì)主動(dòng)往消費(fèi)者里面吐數(shù)據(jù)的，需要消費(fèi)者主動(dòng)去拉去數(shù)據(jù)來處理。那么flink是如何做到基于事件實(shí)時(shí)處理kafka的數(shù)據(jù)呢？在這里浪尖帶著大家看一下源碼，flink1.5.0為例。
1，flink與kafka結(jié)合的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可以看出，數(shù)據(jù)源的入口就是FlinkKafkaConsumer010，當(dāng)然這里面只是簡(jiǎn)單的構(gòu)建了一個(gè)對(duì)象，并進(jìn)行了一些配置的初始化，真正source的啟動(dòng)是在其run方法中run方法的調(diào)用過程在這里不講解，后面會(huì)出教程講解。
首先看一下類的繼承關(guān)系
public class FlinkKafkaConsumer010 extends FlinkKafkaConsumer09
public class FlinkKafkaConsumer09 extends FlinkKafkaConsumerBase
其中，run方法就在FlinkKafkaConsumerBase里，當(dāng)然其中open方法里面對(duì)kafka相關(guān)內(nèi)容進(jìn)行里初始化。
從輸入到計(jì)算到輸出完整的計(jì)算鏈條的調(diào)用過程，后面浪尖會(huì)出文章介紹。在這里只關(guān)心flink如何從主動(dòng)消費(fèi)數(shù)據(jù)，然后變成事件處理機(jī)制的過程。
由于其FlinkKafkaConsumerBase的run比較長(zhǎng)，我這里只看重要的部分，首先是會(huì)創(chuàng)建Kafka09Fetcher。
this.kafkaFetcher = createFetcher(
sourceContext,
subscribedPartitionsToStartOffsets,
periodicWatermarkAssigner,
punctuatedWatermarkAssigner,
(StreamingRuntimeContext) getRuntimeContext(),
offsetCommitMode,
getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
useMetrics);
接著下面有段神器，flink嚴(yán)重優(yōu)越于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());
它定義了一個(gè)線程池對(duì)象，去動(dòng)態(tài)發(fā)現(xiàn)kafka新增的topic（支持正則形式指定消費(fèi)的topic），或者動(dòng)態(tài)發(fā)現(xiàn)kafka新增的分區(qū)。
接著肯定是啟動(dòng)動(dòng)態(tài)發(fā)現(xiàn)分區(qū)或者topic線程，并且啟動(dòng)kafkaFetcher。
discoveryLoopThread.start();
kafkaFetcher.runFetchLoop();
// --------------------------------------------------------------------
自我介紹一下，小編13年上海交大畢業(yè)，曾經(jīng)在小公司待過，也去過華為、OPPO等大廠，18年進(jìn)入阿里一直到現(xiàn)在。
深知大多數(shù)Java工程師，想要提升技能，往往是自己摸索成長(zhǎng)或者是報(bào)班學(xué)習(xí)，但對(duì)于培訓(xùn)機(jī)構(gòu)動(dòng)則幾千的學(xué)費(fèi)，著實(shí)壓力不小。自己不成體系的自學(xué)效果低效又漫長(zhǎng)，而且極易碰到天花板技術(shù)停滯不前！
因此收集整理了一份《2024年Java開發(fā)全套學(xué)習(xí)資料》，初衷也很簡(jiǎn)單，就是希望能夠幫助到想自學(xué)提升又不知道該從何學(xué)起的朋友，同時(shí)減輕大家的負(fù)擔(dān)。
既有適合小白學(xué)習(xí)的零基礎(chǔ)資料，也有適合3年以上經(jīng)驗(yàn)的小伙伴深入學(xué)習(xí)提升的進(jìn)階課程，基本涵蓋了95%以上Java開發(fā)知識(shí)點(diǎn)，真正體系化！
由于文件比較大，這里只是將部分目錄截圖出來，每個(gè)節(jié)點(diǎn)里面都包含大廠面經(jīng)、學(xué)習(xí)筆記、源碼講義、實(shí)戰(zhàn)項(xiàng)目、講解視頻，并且會(huì)持續(xù)更新！
如果你覺得這些內(nèi)容對(duì)你有幫助，可以掃碼獲取?。。▊渥ava獲取）
最后
《一線大廠Java面試題解析+核心總結(jié)學(xué)習(xí)筆記+最新講解視頻+實(shí)戰(zhàn)項(xiàng)目源碼》，點(diǎn)擊傳送門即可獲??！
碼獲?。。。▊渥ava獲取）**
最后
[外鏈圖片轉(zhuǎn)存中…(img-MBKKfufj-1711968643609)]文章來源：http://www.zghlxwxcb.cn/news/detail-854696.html
[外鏈圖片轉(zhuǎn)存中…(img-tuKAnOdP-1711968643611)]
《一線大廠Java面試題解析+核心總結(jié)學(xué)習(xí)筆記+最新講解視頻+實(shí)戰(zhàn)項(xiàng)目源碼》，點(diǎn)擊傳送門即可獲??！文章來源地址http://www.zghlxwxcb.cn/news/detail-854696.html
到了這里，關(guān)于Flink與Spark Streaming在與kafka結(jié)合的區(qū)別！的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
