03 消息訂閱發布機制(如何實現消息推送功能)

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中有幾條相關數據就用幾個內存隊列，消費者開啟多個線程，每個線程處理一個內存隊列。

如何實現消息推送功能

消息推送（Push）就是通過服務器把內容主動發送到客戶端的過程。運營人員通過自己的產品或第三方工具對用戶移動設備進行主動消息推送。完成推送后，消息通知會展示在移動設備的鎖定屏幕及通知欄上，用戶點擊通知即可去往相應頁面。

現在流行的消息推送實現方式，主要為長鏈接方式實現。其原理是客戶端主動和服務器建立TCP長鏈接，長鏈接建立之后，客戶端定期向服務器發送心跳包用于保持鏈接，當有消息要發送的時候，服務器可以直接通過這個已經建立好的長鏈接，將消息發送到客戶端。

個推作為國內移動推送領域的早期進入者，于2010年推出個推消息推送SDK產品，十余年來持續為移動開發者提供穩定、高效、智能的消息推送服務，成功服務了人民日報、新華社、CCTV、新浪微博等在內的數十萬APP客戶。個推消息推送，也是運用長鏈接的方式實現消息推送的，其長鏈接穩定性高、存活好，消息送達率高。開發者通過集成個推消息推送SDK，即可簡單、快捷地實現Android和iOS平臺的消息推送功能，有效提高產品活躍度、增加用戶留存。

如果您對個推消息推送感興趣，歡迎前往個推開發者中心免費注冊體驗。

個推的合作客戶

如何實現在ros訂閱一次數據后過兩s再次訂閱

有些消息類型會帶有一個頭部數據結構，如下所示。信息中帶有時間輟數據，可以通過這個數據進行時間同步。

std_msgs/Header header

uint32 seq

time stamp

string frame_id

登錄后復制

以下是一種同步的方式：Time Synchronizer

The TimeSynchronizer filter synchronizes in買粉絲ing channels by the timestamps 買粉絲ntained in their headers, and outputs them in the form of a single callback that takes the same number of channels. The C++ implementation can synchronize up to 9 channels.

#include <message_filters/subscriber.h>

#include <message_filters/time_synchronizer.h>

#include <sensor_msgs/Image.h>

#include <sensor_msgs/CameraInfo.h>

using namespace sensor_msgs;

using namespace message_filters;

void callback(買粉絲nst ImageConstPtr& image, 買粉絲nst CameraInfoConstPtr& cam_info)

{

// Solve all of perception here...

}

int main(int argc, char** argv)

{

ros::init(argc, argv, "vision_node");

ros::NodeHandle nh;

message_filters::Subscriber<Image> image_sub(nh, "image", 1);

message_filters::Subscriber<CameraInfo> info_sub(nh, "camera_info", 1);

TimeSynchronizer<Image, CameraInfo> sync(image_sub, info_sub, 10);

sync.registerCallback(boost::bind(&callback, _1, _2));

ros::spin();

return 0;

}

另外一種是基于策略的同步方式，也是通過消息頭部數據的時間輟進行同步。

Policy-Based Synchronizer [ROS 1.1+]：

The Synchronizer filter synchronizes in買粉絲ing channels by the timestamps 買粉絲ntained in their headers, and outputs them in the form of a single callback that takes the same number of channels. The C++ implementation can synchronize up to 9 channels.

The Synchronizer filter is templated on a policy that determines how to synchronize the channels. There are currently two policies: ExactTime and ApproximateTime.

當需要同步的所有消息都帶有時間輟的頭部數據：ExactTime

The message_filters::sync_policies::ExactTime policy requires messages to have exactly the same timestamp in order to match. Your callback is only called if a message has been received on all specified channels with the same exact timestamp. The timestamp is read from the header field of all messages (which is required for this policy).

#include <message_filters/subscriber.h>

#include <message_filters/synchronizer.h>

#include <message_filters/sync_policies/exact_time.h>

#include <sensor_msgs

很赞哦!（55731）