了解了kafka原理之后，對kafka的的應用場景有了一些想法，在下面的一系列討論中把最近一個專案中關于kafka的應用介紹一下，
先介紹一下使用kafka的起因：任何進銷存系統，銷售開單部分都應該算是主體部分了，簡單的說，這是一個包括銷售資料錄入、庫存數扣減兩個動作的程序，銷售專案錄入與庫存扣減之間的聯系可以是單向的，如錄入一個銷售商品導致該商品庫存扣減、也可以是雙向的，即商品銷售需要參考當前庫存數量，依照具體的業務需求，銷售開單程序對當前庫存的依賴決定了更新庫存的方式：當前庫存數從作為參考值到必備條件分別代表事后批次更新或程序中實時更新，當然，從業務方面考慮，錄入銷售專案、立即扣減庫存并作為下一筆錄入的參考數是最理想的了，但在現實中，一個多用戶的環境里，大量執行緒同時對一個商品的庫存數進行更新，這時必須對資料庫表進行鎖定，那么由鎖表造成的問題就無可避免了：輕者造成資料的遺失、重復、偏差，或者拖慢行程，重者鎖死整個系統，這是經典進銷存業務系統普遍面臨的問題，也是促使許多商業軟體開發人員紛紛轉向新的分布式大資料模式去尋求解決方案的主要原因，
如果我們再實際點，能夠容許些微的資料更新延遲，比如說：毫秒級的，那么就可以把銷售專案錄入和庫存扣減兩個動作拆分到兩個相互獨立的程序里，就像DDD模式里的兩個聚合根（aggregate root）, 分別在兩個獨立業務域中實作這兩個動作，獨立的域之間是松散耦合，互不影響的，所以，兩個獨立域的計算模式可以是不同的，例如：銷售專案錄入必須是多人操作，多執行緒高并發的，而庫存扣減卻可以設計成單執行緒或者限定執行緒數量的，這可以是一種典型的讀寫分離CQRS模式：扣減庫存作為一項資料更新動作可以在另外一個模塊，甚至另外一個軟體里，在一個可控的、限定執行緒的環境里獨立運算，和銷售資料錄入部分不發生任何關系，當然，資料錄入完成到庫存更新出結果之間一定會存在延遲，這種延遲不單只是與庫存更新演算法和運算效率有著直接關系，它也和兩個獨立域之間的資料交換速度有莫大關系，kafka，作為一個高效率、高吞吐量、高可用性的訊息佇列系統，足夠擔負起獨立域與域之間的資料交換任務，而且kafka的訊息是持久性的，有重復消費控制機制可以實作資料狀態的重新計算，是事件源event-sourcing模式的一項理想工具選擇，這就是我選擇kafka的原因，
好了，說說這個案例的具體業務需求：這是一個零售業POS軟體云租賃平臺，初步規劃上千獨立門店及上萬級的門店業務操作終端，包括收銀終端、查詢終端、業務管理終端，可想而知，系統應該容許上萬用戶同時進行資訊錄入操作，高并發、高頻率的資料錄入部分（特別是收銀終端商品條碼掃描銷售）已經通過event-sourcing,CQRS等模式實作了，接著需要后端的資料處理部分，特別是當前庫存狀態更新，因為零售店其它業務，如：添訂貨、收發貨、配退貨等都需要及時、準確庫存資料的支持，我們把這個庫存更新功能的實作作為典型的kafka應用案例來介紹，然后再在程序中對akka系列alpakka-kafka的使用進行講解和示范，
首先，后端業務功能與前端資料采集是松散耦合的，特別是后端資料處理應該是所有前端系統共享的業務功能，這個容易理解，我們不可能給每個門店運行一個后端，那樣就需要幾千個后端系統同時運行了，所以，可以把這個庫存更新功能做成一個獨立的庫存管理平臺，為所有業務模塊，或者第三方業務軟體提供庫存狀態支持，
現在我們可以把這個獨立的庫存管理平臺作為一個典型的kafka應用示范，簡單來講，kafka就是一個訊息佇列MQ，從一端寫入訊息（produce）、另一端按寫入順序讀出訊息（consume），中間是一堆復雜的機制去保證集群節點協調、訊息輸出順序、訊息持久化及訊息重復消費等等，在我們的案例里，以庫存管理平臺為核心，一端通過kafka連接所有的平臺用戶，這些分布在各處的應用通過kafka的集群功能同時向kafka的寫入端寫入訊息，這些訊息實際是序列化的庫存更新指令，平臺再通過kafka消費端讀取這些指令，反序列化決議后按順序執行這些更新庫存命令，值得注意的是：平臺此時可以在一個單執行緒里按發出的順序，逐個執行指令，避免了多執行緒產生的不確定因素，
從kafka角度描述：庫存管理平臺用戶即訊息發布者producer，這種訊息發布必須是高并發、高吞吐量的，簡單講就是同時集中大批量的向kafka寫入資料，對平臺各用戶來講，就是一種寫完就了fire-and-go模式，實作起來比較簡單，alpakka-kafka提供了很多型別的sink來實作寫produce功能，下面是一個實際的例子：
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)
}
SendProducer.send就是alpakka-kafka最簡單的一個sink：
object SendProducer {
def apply[K, V](settings: ProducerSettings[K, V])(implicit system: ClassicActorSystemProvider): SendProducer[K, V] =
new SendProducer(settings, system.classicSystem)
}
def send(record: ProducerRecord[K, V]): Future[RecordMetadata] = {
producerFuture.flatMap { producer =>
sendSingle(producer, record, identity)
}
}
send(ProducerRecord)把一條ProducerRecord[K,V]寫入kafka，ProducerRecord型別如下：
/**
* Creates a record with a specified timestamp to be sent to a specified topic and partition
*
* @param topic The topic the record will be appended to
* @param partition The partition to which the record should be sent
* @param timestamp The timestamp of the record, in milliseconds since epoch. If null, the producer will assign
* the timestamp using System.currentTimeMillis().
* @param key The key that will be included in the record
* @param value The record contents
* @param headers the headers that will be included in the record
*/
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
if (topic == null)
throw new IllegalArgumentException("Topic cannot be null.");
if (timestamp != null && timestamp < 0)
throw new IllegalArgumentException(
String.format("Invalid timestamp: %d. Timestamp should always be non-negative or null.", timestamp));
if (partition != null && partition < 0)
throw new IllegalArgumentException(
String.format("Invalid partition: %d. Partition number should always be non-negative or null.", partition));
this.topic = topic;
this.partition = partition;
this.key = key;
this.value =https://www.cnblogs.com/tiger-xc/p/ value;
this.timestamp = timestamp;
this.headers = new RecordHeaders(headers);
}
ProducerRecord的幾個重要屬性：topic,partition,key,value，具體使用如下：
SendProducer(producerKafka.producerSettings.get)
.send(new ProducerRecord[String, String](producerKafka.publisherSettings.topic, doc.shopId, toJson(doc)))
以上示范中：key=shopId, value=https://www.cnblogs.com/tiger-xc/p/toJson(doc)，partition由kafka自動指定，key以每個門店的店號表示，意思是使用kafka默認的演算法按門店號來自動產生訊息對應的partition，具體訊息value是json格式序列化的一個類值，其實對應一條庫存交易記錄，
kafka的另外一端，消費端consumer就是我們這次示范案例的主要部分，庫存管理平臺了，這個平臺是一個以alpakka-kafka-stream為主要運算框架的流計算軟體，我們可以通過這次示范深入了解alpakka-kafka-stream的原理和應用，
庫存管理平臺是一個典型的kafka消費端應用，核心技術是對kafka訊息的讀取方式：既要實作高并發的高吞吐量，又要保證嚴格按照既定順序讀取的嚴密準確性，Kafka是通過對topic的分片partition來實作資料高吞吐量的，一個topic可以對應多個partition，每一個kafka消費端應用對于kafka來講就是一個獨立的consumer，或reader，一個reader可以對應多個partition，為了實作高流量的資料消費，在設計應用系統時可以考慮構建多個kafka消費端，也就是多個reader，這樣，每個reader負責讀取的對應partition數量就減少了，讀取資料任務就可以由多個reader共同負擔了，如此通過增加reader就可以有效提高資料消費的效率，在alpakka-kafka，reader可以用一個stream-source來表示，如下：
val commitableSource = Consumer
.committableSource(consumerSettings, subscription)
run這個source就可以開始從kafka里讀取資料了，如下：
commitableSource
.to(Sink.ignore)
.run()
具體的資料處理（庫存扣減）邏輯在每讀出一條資料后立即執行，run()的主要作用只是推動這個stream的流動，具體例子如下：
committablesource
.mapAsync(1) { msg =>
for {
curstks <- curStk.updateStk(msg.record.value())
} yield "Completed"
FastFuture.successful(msg.committableOffset)
}
.toMat(Committer.sink(committerSettings))(Keep.left)
.run()
這個source代表一個reader，負責讀取這個kafka節點上所有partition的資料，如果同時運行多個source，也就是在一個kafka節點上同一topic有多個reader，即一個reader group，reader間可以分擔流量，下面例子里同時啟動一個包括10個reader的reader-group：
def startReading = {
(1 to 10).toList.map { _ =>
commitableSource
.mapAsync(1) { msg =>
for {
curstks <- curStk.updateStk(msg.record.value())
} yield "Completed"
FastFuture.successful(msg.committableOffset)
}
.toMat(Committer.sink(committerSettings))(Keep.left)
.run()
}
}
如果把這個庫存管理平臺部署在一個多節點的集群里，那么，每一個節點都可以有一組10個庫存更新程序同時運算以達到更高的運算效率，當然，實際情況并不像理論那么簡單，首先，要考慮資料的安全性，也就是kafka訊息消費模式，包括：至少一次at-least-once、至多一次at-most-once、保證一次exactly-once，然后，是否采用分布式運算模式，如何解決多執行緒競爭問題，這些問題都比較復雜，用一篇博客無法完全解釋清楚，就留著在下面的博客中再詳細描述吧，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/296032.html
標籤：Scala
上一篇：R語言學習（瑞士軍刀）