Kafka是由LinkedIn設(shè)計(jì)的一個高吞吐量、分布式、基于發(fā)布訂閱模式的消息系統(tǒng)，使用Scala編寫，它以可水平擴(kuò)展、可靠性、異步通信和高吞吐率等特性而被廣泛使用。目前越來越多的開源分布式處理系統(tǒng)都支持與Kafka集成，其中SparkStreaming作為后端流引擎配合Kafka作為前端消息系統(tǒng)正成為當(dāng)前流處理系統(tǒng)的主流架構(gòu)之一。然而，當(dāng)下越來越多的安全漏洞、數(shù)據(jù)泄露等問題的爆發(fā)，安全正成為系統(tǒng)選型不得不考慮的問題，Kafka由于其安全機(jī)制的匱乏，也導(dǎo)致其在數(shù)據(jù)敏感行業(yè)的部署存在嚴(yán)重的安全隱患。本文將圍繞Kafka，先介紹其整體架構(gòu)和關(guān)鍵概念，再深入分析其架構(gòu)之中存在的安全問題，最后分享下Transwarp在Kafka安全性上所做的工作及其使用方法。Kafka架構(gòu)與安全首先，我們來了解下有關(guān)Kafka的幾個基本概念：Topic：Kafka把接收的消息按種類劃分，每個種類都稱之為Topic，由唯一的TopicName標(biāo)識。Producer：向Topic發(fā)布消息的進(jìn)程稱為Producer。Consumer：從Topic訂閱消息的進(jìn)程稱為Consumer。Broker：Kafka集群包含一個或多個服務(wù)器，這種服務(wù)器被稱為Broker。Kafka的整體架構(gòu)如下圖所示，典型的Kafka集群包含一組發(fā)布消息的Producer，一組管理Topic的Broker，和一組訂閱消息的Consumer。Topic可以有多個分區(qū)，每個分區(qū)只存儲于一個Broker。Producer可以按照一定的策略將消息劃分給指定的分區(qū)，如簡單的輪詢各個分區(qū)或者按照特定字段的Hash值指定分區(qū)。Broker需要通過ZooKeeper記錄集群的所有Broker、選舉分區(qū)的Leader，記錄Consumer的消費(fèi)消息的偏移量，以及在ConsumerGroup發(fā)生變化時進(jìn)行relalance.Broker接收和發(fā)送消息是被動的：由Producer主動發(fā)送消息，Consumer主動拉取消息。然而，分析Kafka框架，我們會發(fā)現(xiàn)以下嚴(yán)重的安全問題：1.網(wǎng)絡(luò)中的任何一臺主機(jī)，都可以通過啟動Broker進(jìn)程而加入Kafka集群，能夠接收Producer的消息，能夠篡改消息并發(fā)送給Consumer。2.網(wǎng)絡(luò)中的任何一臺主機(jī)，都可以啟動惡意的Producer/Consumer連接到Broker，發(fā)送非法消息或拉取隱私消息數(shù)據(jù)。3.Broker不支持連接到啟用Kerberos認(rèn)證的ZooKeeper集群，沒有對存放在ZooKeeper上的數(shù)據(jù)設(shè)置權(quán)限。任意用戶都能夠直接訪問ZooKeeper集群，對這些數(shù)據(jù)進(jìn)行修改或刪除。4.Kafka中的Topic不支持設(shè)置訪問控制列表，任意連接到Kafka集群的Consumer（或Producer）都能對任意Topic讀取（或發(fā)送）消息。隨著Kafka應(yīng)用場景越來越廣泛，特別是一些數(shù)據(jù)隱私程度較高的領(lǐng)域（如道路交通的視頻監(jiān)控），上述安全問題的存在猶如一顆定時炸彈，一旦內(nèi)網(wǎng)被黑客入侵或者內(nèi)部出現(xiàn)惡意用戶，所有的隱私數(shù)據(jù)（如車輛出行記錄）都能夠輕易地被竊取，而無需攻破Broker所在的服務(wù)器。Kafka安全設(shè)計(jì)基于上述分析，Transwarp從以下兩個方面增強(qiáng)Kafka的安全性：身份認(rèn)證（Authentication）：設(shè)計(jì)并實(shí)現(xiàn)了基于Kerberos和基于IP的兩種身份認(rèn)證機(jī)制。前者為強(qiáng)身份認(rèn)證，相比于后者具有更好的安全性，后者適用于IP地址可信的網(wǎng)絡(luò)環(huán)境，相比于前者部署更為簡便。權(quán)限控制（Authorization）：設(shè)計(jì)并實(shí)現(xiàn)了Topic級別的權(quán)限模型。Topic的權(quán)限分為READ（從Topic拉取數(shù)據(jù)）、WRITE（向Topic中生產(chǎn)數(shù)據(jù)）、CREATE（創(chuàng)建Topic）和DELETE（刪除Topic）。基于Kerberos的身份機(jī)制如下圖所示：Broker啟動時，需要使用配置文件中的身份和密鑰文件向KDC（Kerberos服務(wù)器）認(rèn)證，認(rèn)證通過則加入Kafka集群，否則報(bào)錯退出。Producer（或Consumer）啟動后需要經(jīng)過如下步驟與Broker建立安全的Socket連接：1.Producer向KDC認(rèn)證身份，通過則得到TGT（票證請求票證），否則報(bào)錯退出2.Producer使用TGT向KDC請求Kafka服務(wù)，KDC驗(yàn)證TGT并向Producer返回SessionKey（會話密鑰）和ServiceTicket（服務(wù)票證）3.Producer使用SessionKey和ServiceTicket與Broker建立連接，Broker使用自身的密鑰解密ServiceTicket，獲得與Producer通信的SessionKey，然后使用SessionKey驗(yàn)證Producer的身份，通過則建立連接，否則拒絕連接。ZooKeeper需要啟用Kerberos認(rèn)證模式，保證Broker或Consumer與其的連接是安全的。Topic的訪問控制列表（ACL）存儲于ZooKeeper中，存儲節(jié)點(diǎn)的路徑為/acl//，節(jié)點(diǎn)數(shù)據(jù)為R(ead)、W(rite)、C(reate)、D(elete)權(quán)限的集合，如/acl/transaction/jack節(jié)點(diǎn)的數(shù)據(jù)為RW，則表示用戶jack能夠?qū)ransaction這個topic進(jìn)行讀和寫。另外，kafka為特權(quán)用戶，只有kafka用戶能夠賦予/取消權(quán)限。因此，ACL相關(guān)的ZooKeeper節(jié)點(diǎn)權(quán)限為kafka具有所有權(quán)限，其他用戶不具有任何權(quán)限。構(gòu)建安全的Kafka服務(wù)首先，我們?yōu)锽roker啟用Kerberos認(rèn)證模式，配置文件為/etc/kafka/conf/server.properties，安全相關(guān)的參數(shù)如下所示：其中，authentication參數(shù)表示認(rèn)證模式，可選配置項(xiàng)為simple,kerberos和ipaddress，默認(rèn)為simple。當(dāng)認(rèn)證模式為kerberos時，需要額外配置賬戶屬性principal和對應(yīng)的密鑰文件路徑keytab.認(rèn)證模式為ipaddress時，Producer和Consumer創(chuàng)建時不需要做任何改變。而認(rèn)證模式為kerberos時，需要預(yù)先創(chuàng)建好相應(yīng)的principal和keytab，并使用API進(jìn)行登錄，樣例代碼如下所示：publicclassSecureProducerextendsThread{privatefinalkafka.javaapi.producer.Producerproducer;privatefinalStringtopic;privatefinalPropertiesprops=newProperties();publicSecureProducer(Stringtopic){AuthenticationManager.setAuthMethod(“kerberos”);AuthenticationManager.login(“producer1″,“/etc/producer1.keytab”);props.put(“serializer.class”,“kafka.serializer.StringEncoder”);props.put(“metadata.broker.list”,“172.16.1.190:9092,172.16.1.192:9092,172.16.1.193:9092″);//Userandompartitioner.Don’tneedthekeytype.JustsetittoInteger.//ThemessageisoftypeString.producer=newkafka.javaapi.producer.Producer(newProducerConfig(props));this.topic=topic;}...Topic權(quán)限管理Topic的權(quán)限管理主要是通過AuthorizationManager這個類來完成的，其類結(jié)構(gòu)如下圖所示：其中，resetPermission(user,Permissions,topic)為重置user對topic的權(quán)限。grant(user,Permissions,topic)為賦予user對topic權(quán)限。revoke(user,Permissions,topic)為取消user對topic權(quán)限。isPermitted(user,Permissions,topic)為檢查user對topic是否具有指定權(quán)限。調(diào)用grant或revoke進(jìn)行權(quán)限設(shè)置完成后，需要commit命令提交修改到ZooKeeperKerberos模式下，AuthorizationManager需要先使用AuthenticationManager.login方法登錄，與ZooKeeper建立安全的連接，再進(jìn)行權(quán)限設(shè)置。示例代碼如下所示：publicclassAuthzTest{publicstaticvoidmain(String[]args){Propertiesprops=newProperties();props.setProperty(“authentication”,“kerberos”);props.setProperty(“zookeeper.connect”,“172.16.2.116:2181,172.16.2.117:2181,172.16.2.118:2181″);props.setProperty(“principal”,“kafka/host1@TDH”);props.setProperty(“keytab”,“/usr/lib/kafka/config/kafka.keytab”);ZKConfigconfig=newZKConfig(props);AuthenticationManager.setAuthMethod(config.authentication());AuthenticationManager.login(config.principal(),config.keytab());AuthorizationManagerauthzManager=newAuthorizationManager(config);//resetpermissionREADandWRITEtoip172.16.1.87ontopictestauthzManager.resetPermission(“172.16.1.87″,newPermissions(Permissions.READ,Permissions.WRITE),“test”);//grantpermissionWRITEtoip172.16.1.87ontopictestauthzManager.grant(“172.16.1.87″,newPermissions(Permissions.CREATE),“test”);//revokepermissionREADfromip172.16.1.87ontopictestauthzManager.revoke(“172.16.1.87″,newPermissions(Permissions.READ),“test”);//committhepermissionsettingsauthzManager.commit();authzManager.close();}}ipaddress認(rèn)證模式下，取消和賦予權(quán)限的操作如下所示：publicclassAuthzTest{publicstaticvoidmain(String[]args){Propertiesprops=newProperties();props.setProperty(“authentication”,“ipaddress”);props.setProperty(“zookeeper.connect”,“172.16.1.87:2181,172.16.1.88:2181,172.16.1.89:2181″);ZKConfigconfig=newZKConfig(props);//newauthorizationmanagerAuthorizationManagerauthzManager=newAuthorizationManager(config);//resetpermissionREADandWRITEtoip172.16.1.87ontopictestauthzManager.resetPermission(“172.16.1.87″,newPermissions(Permissions.READ,Permissions.WRITE),“test”);//grantpermissionWRITEtoip172.16.1.87ontopictestauthzManager.grant(“172.16.1.87″,newPermissions(Permissions.CREATE),“test”);//revokepermissionREADfromip172.16.1.87ontopictestauthzManager.revoke(“172.16.1.87″,newPermissions(Permissions.READ),“test”);//committhepermissionsettingsauthzManager.commit();authzManager.close();}}總結(jié)與展望本文通過介紹Kafka現(xiàn)有架構(gòu)，深入挖掘其中存在的安全問題，并給出Transwarp在Kafka安全上所做的工作及其使用方式。然而，縱觀Hadoop&Spark生態(tài)系統(tǒng)，安全功能還存在很多問題，各組件的權(quán)限系統(tǒng)獨(dú)立混亂，缺少集中易用的賬戶管理系統(tǒng)。某些組件的權(quán)限管理還很不成熟，如Spark的調(diào)度器缺少用戶的概念，不能限制具體用戶使用資源的多少。Transwarp基于開源版本，在安全方面已有相當(dāng)多的積累，并持續(xù)改進(jìn)開發(fā)，致力于為企業(yè)用戶提供一個易用、高效、安全和穩(wěn)定的基礎(chǔ)數(shù)據(jù)平臺。
還在糾結(jié)Kafka 和 RabbitMQ用哪個?一篇文章告訴你他們的區(qū)別
前言 經(jīng)常有人問我，有 xx 需求，我應(yīng)該用 Kafka 還是 RabbitMQ ？ 這個問題很常見，很多人對二者的選擇把握不好。所以我決定寫篇文章來詳細(xì)說一下：Kafka 和 RabbitMQ 的區(qū)別，適用于什么場景？一、消息的順序 在訂單狀態(tài)變化的消息發(fā)送給所有關(guān)心訂單變化的系統(tǒng)時，我們需要的是消息順序性。我們來...
Java使用kafka發(fā)送消息沒有生效
一般消息發(fā)不出去很大可能都是配置或環(huán)境的問題 1、排查環(huán)境是否有問題，zookeeper節(jié)點(diǎn)是否存活，kafka節(jié)點(diǎn)是否存活，通過命令行的方式能否發(fā)出去消息（使用kafka-console-producer.sh）,如果通過命令行都發(fā)不出去那就是集群的問題了。2、網(wǎng)絡(luò)問題，調(diào)用機(jī)器和集群之間網(wǎng)絡(luò)是否通暢 3、調(diào)用時配置的host、port...
有用過kafka的延遲隊(duì)列功能的人嗎?
時間輪設(shè)計(jì)源于生活，以實(shí)現(xiàn)延遲功能。以Kafka的時間輪為例，其底層為環(huán)形數(shù)組，數(shù)組中每個元素存放一個雙向鏈表TimerTaskList，封裝延時任務(wù)。Kafka中時間輪由20個時間格組成，每個格的時間跨度為1ms，總跨度為20s。每個時間格上顯示的數(shù)字代表任務(wù)的過期時間。時間輪指針隨時間前進(jìn)，當(dāng)指針到達(dá)時間格時，...
Kafka和RabbitMQ有哪些區(qū)別,各自適合什么場景?
消息順序：當(dāng)需要確保消息順序時，RabbitMQ為每個消費(fèi)者創(chuàng)建獨(dú)立隊(duì)列可能導(dǎo)致亂序，而Kafka通過發(fā)布訂閱模式和分區(qū)機(jī)制，避免了這個問題，更適合這類需求。消息匹配：RabbitMQ通過routing_key或自定義消息頭輕松實(shí)現(xiàn)匹配，而Kafka需要消費(fèi)者主動尋找消息，開發(fā)成本更高，RabbitMQ在這點(diǎn)上較為方便。消息超時：電商...
kafka、rabbitmq、rocketmq之間的區(qū)別是什么?
消息超時處理是電商場景中常見的需求。RabbitMQ 通過 TTL 字段實(shí)現(xiàn)延遲隊(duì)列，但存在消息按照先進(jìn)先出規(guī)則進(jìn)入死信隊(duì)列的局限。而 Kafka 雖然在實(shí)現(xiàn)延遲隊(duì)列上更為復(fù)雜，但其持久化特性使得消息在被消費(fèi)后不會立即被刪除，提供了更靈活的處理方式。在事件溯源模式下，事件被當(dāng)成消息發(fā)送到消息隊(duì)列中。Kafka ...
kafka如何保證不重復(fù)消費(fèi)又不丟失數(shù)據(jù)?
面試官：今天我想問下，你覺得Kafka會丟數(shù)據(jù)嗎？候選者：嗯，使用Kafka時，有可能會有以下場景會丟消息。候選者：比如說，我們用Producer發(fā)消息至Broker的時候，就有可能會丟消息。候選者：如果你不想丟消息，那在發(fā)送消息的時候，需要選擇帶有 callBack的api進(jìn)行發(fā)送。候選者：其實(shí)就意味著，如果你發(fā)送...
Kafka事務(wù)解析
在Kafka的消息傳輸過程中，存在三種不同的消息傳輸語義，分別為至少一次（at least once）、最多一次（at most once）和精確一次（exactly once）。當(dāng)生產(chǎn)者向Kafka發(fā)送消息時，只要消息被成功提交到日志文件，由于多副本的存在，這條消息就不會丟失。但若生產(chǎn)者在發(fā)送消息過程中遇到網(wǎng)絡(luò)問題導(dǎo)致通信中斷，...
解決Kafka消息重復(fù)消費(fèi)問題的策略與實(shí)踐
問題一：Kafka消息端的自動提交延遲。Kafka的消費(fèi)端默認(rèn)在消費(fèi)完消息后延遲5秒提交偏移量，這可能導(dǎo)致消費(fèi)者意外終止或系統(tǒng)故障時，偏移量未及時更新，造成重復(fù)消費(fèi)。如果能提高消費(fèi)端處理性能，縮短消費(fèi)時長或調(diào)整超時時間，這將有助于避免此問題。原因二：再均衡機(jī)制引發(fā)的重復(fù)消費(fèi)。Kafka通過再均衡機(jī)制平衡...
kafka如何解決重復(fù)消費(fèi)?
使用 Kafka 的問題與解決 在前文中，我們討論了在使用 Kafka 時可能會遇到的六個主要問題，包括消息順序問題、消息積壓、主鍵沖突、數(shù)據(jù)庫主從延遲、重復(fù)消費(fèi)以及多環(huán)境消費(fèi)問題。下文將對這些問題進(jìn)行詳細(xì)闡述并提供解決策略。1. 消息順序問題 為了保證消息的順序，我們需要確保生產(chǎn)者將相同商戶編號的消息...
Kafka到底有多高可靠?(RNG NB)
若producer收到響應(yīng)消息前l(fā)eader副本掛掉,那么producer會因未收到消息重復(fù)發(fā)送消息,那就可能造成數(shù)據(jù)重復(fù)。怎么解決呢?只要保證業(yè)務(wù)冪等就行。 我們可以通過request.required.acks這個參數(shù)控制消息的發(fā)送頻率。3.消息語義 消息集群整體是一個復(fù)雜的系統(tǒng),所以過程中可能會因?yàn)楦鞣N原因?qū)е孪鬟f出錯,Kafka對于這些可能遇到的...
相關(guān)評說：
雙鑰13327007032： 為什么發(fā)信息信息報(bào)告顯示等待中
龍井市閉式：
______ 通話中, 或者 手機(jī)信號不好 都會這樣.. 發(fā)送 超時. 如果是關(guān)機(jī)的話, 很容易 發(fā)出去的
雙鑰13327007032： QQ 發(fā)信息時說連接超時 -
龍井市閉式：
______ 那有時候是服務(wù)器出錯的問題,并不是真的沒有發(fā)送成功,有時候已經(jīng)發(fā)送成功了也提示那樣的信息,不用太在意的.因?yàn)槲覀儼l(fā)信息都是點(diǎn)對點(diǎn)的信息傳輸,一般并不經(jīng)過服務(wù)器,除非你通過代理的.當(dāng)我們發(fā)送的信息稍微不通暢的時候,服務(wù)器就會提示超時,其實(shí)并不是沒有發(fā)送成功的,不信你發(fā)現(xiàn)這個提示,你問問接受信息的好友有沒有收到你的信息,我想答案應(yīng)該是肯定,也可以說是95%可以送到的^^^^
雙鑰13327007032： 為什么發(fā)送手機(jī)消息總是發(fā)送超時
龍井市閉式：
______ 發(fā)完短信之后最好不要進(jìn)行其他操作,讓他平穩(wěn)的把短信發(fā)出. 最好開啟信息發(fā)送報(bào)告,可以確定發(fā)送情況. 你說的情況一般是網(wǎng)絡(luò)問題,和個人影響不大
雙鑰13327007032： 為什么用手機(jī)給好友發(fā)信息.會說發(fā)送超時.是什么意思 -
龍井市閉式：
______ 信號不好,短時間內(nèi)無法連接網(wǎng)絡(luò).重發(fā)幾次就好了
雙鑰13327007032： jsp中出現(xiàn)的404,500,還有其他的數(shù)字錯誤是什么意思 -
龍井市閉式：
______ 404 是 找不到訪問的頁面或者資源500是你的JSP頁面代碼有錯誤
雙鑰13327007032： 我發(fā)個好友的信息怎么會是發(fā)送超時但是打開聊天記錄沒有了?
龍井市閉式：
______ 掉網(wǎng),更本就沒發(fā)過去.或者是卡了
雙鑰13327007032： 關(guān)于.NET的一個報(bào)錯 -
龍井市閉式：
______ 這個錯誤是告訴你,向服務(wù)器發(fā)送的請求超時了.一般是因?yàn)槟愕姆?wù)器地址端口設(shè)置不正確,或者服務(wù)器關(guān)了之類的原因,總之就是連不上服務(wù)器.
雙鑰13327007032： 手機(jī)發(fā)信息超時問題 -
龍井市閉式：
______ 一、可能是你的卡停機(jī)了、你需要充話費(fèi) 二、可能是你卡的短信功能被關(guān)閉了、你可以到掌上營業(yè)廳開啟或者直接打人工服務(wù)讓他們給你開通、 三、可能是你的短信中心號碼不正確、你按以下操作:進(jìn)短信功能→找到短信設(shè)置→選擇你需要設(shè)置的SIM卡→把中心號碼改成+8613800230500 望采納、謝謝
雙鑰13327007032： kafka消費(fèi)者java版本讀取不到消息怎么辦 -
龍井市閉式：
______ 是不是linux上配置的zookeeper和kafka地址寫成127.0.0.1了.要配置成對外的IP才可以收到消息的.
雙鑰13327007032： linux 多條消息超時控制問題 -
龍井市閉式：
______ 最簡單的辦法,用 select 來檢查 socket 狀態(tài),select支持超時操作,如果超時,你就可以繼續(xù)下一條了.簡單的代碼#include ...void readsocket(int sock) { int nfds, rc; fd_set rfds; struct timeval tv; FD_ZERO(&rfds); FD_SET(sock, &rfds); nfds = ...
