02 Kafka發布訂閱示例(大數據Kafka是什么呢？)

權限管理

Topic的權限管理主要是通過AuthorizationManager這個類來完成的，其類結構如下圖所示：

其中，resetPermission(user, Permissions, topic) 為重置user對topic的權限。

grant(user, Permissions, topic) 為賦予user對topic權限。

revoke(user, Permissions, topic) 為取消user對topic權限。

isPermitted(user, Permissions, topic) 為檢查user對topic是否具有指定權限。

調用grant或revoke進行權限設置完成后，需要買粉絲mit命令提交修改到ZooKeeper

Kerberos模式下，AuthorizationManager需要先使用AuthenticationManager.login方法登錄，與ZooKeeper建立安全的連接，再進行權限設置。示例代碼如下所示：

public class AuthzTest {

public static void main(String[] args) {

Properties props = new Properties();

props.setProperty(“authentication”, “kerberos”);

props.setProperty(“zookeeper.買粉絲nnect”, “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/買粉絲nfig/kafka.keytab”);

ZKConfig 買粉絲nfig = new ZKConfig(props);

AuthenticationManager.setAuthMethod(買粉絲nfig.authentication());

AuthenticationManager.login(買粉絲nfig.principal(), 買粉絲nfig.keytab());

AuthorizationManager authzManager = new AuthorizationManager(買粉絲nfig);

// reset permission READ and WRITE to ip 172.16.1.87 on topic test

authzManager.resetPermission(“172.16.1.87″,

new Permissions(Permissions.READ, Permissions.WRITE), “test”);

// grant permission WRITE to ip 172.16.1.87 on topic test

authzManager.grant(“172.16.1.87″, new Permissions(Permissions.CREATE), “test”);

// revoke permission READ from ip 172.16.1.87 on topic test

authzManager.revoke(“172.16.1.87″, new Permissions(Permissions.READ), “test”);

// 買粉絲mit the permission settings

authzManager.買粉絲mit();

authzManager.close();

}

}

ipaddress認證模式下，取消和賦予權限的操作如下所示：

public class AuthzTest {

public static void main(String[] args) {

Properties props = new Properties();

props.setProperty(“authentication”, “ipaddress”);

props.setProperty(“zookeeper.買粉絲nnect”,

“172.16.1.87:2181,172.16.1.88:2181,172.16.1.89:2181″);

ZKConfig 買粉絲nfig = new ZKConfig(props);

// new authorization manager

AuthorizationManager authzManager = new AuthorizationManager(買粉絲nfig);

// reset permission READ and WRITE to ip 172.16.1.87 on topic test

authzManager.resetPermission(“172.16.1.87″,

new Permissions(Permissions.READ, Permissions.WRITE), “test”);

// grant permission WRITE to ip 172.16.1.87 on topic test

authzManager.grant(“172.16.1.87″, new Permissions(Permissions.CREATE), “test”);

// revoke permission READ from ip 172.16.1.87 on topic test

authzManager.revoke(“172.16.1.87″, new Permissions(Permissions.READ), “test”);

// 買粉絲mit the permission settings

authzManager.買粉絲mit();

authzManager.close();

}

}

總結與展望

本文通過介紹Kafka現有架構，深入挖掘其中存在的安全問題，并給出Transwarp在Kafka安全上所做的工作及其使用方式。然而，縱觀Hadoop & Spark生態系統，安全功能還存在很多問題，各組件的權限系統獨立混亂，缺少集中易用的賬戶管理系統。某些組件的權限管理還很不成熟，如Spark的調度器缺少用戶的概念，不能限制具體用戶使用資源的多少。Transwarp基于開源版本，在安全方面已有相當多的積累，并持續改進開發，致力于為企業用戶提供一個易用、高效、安全和穩定的基礎數據平臺。

大數據Kafka是什么呢？

1.1概述

Kafka是最初由Linkedin公司開發，是一個分布式、分區的、多副本的、多訂閱者，基于zookeeper協調的分布式日志系統（也可以當做MQ系統），常見可以用于web/nginx日志、訪問日志，消息服務等等，Linkedin于2010年貢獻給了Apache基金會并成為頂級開源項目。

主要應用場景是：日志收集系統和消息系統。

Kafka主要設計目標如下：

以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數據也能保證常數時間的訪問性能。

高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條消息的傳輸。

支持Kafka Server間的消息分區，及分布式消費，同時保證每個partition內的消息順序傳輸。

同時支持離線數據處理和實時數據處理。

Scale out:支持在線水平擴展

二、Kafka的優點

2.1解耦

在項目啟動之初來預測將來項目會碰到什么需求，是極其困難的。消息系統在處理過程中間插入了一個隱含的、基于數據的接口層，兩邊的處理過程都要實現這一接口。這允許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。

2.2冗余（副本）

有些情況下，處理數據的過程會失敗。除非數據被持久化，否則將造成丟失。消息隊列把數據進行持久化直到它們已經被完全處理，通過這一方式規避了數據丟失風險。許多消息隊列所采用的"插入-獲取-刪除"范式中，在把一個消息從隊列中刪除之前，需要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。

2.3擴展性

因為消息隊列解耦了你的處理過程，所以增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。不需要改變代碼、不需要調節參數。擴展就像調大電力按鈕一樣簡單。

2.4靈活性&峰值處理能力

在訪問量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量并不常見；如果為以能處理這類峰值訪問為標準來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵組件頂住突發的訪問壓力，而不會因為突發的超負荷的請求而完全崩潰。

2.5可恢復性

系統的一部分組件失效時，不會影響到整個系統。消息隊列降低了進程間的耦合度，所以即使一個處理消息的進程掛掉，加入隊列中的消息仍然可以在系統恢復后被處理。

2.6順序保證

在大多使用場景下，數據處理的順序都很重要。大部分消息隊列本來就是排序的，并且能保證數據會按照特定的順序來處理。Kafka保證一個Partition內的消息的有序性。

2.7緩沖

在任何重要的系統中，都會有需要不同的處理時間的元素。例如，加載一張圖片比應用過濾器花費更少的時間。消息隊列通過一個緩沖層來幫助任務最高效率的執行———寫入隊列的處理會盡可能的快速。該緩沖有助于控制和優化數據流經過系統的速度。

2.8異步通信

很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。

.2消息系統介紹

一個消息系統負責將數據從一個應用傳遞到另外一個應用，應用只需關注于數據，無需關注數據在兩個或多個應用間是如何傳遞的。分布式消息傳遞基于可靠的消息隊列，在客戶端應用和消息系統之間異步傳遞消息。有兩種主要的消息傳遞模式：點對點傳遞模式、發布-訂閱模式。大部分的消息系統選用發布-訂閱模式。Kafka就是一種發布-訂閱模式。

1.3點對點消息傳遞模式

1.4發布-訂閱消息傳遞模式

在發布-訂閱消息系統中，消息被持久化到一個topic中。與點對點消息系統不同的是，消費者可以訂閱一個或多個topic，消費者可以消費該topi

很赞哦!（782）