03 發布訂閱消息傳遞模式(Redis發布訂閱和Stream)

at most once: 消費者fetch消息，然后保存offset，然后處理消息；當client保存offset之后，但是在消息處理過程中出現了異常，導致部分消息未能繼續處理。那么此后"未處理"的消息將不能被fetch到，這就是"at most once"。

at least once: 消費者fetch消息，然后處理消息，然后保存offset。如果消息處理成功之后，但是在保存offset階段zookeeper異常導致保存操作未能執行成功，這就導致接下來再次fetch時可能獲得上次已經處理過的消息，這就是"at least once"，原因offset沒有及時的提交給zookeeper，zookeeper恢復正常還是之前offset狀態。

exactly once: kafka中并沒有嚴格的去實現(基于2階段提交，事務)，我們認為這種策略在kafka中是沒有必要的。

通常情況下“at-least-once”是我們首選。(相比at most once而言，重復接收數據總比丟失數據要好)。

kafka高可用由多個broker組成，每個broker是一個節點；

創建一個topic，這個topic會劃分為多個partition，每個partition存在于不同的broker上，每個partition就放一部分數據。

kafka是一個分布式消息隊列，就是說一個topic的數據，是分散放在不同的機器上，每個機器就放一部分數據。

在0.8版本以前，是沒有HA機制的，就是任何一個broker宕機了，那個broker上的partition就廢了，沒法寫也沒法讀，沒有什么高可用性可言。

0.8版本以后，才提供了HA機制，也就是就是replica副本機制。每個partition的數據都會同步到其他的機器上，形成自己的多個replica副本。然后所有replica會選舉一個leader出來，那么生產和消費都跟這個leader打交道，然后其他replica就是follower。

寫的時候，leader會負責把數據同步到所有follower上去，讀的時候就直接讀leader上數據即可。

kafka會均勻的將一個partition的所有replica分布在不同的機器上，從而提高容錯性。

如果某個broker宕機了也沒事，它上面的partition在其他機器上都有副本的，如果這上面有某個partition的leader，那么此時會重新選舉一個新的leader出來，大家繼續讀寫那個新的leader即可。這就有所謂的高可用性了。

寫數據的時候，生產者就寫leader，然后leader將數據落地寫本地磁盤，接著其他follower自己主動從leader來pull數據。一旦所有follower同步好數據了，就會發送ack給leader，leader收到所有follower的ack之后，就會返回寫成功的消息給生產者。

消息丟失會出現在三個環節，分別是生產者、mq中間件、消費者：

RabbitMQ

Kafka

大體和RabbitMQ相同。

Rabbitmq

需要保證順序的消息投遞到同一個queue中，這個queue只能有一個買粉絲nsumer，如果需要提升性能，可以用內存隊列做排隊，然后分發給底層不同的worker來處理。

Kafka

寫入一個partition中的數據一定是有序的。生產者在寫的時候 ，可以指定一個key，比如指定訂單id作為key，這個訂單相關數據一定會被分發到一個partition中去。消費者從partition中取出數據的時候也一定是有序的，把每個數據放入對應的一個內存隊列，一個partition中有幾條相關數據就用幾個內存隊列，消費者開啟多個線程，每個線程處理一個內存隊列。

《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。

如何采用MQTT協議實現android消息推送

MQTT協議實現android消息推送,我想每個Android開發人員對它應該都是比較熟悉的

 MQ 遙測傳輸 (MQTT) 是輕量級基于代理的發布/訂閱的消息傳輸協議,設計思想是開放、簡單、輕量、易于實現

這些特點使它適用于受限環境

例如,但不僅限于此:網絡代價昂貴,帶寬低、不可靠

在嵌入設備中運行,處理器和內存資源有限

該協議的特點有:使用發布/訂閱消息模式,提供一對多的消息發布,解除應用程序耦合

對負載內容屏蔽的消息傳輸

使用 TCP/IP 提供網絡連接

有三種消息發布服務質量:“至多一次”,消息發布完全依賴底層 TCP/IP 網絡

會發生消息丟失或重復

這一級別可用于如下情況,環境傳感器數據,丟失一次讀記錄無所謂,因為不久后還會有第二次發送

“至少一次”,確保消息到達,但消息重復可能會發生

“只有一次”,確保消息到達一次

這一級別可用于如下情況,在計費系統中,消息重復或丟失會導致不正確的結果

小型傳輸,開銷很小(固定長度的頭部是 2 字節),協議交換最小化,以降低網絡流量

使用 Last Will 和 Testament 特性通知有關各方客戶端異常中斷的機制MQTT最簡單的使用包括兩種,一種是發消息,一種是訂閱消息

發消息就是向一個固定IP地址的某個主題發送消息(publish)訂閱消息是向服務器端訂閱某些主題,當其他客戶端向服務器的這個主題廣播消息時,那么所有訂閱這個主題的客戶端就都能收到了MQTT是一項消息傳遞技術,由IBM再2001年發布

總結一下,機制就是使用一個代理服務器message broker,客戶端client連接上這個服務器,然后告訴服務器說,我可以接收哪些類型的消息,同時,client也可以發布自己的消息,這些消息根據協議的內容,可以被其他client獲取

只要手機客戶端,連上服務器,然后就可以接收和發布消息了,不用自己寫socket什么了,低帶寬,低耗電量,代碼量也少,很簡單吧

package 買粉絲

pig

test

mqtt;002003 import 買粉絲

ibm

mqtt

MqttClient;004 import 買粉絲

ibm

mqtt

MqttException;005 import 買粉絲

ibm

mqtt

MqttSimpleCallback;006007 public class SubscribeClient { 008 private final static String CONNECTION_STRING = "tcp://192

168

1

60:1883";009 private final static boolean CLEAN_START = true;010 private final static short KEEP_ALIVE = 30;//低耗網絡,但是又需要及時獲取數據,心跳30s011 private final static String CLIENT_ID = "client1";012 private final static String[] TOPICS = { 013 "Test/TestTopics/Topic1",014 "Test/TestTopics/Topic2",015 "Test/TestTopics/Topic3",016 "toku/client1"017 };018 private final static int[] QOS_VALUES = { 0, 0, 2, 0};019020 //////////////////021 private MqttClient mqttClient = null;02202

很赞哦!（9）