上篇描述的kafka案例是個庫存管理平臺，是一個公共服務平臺，為其它軟體模塊或第三方軟體提供庫存狀態管理服務，當然，平臺管理的目標必須是共享的，即庫存是作為公共資源開放的，這個庫存管理平臺是一個Kafka消費端獨立運行的軟體，kafka的生產方即平臺的服務物件通過kafka生產端producer從四面八方同時、集中將訊息寫入kafka，庫存管理平臺在kafka消費端不間斷監控kafka里新的未讀過的訊息并及時讀取，決議訊息獲取發布者對庫存管理的指令，然后按指令更新庫存狀態，
設計這個庫存管理平臺最主要的目的先是為了保證庫存狀態的時效性、準確性，然后才是庫存更新的效率，由于庫存更新指令的產生是在一個高并發、異類系統、分布式環境里，上篇已經提到多執行緒環境下更新共享資料會產生的問題，不過通過kafka把并發產生的指令轉換成佇列然后按順序單執行緒逐句執行就能解決主要問題了，現在，平臺的資料來源變成kafka消費埠上的一個資料流了，資料的讀取和消費自然也變成了逐條的，kafka提供了某種游標機制來記錄資料讀取的最新位置，防止資料消費程序中的遺漏、重復，記錄當前讀取位置offset的方式就是所謂資料消費模式代表資料消費不同程度的安全/效率比例，安全系數越高，流量越低，具體讀取位置offset可以存放在kafka內部，或者保存在某種資料庫表里，簡單來講，資料消費模式分三種：至多一次at-most-once，至少一次at-least-once，只此一次exactly-once，
從由kafka中讀出指令到成功完成執行指令整個訊息消費程序可能經歷多個步驟，每個步驟都可能有失敗的可能，從而中斷程序影響資料消費結果，保存offset即offset-commit的時間點代表了三種消費模式的特性：
1、至多一次at-most-once：讀出資料立即commit-offset，然后才開始消費資料，無論消費程序中發生例外與否，下次都會從新的位置開始讀取，過去不再，如果一條資料在消費程序中發生事故中斷了程序，那這條資料就沒有發生應有的作用，就等于遺失了，
2、至少一次at-least-once：讀出資料、消費資料、然后才commit-offset，如果消費程序出現問題中斷，那么offset就得不到保存，下次再讀取時還是從原先位置重新開始，所以，一條資料有可能被多次讀取，造成重復消費的效果，
3、只此一次exactly-once：把保存offse和消費程序放到同一個事務transaction里，這種模式需要資料庫事物處理支持，也就是說offset-commit和資料處理都必須在同一種提供事物處理支持的資料庫環境里進行，offset-commit只會在確保消費程序成功完成后才進行，
at-most-once和at-least-once都使用kafka內部commit機制保存offset，at-least-once可以利用kafka的自動commit機制實作offset保存，只要通過kafka配置就可以了，下面是這個配置的示范：
val consumerSettings =
ConsumerSettings(consumerConfig, new StringDeserializer, new StringDeserializer)
.withBootstrapServers(bootstrapServers)
.withGroupId(group)
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset)
.withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
.withProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs.toString)
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG = "true" 代表開啟auto-commit模式，ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG設定了auto-commit之間的毫秒時間間隔，在這個間隔內如果中斷消費程序，那么在這個間隔內讀取所有資料的offset都未能commit，但其中有些資料已經完成消費了，重啟讀取就會從這個間隔開始時的offset從頭讀取，那么之前消費的資料就會再次消費，等于重復消費了，auto-commit間隔設定的越短，重復消費的資料就越少，不過kafka需要更密集的進行commit-offset，運行效率就越低，反之，重復消費的資料量就越大，消費計算精確度越低，但運行效率就會提高，
在alpakka-kafka里用一個普通的Source就可以實作at-least-once消費模式了：
val consumerSettings =
ConsumerSettings(consumerConfig, new StringDeserializer, new StringDeserializer)
.withBootstrapServers(bootstrapServers)
.withGroupId(group)
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset)
.withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
.withProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs.toString)
var subscription = Subscriptions
.topics(topic)
val stkTxns = new DocToStkTxns(trace)
val curStk = new CurStk(trace)
val pcmTxns = new PcmTxns(trace)
val plainSource = Consumer
.plainSource(consumerSettings,subscription)
run這個plainSource形成的akka-stream就實作了一個完整kafka-reader功能：
plainSource
.mapAsync(1) {msg =>
updateStock(msg)
}
.toMat(Sink.seq)(Keep.left)
.run()
offset-commit在這個reader-stream里是不可控的，是kafka按預先設定自動進行的，
plainSource是一個獨立的stream，代表單個reader，為了充分利用平臺的硬體資源，首先考慮的是同時運行多個stream，如下：
(1 to numReaders).toList.map { _ =>
plainSource
.mapAsync(1) {msg =>
updateStock(msg)
}
.toMat(Sink.seq)(Keep.left)
.run()
}
這樣可以同時運行numReaders條stream，不過，現在設計方案又回傳了多執行緒環境，好像又要面臨多并發所產生的一系列問題了，我們來分析分析：首先，前面描述的庫存更新多執行緒競爭問題主要是針對同一門店，同一商品，同時更新庫存狀態引發的，以上設計中每條stream，即每個reader，如果屬于同一個reader-group（group-id相同）的話，應共同分別負責所有partition中的部分partition，是不會共享partition的，那么，寫入每個partition的資料是否交叉重復就很關鍵了，實際上，在上游訊息發布階段決定了訊息應該寫入的具體partition，如下：
def writeToKafka(posTxn: PosTxns)(implicit producerKafka: ProducerKafka) = {
val doc = BizDoc.fromPosTxn(posTxn)
if (producerKafka.producerSettings.isDefined) {
implicit val producer = producerKafka.akkaClassicSystem.get
SendProducer(producerKafka.producerSettings.get)
.send(new ProducerRecord[String, String](producerKafka.publisherSettings.topic, doc.shopId, toJson(doc)))
} else FastFuture.successful(Completed)
}
ProducerRecord[K,V] 的key設定為shopId，具體目標partition由kafka的默認指派演算法根據key的值產生，保證同一key值一定會指派給同一個partition，雖然在門店數量>partition數量的情況下每個partition可以包含多個shopId, 但各partition所包含的shopId不會交叉重復，所以，以上多reader同時運行的設計中，只要屬于同一個reader-group，shopId就不會相同，就不會產生執行緒競爭問題，
那么，在同一個reader的消費程序中是否能使用多執行緒方式呢？上面的例子中使用了mapAsync(parallelism=1)，這個代表了stream里的一個階段，這個階段容許收到上游資料后以parallelism個future來并行處理，同時可以保證流出下游的資料遵守上游流入資料的順序，但是，在同一階段用多執行緒方式計算方式在遇到同門店、同商品庫存更新時同樣會產生多執行緒競爭問題，所以只能取parallelism=1，不過，可以考慮把資料處理程序分割成幾個階段，因為每個階段流入流出的資料是同循序的，所以可以容許多個階段在在各自的執行緒里運算，如：
(1 to numReaders).toList.map { _ =>
plainSource
.mapAsync(1) {msg =>
produceStkTxns(msg)
}
asyn.mapAsync(1) {msg =>
updateCurStock(msg)
}
asyn.mapAsync(1) {msg =>
updatePurchase(msg)
}
.toMat(Sink.seq)(Keep.left)
.run()
}
可以用asyn.mapAsync來分割異執行緒域async-boundary以實作多執行緒運算效果，
下面的完整例子里把例外處理和重啟也考慮了進去：
def start =
(1 to numReaders).toList.map { _ =>
RestartSource
.onFailuresWithBackoff(restartSource) { () => plainSource }
// .viaMat(KillSwitches.single)(Keep.right)
.async.mapAsync(1) { msg =>
for {
_ <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-msg: $msg")(Messages.MachineId("", ""))
}
_ <- stkTxns.docToStkTxns(msg.value())
pmsg <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-docToStkTxns: $msg")(Messages.MachineId("", ""))
msg
}
} yield pmsg
}
.async.mapAsync(1) { msg =>
for {
_ <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-updateStk: msg: $msg")(Messages.MachineId("", ""))
}
curstks <- curStk.updateStk(msg.value())
pmsg<- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-updateStk: curstks-$curstks")(Messages.MachineId("", ""))
msg
}
} yield pmsg
}
.async.mapAsync(1) { msg =>
for {
_ <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-writePcmTxn: msg: $msg")(Messages.MachineId("", ""))
}
pcm <- pcmTxns.writePcmTxn(msg.value())
pmsg <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-updateStk: writePcmTxn-$pcm")(Messages.MachineId("", ""))
msg
}
} yield pmsg
}
.async.mapAsync(1) { msg =>
for {
_ <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-updatePcm: msg: $msg")(Messages.MachineId("", ""))
}
_ <- pcmTxns.updatePcm(msg.value())
_ <- FastFuture.successful {
log.step(s"AtLeastOnceReaderGroup-updateStk: updatePcm-$msg")(Messages.MachineId("", ""))
}
} yield "Completed"
}
.toMat(Sink.seq)(Keep.left)
.run()
}
下面是幾個消費模式的測驗示范代碼：
package com.datatech.txn.server
import akka.actor.ActorSystem
import scala.concurrent._
import MgoRepo._
import com.typesafe.config.ConfigFactory
import scala.jdk.CollectionConverters._
object ConsumeModeTest extends App with JsonConverter {
val config_onenode = ConfigFactory.load("onenode")
implicit val system = ActorSystem("kafka-sys",config_onenode)
var config = ConfigFactory.load()
implicit val ec: ExecutionContext = system.dispatcher //mat.executionContext
var httpport: Int = 53081
var mongohosts = List("localhost:27017")
var elastichost = "http://localhost:9200"
var _http_parallelism: Int = 8
var _seednodes: String = ""
val txnCfg = ConfigFactory.load("txnserver.conf").getConfig("txn.server")
try {
mongohosts = txnCfg.getStringList("mongohosts").asScala.toList
elastichost = txnCfg.getString("elastichost")
_http_parallelism = txnCfg.getInt("http_parallelism")
_seednodes = txnCfg.getString("seednodes")
httpport = txnCfg.getInt("httpport")
}
catch {
case excp: Throwable =>
httpport = 53081
mongohosts = List("localhost:27017")
elastichost = "http://localhost:9200"
_http_parallelism = 8
}
implicit val mgoClient = mongoClient(mongohosts)
val readerConfig = config.getConfig("akka.kafka.consumer")
val readerSettings = ReaderSettings(config.getConfig("kafka-txnserver-consumer"))
implicit val idxer = new TxnIndex(elastichost,true)
readerSettings.consumeMode.toLowerCase() match {
case "atleastonce" =>
val readerGroup = AtLeastOnceReaderGroup(readerConfig,readerSettings, true)
readerGroup.start
case "atmostonce" =>
val readerGroup = AtMostOnceReaderGroup(readerConfig,readerSettings, true)
readerGroup.start
case "exactlyonce" =>
val readerGroup = ExactlyOnceReaderGroup(readerConfig,readerSettings, true)
readerGroup.start
case _ =>
val readerGroup = AtLeastOnceReaderGroup(readerConfig,readerSettings, true)
readerGroup.start
}
scala.io.StdIn.readLine()
idxer.close()
scala.io.StdIn.readLine()
system.terminate()
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/296033.html
標籤：Scala