第4章分布式消息系統(tǒng)Kafka目

錄4.1Kafka簡(jiǎn)介4.2Kafka在大數(shù)據(jù)生態(tài)系統(tǒng)中的作用4.3Kafka與Flume的區(qū)別與聯(lián)系4.4Kafka相關(guān)概念4.5Kafka的安裝和使用4.6使用Python操作Kafka4.7Kafka與MySQL的組合使用4.7Kafka與MySQL的組合使用4.7Kafka與MySQL的組合使用這里通過(guò)一個(gè)實(shí)例來(lái)演示Kafka與MySQL的組合使用。需要完成的任務(wù)是，把JSON格式數(shù)據(jù)放入Kafka發(fā)送出去，然后，再?gòu)腒afka中獲取到JSON格式數(shù)據(jù)，對(duì)其進(jìn)行解析并寫入到MySQL數(shù)據(jù)庫(kù)。請(qǐng)參照第2章的內(nèi)容完成MySQL數(shù)據(jù)庫(kù)的安裝，并學(xué)習(xí)其使用方法。4.7Kafka與MySQL的組合使用編寫一個(gè)生產(chǎn)者程序producer_json.py：#producer_json.pyfromkafkaimportKafkaProducerimportjson

producer=KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=lambdav:json.dumps(v).encode('utf-8'))#連接kafka

data={"sno":"95001","name":"John","sex":"M","age":23}

producer.send('json_topic',data)#發(fā)送的topic為json_topicproducer.close()4.7Kafka與MySQL的組合使用編寫一個(gè)消費(fèi)者程序consumer_json.py：#consumer_json.pyfromkafkaimportKafkaConsumerimportjsonimportpymysql.cursors

consumer=KafkaConsumer('json_topic',bootstrap_servers=['localhost:9092'],group_id=None,auto_offset_reset='earliest')formsginconsumer:msg1=str(msg.value,encoding="utf-8")#字節(jié)數(shù)組轉(zhuǎn)成字符串dict=json.loads(msg1)#字符串轉(zhuǎn)換成字典#連接數(shù)據(jù)庫(kù)connect=pymysql.Connect(host='localhost',port=3306,user='root',#數(shù)據(jù)庫(kù)用戶名passwd='123456',#密碼db='school',charset='utf8')

4.7Kafka與MySQL的組合使用#獲取游標(biāo)cursor=connect.cursor()

#插入數(shù)據(jù)sql="INSERTINTOstudent(sno,sname,ssex,sage)VALUES('%s','%s','%s',%d)"data=(dict['sno'],dict['name'],dict['sex'],dict['age'])cursor.execute(sql%data)mit()print('成功插入數(shù)據(jù)')

#關(guān)閉數(shù)據(jù)庫(kù)連接connect.close()4.7Kafka與MySQL的組合使用在Windows系統(tǒng)中啟動(dòng)MySQL服務(wù)，然后，打開(kāi)MySQL數(shù)據(jù)庫(kù)的命令行界面，輸入如下SQL語(yǔ)句創(chuàng)建數(shù)據(jù)庫(kù)school：mysql>CREATEDATABASEschool;創(chuàng)建好數(shù)據(jù)庫(kù)school以后，可以使用如下SQL語(yǔ)句打開(kāi)數(shù)據(jù)庫(kù)：mysql>USEschool;使用如下SQL語(yǔ)句創(chuàng)建一個(gè)表student：mysql>CREATETABLEstudent(->snochar(5),->snamechar(10),->ssexchar(2),->sageint);4.7Kafka與MySQL的組合使用使用如下SQL語(yǔ)句查看已經(jīng)創(chuàng)建的表：mysql>SHOWTABLES;在Windows系統(tǒng)中啟動(dòng)Zookeeper服務(wù)和Kafka服務(wù)，然后，先執(zhí)行生產(chǎn)者程序producer_json.py，再執(zhí)行消費(fèi)者程序consumer_json.py，執(zhí)行成功以后，使用如下命令查看MySQL數(shù)據(jù)庫(kù)中新插入的記錄：mysql>SELECT*FROMstudent;可以看到，一條記錄已經(jīng)被成功地插入到了MySQL數(shù)據(jù)庫(kù)。4.8本章小結(jié)Kafka是一個(gè)分布式、分區(qū)的、多副本的、多訂閱者、基于Zookeeper協(xié)調(diào)的分布式日志系統(tǒng)，主要應(yīng)用場(chǎng)景是日志收集系統(tǒng)和消息系統(tǒng)。LinkedIn于2010年把Kafka貢獻(xiàn)給了Apache基金會(huì)并成為頂級(jí)開(kāi)源項(xiàng)目。Kafka能夠以時(shí)間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對(duì)TB級(jí)以上數(shù)據(jù)也能保證常數(shù)時(shí)間的訪問(wèn)性能。