01 Kafka發布訂閱示例(RocketMQ如何動態擴容和縮容)

《Kafka權威指南》pdf下載在線閱讀，求百度網盤云資源

《Kafka權威指南》（Neha Narkhede）電子書網盤下載免費在線閱讀

鏈接：買粉絲s://pan..買粉絲/s/17wJr53y7UtHbj-H7Y2H-4w 提取碼：1234    

書名：Kafka權威指南

作者：Neha Narkhede

譯者：薛命燈

豆瓣評分：8.5

出版社：人民郵電出版社

出版年份：2017-12-26

頁數：214

內容簡介：

每個應用程序都會產生數據，包括日志消息、度量指標、用戶活動記錄、響應消息等。如何移動數據，幾乎變得與數據本身一樣重要。如果你是架構師、開發者或者產品工程師，同時也是Apache Kafka新手，那么這本實踐指南將會幫助你成為流式平臺上處理實時數據的專家。

本書由出身于LinkedIn的Kafka核心作者和一線技術人員共同執筆，詳細介紹了如何部署Kafka集群、開發可靠的基于事件驅動的微服務，以及基于Kafka平臺構建可伸縮的流式應用程序。通過詳盡示例，你將會了解到Kafka的設計原則、可靠性保證、關鍵API，以及復制協議、控制器和存儲層等架構細節。

● 了解發布和訂閱消息模型以及該模型如何被應用在大數據生態系統中

● 學習使用Kafka生產者和消費者來生成消息和讀取消息

● 了解Kafka保證可靠性數據傳遞的模式和場景需求

● 使用Kafka構建數據管道和應用程序的最佳實踐

● 在生產環境中管理Kafka，包括監控、調優和維護

● 了解Kafka的關鍵度量指標

● 探索Kafka如何成為流式處理利器

作者簡介：

Neha Narkhede， Confluent聯合創始人、CTO，曾在LinkedIn主導基于Kafka和Apache Samza構建流式基礎設施，是Kafka作者之一。

Gwen Shapira， Confluent系統架構師，幫助客戶構建基于Kafka的系統，在可伸縮數據架構方面擁有十余年經驗；曾任Cloudera公司解決方案架構師。另著有《Hadoop應用架構》。

Todd Palino， LinkedIn主任級SRE，負責部署管理大型的Kafka、Zookeeper和Samza集群。

【譯者簡介】

薛命燈，畢業于廈門大學軟件學院，十余年軟件開發和架構經驗，InfoQ高級社區編輯。譯有《硅谷革命》《生產微服務》等書。買粉絲買粉絲CodeDeep。

kafka問題求助

Kafka是由LinkedIn設計的一個高吞吐量、分布式、基于發布訂閱模式的消息系統，使用Scala編寫，它以可水平擴展、可靠性、異步通信和高吞吐率等特性而被廣泛使用。目前越來越多的開源分布式處理系統都支持與Kafka集成，其中Spark Streaming作為后端流引擎配合Kafka作為前端消息系統正成為當前流處理系統的主流架構之一。

然而，當下越來越多的安全漏洞、數據泄露等問題的爆發，安全正成為系統選型不得不考慮的問題，Kafka由于其安全機制的匱乏，也導致其在數據敏感行業的部署存在嚴重的安全隱患。本文將圍繞Kafka，先介紹其整體架構和關鍵概念，再深入分析其架構之中存在的安全問題，最后分享下Transwarp在Kafka安全性上所做的工作及其使用方法。

Kafka架構與安全

首先，我們來了解下有關Kafka的幾個基本概念：

Topic：Kafka把接收的消息按種類劃分，每個種類都稱之為Topic，由唯一的Topic Name標識。

Procer：向Topic發布消息的進程稱為Procer。

Consumer：從Topic訂閱消息的進程稱為Consumer。

Broker：Kafka集群包含一個或多個服務器，這種服務器被稱為Broker。

Kafka的整體架構如下圖所示，典型的Kafka集群包含一組發布消息的Procer，一組管理Topic的Broker，和一組訂閱消息的Consumer。Topic可以有多個分區，每個分區只存儲于一個Broker。Procer可以按照一定的策略將消息劃分給指定的分區，如簡單的輪詢各個分區或者按照特定字段的Hash值指定分區。Broker需要通過ZooKeeper記錄集群的所有Broker、選舉分區的Leader，記錄Consumer的消費消息的偏移量，以及在Consumer Group發生變化時進行relalance. Broker接收和發送消息是被動的：由Procer主動發送消息，Consumer主動拉取消息。

然而，分析Kafka框架，我們會發現以下嚴重的安全問題：

1.網絡中的任何一臺主機，都可以通過啟動Broker進程而加入Kafka集群，能夠接收Procer的消息，能夠篡改消息并發送給Consumer。

2.網絡中的任何一臺主機，都可以啟動惡意的Procer/Consumer連接到Broker，發送非法消息或拉取隱私消息數據。

3.Broker不支持連接到啟用Kerberos認證的ZooKeeper集群，沒有對存放在ZooKeeper上的數據設置權限。任意用戶都能夠直接訪問ZooKeeper集群，對這些數據進行修改或刪除。

4.Kafka中的Topic不支持設置訪問控制列表，任意連接到Kafka集群的Consumer（或Procer）都能對任意Topic讀取（或發送）消息。

隨著Kafka應用場景越來越廣泛，特別是一些數據隱私程度較高的領域（如道路交通的視頻監控），上述安全問題的存在猶如一顆定時炸彈，一旦內網被黑客入侵或者內部出現惡意用戶，所有的隱私數據（如車輛出行記錄）都能夠輕易地被竊取，而無需攻破Broker所在的服務器。

Kafka安全設計

基于上述分析，Transwarp從以下兩個方面增強Kafka的安全性：

身份認證（Authentication）：設計并實現了基于Kerberos和基于IP的兩種身份認證機制。前者為強身份認證，相比于后者具有更好的安全性，后者適用于IP地址可信的網絡環境，相比于前者部署更為簡便。

權限控制（Authorization）：設計并實現了Topic級別的權限模型。Topic的權限分為READ（從Topic拉取數據）、WRITE（向Topic中生產數據）、CREATE（創建Topic）和DELETE（刪除Topic）。

基于Kerberos的身份機制如下圖所示：

Broker啟動時，需要使用配置文件中的身份和密鑰文件向KDC（Kerberos服務器）認證，認證通過則加入Kafka集群，否則報錯退出。

Procer（或Consumer）啟動后需要經過如下步驟與Broker建立安全的Socket連接：

1.Procer向KDC認證身份，通過則得到TGT（票證請求票證），否則報錯退出

2.Procer使用TGT向KDC請求Kafka服務，KDC驗證TGT并向Procer返回SessionKey（會話密鑰）和ServiceTicket（服務票證）

3.Procer使用SessionKey和ServiceTicket與Broker建立連接，Broker使用自身的密鑰解密ServiceTicket，獲得與Procer通信的SessionKey，然后使用SessionKey驗證Procer的身份，通過則建立連接，否則拒絕連接。

ZooKeeper需要啟用Kerberos認證模式，保證Broker或Consumer與其的連接是安全的。

Topic的訪問控制列表（ACL）存儲于ZooKeeper中，存儲節點的路徑為/acl/<topic>/<user>，節點數據為R(ead)、W(rite)、C(reate)、D(elete)權限的集合，如/acl/transaction/jack節點的數據為RW，則表示用戶jack能夠對transaction這個topic進行讀和寫。

另外，kafka為特權用戶，只有kafka用戶能夠賦予/取消權限。因此，ACL相關的ZooKeeper節點權限為kafka具有所有權限，其他用戶不具有任何權限。

構建安全的Kafka服務

首先，我們為Broker啟用Kerberos認證模式，配置文件為/etc/kafka/買粉絲nf/server.properties，安全相關的參數如下所示：

其中，authentication參數表示認證模式，可選配置項為simple, kerberos和ipaddress，默認為simple。當認證模式為kerberos時，需要額外配置賬戶屬性principal和對應的密鑰文件路徑keytab.

認證模式為ipaddress時，Procer和Consumer創建時不需要做任何改變。而認證模式為kerberos時，需要預先創建好相應的principal和keytab，并使用API進行登錄，樣例代碼如下所示：

public class SecureProcer extends Thread {

private final kafka.javaapi.procer.Procer<Integer, String> procer;

private final String topic;

private final Properties props = new Properties();

public SecureProcer(String topic) {

AuthenticationManager.setAuthMethod(“kerberos”);

AuthenticationManager.login(“procer1″, “/etc/procer1.keytab”);

props.put(“serializer.class”, “kafka.serializer.StringEn買粉絲der”);

props.put(“metadata.broker.list”,

“172.16.1.190:9092,172.16.1.192:9092,172.16.1.193:9092″);

// Use random partitioner. Don’t need the key type. Just set it to Integer.

// The message is of type String.

procer = new kafka.javaapi.procer.Procer<Integer, String>(

new ProcerConfig(props));

this.topic = topic;

}

. . .

Topic

很赞哦!（5）