2. 程式人生 > >RabbitMQ訊息推送總結

RabbitMQ訊息推送總結

阿新 • • 發佈：2018-11-09

藉助於 RabbitMQ 的 Web STOMP 外掛，實現瀏覽器與服務端的全雙工通訊。從本質上說，RabbitMQ 的 Web STOMP 外掛也是利用 WebSocket 對 STOMP 協議進行了一次橋接，從而實現瀏覽器與服務端的雙向通訊。

安裝 RabbitMQ 服務

mac安裝rabbitmq

brew update
brew install rabbitmq

耐心等待，安裝完成後需要將/usr/local/sbin新增到$PATH，可以將下面這兩行加到~/.bash_profile：

# RabbitMQ Config
export PATH=$PATH:/usr/local/sbin

編輯完後:wq儲存退出，使環境變數立即生效。

source ~/.bash_profile

啟動rabbitmq

rabbitmq-server

啟用相關外掛

rabbitmq-plugins enable rabbitmq_management rabbitmq_web_stomp rabbitmq_stomp

重啟 RabbitMQ 服務

service rabbitmq-server restart

驗證是否安裝成功

通過 Web 瀏覽器來檢視 RabbitMQ 的執行狀態，瀏覽器中輸入 http://{server_ip}:15672，用 guest/guest 預設的使用者和密碼登入後即可檢視 RabbitMQ 的執行狀態。

基於 RabbitMQ 的實時訊息推送

RabbitMQ 有很多第三方外掛，可以在 AMQP 協議基礎上做出許多擴充套件的應用。Web STOMP 外掛就是基於 AMQP 之上的 STOMP 文字協議外掛，利用 WebSocket 能夠輕鬆實現瀏覽器和伺服器之間的實時訊息傳遞。

Paste_Image.png

訊息傳送者

Java 作為 RabbitMQ 客戶端訊息傳送者，Web 瀏覽器作為訊息消費者。

package com.ibm.cdl.itaas.stomp;

import java.io.IOException;
import
 
 java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Program {

    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("rabbitmq", "fanout");
        String routingKey = "rabbitmq_routingkey";
        String message = "{\"name\":\"推送測試資料!\"}";
        channel.basicPublish("rabbitmq", routingKey,null, message.getBytes());
        System.out.println("[x] Sent Message:"+message);
        channel.close();
        connection.close();
    }

}

這裡我們利用 RabbitMQ 官方提供的 Java Client Library 來實現訊息的傳送

Paste_Image.png

exchange 接收到訊息後，會根據訊息的 key 和已經設定的 binding 進行訊息路由，最終投遞到一個或多個佇列裡進行訊息處理。RabbitMQ 預置了一些 exchange，如果客戶端未宣告 exchange 時，RabbitMQ 會根據 exchange 型別使用預設的 exchange。

預置exchange名稱

Paste_Image.png

Exchange 型別

1.Direct exchange
Direct exchange 完全根據 key 進行投遞，只有 key 與繫結時的 routing key 完全一致的訊息才會收到訊息，參考官網提供的圖 4 更直觀地瞭解 Direct exchange。

Paste_Image.png

2.Fanount exchange
Fanount 完全不關心 key，直接採取廣播的方式進行訊息投遞，與該交換機繫結的所有佇列都會收到訊息

Paste_Image.png

3.Topic exchange
Topic exchange 會根據 key 進行模式匹配然後進行投遞，與設定的 routing key 匹配上的佇列才能收到訊息。

4.Headers exchange
Header exchange 使用訊息頭代替 routing key 作為關鍵字進行路由，不過在實際應用過程中這種型別的 exchange 使用較少。

訊息持久化

RabbitMQ 支援訊息的持久化，即將訊息資料持久化到磁碟上，如果訊息伺服器中途斷開，下次開啟會將持久化的訊息重新發送，訊息佇列持久化需要保證 exchange（指定 durable=1）、queue（指定 durable=1）和訊息（delivery_mode=2）3 個部分都是持久化。出於資料安全考慮，一般訊息都會進行持久化。

訊息接收者

html程式碼

<!DOCTYPE html>
<html><head>
  <script src="jquery/jquery-1.9.1.min.js"></script>
  <script src="rabbitmq/sockjs-0.3.js"></script>
  <script src="rabbitmq/stomp.js"></script>

  <style>
      .box {
          width: 440px;
          float: left;
          margin: 0 20px 0 20px;
      }

      .box div, .box input {
          border: 1px solid;
          -moz-border-radius: 4px;
          border-radius: 4px;
          width: 100%;
          padding: 5px;
          margin: 3px 0 10px 0;
      }

      .box div {
          border-color: grey;
          height: 300px;
          overflow: auto;
      }

      div code {
          display: block;
      }

      #first div code {
          -moz-border-radius: 2px;
          border-radius: 2px;
          border: 1px solid #eee;
          margin-bottom: 5px;
      }

      #second div {
          font-size: 0.8em;
      }
  </style>
  <title>RabbitMQ Web STOMP Examples : Echo Server</title>
  <link href="main.css" rel="stylesheet" type="text/css"/>
</head><body lang="en">
    <h1><a href="index.html">RabbitMQ Web STOMP Examples</a> > Echo Server</h1>

    <div id="first" class="box">
      <h2>Received</h2>
      <div></div>
      <form><input autocomplete="off" value="Type here..."></input></form>
    </div>

    <div id="second" class="box">
      <h2>Logs</h2>
      <div></div>
    </div>

    <script>
        var has_had_focus = false;
        var pipe = function(el_name, send) {
            var div  = $(el_name + ' div');
            var inp  = $(el_name + ' input');
            var form = $(el_name + ' form');

            var print = function(m, p) {
                p = (p === undefined) ? '' : JSON.stringify(p);
                div.append($("<code>").text(m + ' ' + p));
                div.scrollTop(div.scrollTop() + 10000);
            };

            if (send) {
                form.submit(function() {
                    send(inp.val());
                    inp.val('');
                    return false;
                });
            }
            return print;
        };
        
        var print_first = pipe('#first', function(data) {
//            client.send('/topic/test', {"content-type":"text/plain"}, data);
        });

// Stomp.js boilerplate
if (location.search == '?ws') {
    var ws = new WebSocket('ws://127.0.0.1:15674/ws');
} else {
    var ws = new SockJS('http://127.0.0.1:15674/stomp');
}

// Init Client
var client = Stomp.over(ws);

// SockJS does not support heart-beat: disable heart-beats
client.heartbeat.outgoing = 0;
client.heartbeat.incoming = 0;
client.debug = pipe('#second');

// Declare on_connect
var on_connect = function(x) {
    client.subscribe("/exchange/rabbitmq/rabbitmq_routingkey", function(d) {
        print_first(d.body);
    });
};

// Declare on_error
var on_error =  function() {
  console.log('error');
};

// Conect to RabbitMQ
client.connect('guest', 'guest', on_connect, on_error, '/');

      $('#first input').focus(function() {
          if (!has_had_focus) {
              has_had_focus = true;
              $(this).val("");
          }
      });
    </script>
</body>
</html>

測試stomp外掛是否好用

Paste_Image.png

重要JavaScript程式碼

// Stomp.js boilerplate
if (location.search == '?ws') {
 var ws = new WebSocket('ws://192.168.1.102:15674/ws');
} else {
 var ws = new SockJS('http://192.168.1.102:15674/stomp');
}
// Init Client
var client = Stomp.over(ws);
// SockJS does not support heart-beat: disable heart-beats
client.heartbeat.outgoing = 0;
client.heartbeat.incoming = 0;
// Declare on_connect
var on_connect = function(x) {
 client.subscribe("/exchange/rabbitmq/rabbitmq_routingkey", function(d) {
 print_first(d.body);
 });
};
// Declare on_error
var on_error = function() {
 console.log('error');
};
// Conect to RabbitMQ
client.connect('guest', 'guest', on_connect, on_error, '/');

RabbitMQ Web STOMP 外掛可以理解為 HTML5 WebSocket 與 STOMP 協議間的橋接，目的也是為了讓瀏覽器能夠使用 RabbitMQ。當 RabbitMQ 訊息伺服器開啟了 STOMP 和 Web STOMP 外掛後，瀏覽器端就可以輕鬆地使用 WebSocket 或者 SockerJS 客戶端實現與 RabbitMQ 伺服器進行通訊。

RabbitMQ Web STOMP 是對 STOMP 協議的橋接，因此其語法也完全遵循 STOMP 協議。STOMP 是基於 frame 的協議，與 HTTP 的 frame 相似。一個 Frame 包含一個 command，一系列可選的 headers 和一個 body。STOMP client 的使用者代理可以充當兩個角色，當然也可能同時充當：作為生產者，通過 SEND frame 傳送訊息到伺服器；作為消費者，傳送 SUBCRIBE frame 到目的地並且通過 MESSAGE frame 從伺服器獲取訊息。

在 Web 頁面中利用 WebSocket 使用 STOMP 協議只需要下載 stomp.js 即可，考慮到老版本的瀏覽器不支援 WebSocket，SockJS 則提供了 WebSocket 的模擬支援。Web 頁面中使用 STOMP 協議詳見下列程式碼清單如下

// 初始化 ws 物件
if (location.search == '?ws') {
 var ws = new WebSocket('ws://192.168.1.102:15674/ws');
} else {
 var ws = new SockJS('http://192.168.1.102:15674/stomp');
}
// 建立連線
var client = Stomp.over(ws);
// SockJS does not support heart-beat: disable heart-beats
client.heartbeat.outgoing = 0;
client.heartbeat.incoming = 0;
// 定義連線成功回撥函式
var on_connect = function(x) {
console.log('connect successfully');
 // 傳送訊息
client.send(destination,head,body);
 // 傳送訊息
client.subcribe(destination,callback);
 // 預設主動 ACK，手動 ACK
client.subcribe(destination,function(message){
 Message.ack();
},{ack:'client'});
// 事務支援
var tx = client.begin();
client.send(destination,head,body);
tx.commit();
};
// 定義連線失敗回撥函式
var on_error = function(error) {
 console.log(error.headers.message);
};
// 連線訊息伺服器
client.connect(login, password, on_connect, on_error, '/');

destination 在 RabbitMQ Web STOM 中進行了相關的定義，根據使用場景的不同，主要有以下 4 種：

  • 1./exchange/<exchangeName>
    對於 SUBCRIBE frame，destination 一般為/exchange/<exchangeName>/[/pattern] 的形式。該 destination 會建立一個唯一的、自動刪除的、名為<exchangeName>的 queue，並根據 pattern 將該 queue 繫結到所給的 exchange，實現對該佇列的訊息訂閱。
    對於 SEND frame，destination 一般為/exchange/<exchangeName>/[/routingKey] 的形式。這種情況下訊息就會被髮送到定義的 exchange 中，並且指定了 routingKey。

  • 2./queue/<queueName>
    對於 SUBCRIBE frame，destination 會定義<queueName>的共享 queue，並且實現對該佇列的訊息訂閱。
    對於 SEND frame，destination 只會在第一次傳送訊息的時候會定義<queueName>的共享 queue。該訊息會被髮送到預設的 exchange 中，routingKey 即為<queueName>。

  • 3./amq/queue/<queueName>
    這種情況下無論是 SUBCRIBE frame 還是 SEND frame 都不會產生 queue。但如果該 queue 不存在，SUBCRIBE frame 會報錯。
    對於 SUBCRIBE frame，destination 會實現對佇列<queueName>的訊息訂閱。
    對於 SEND frame，訊息會通過預設的 exhcange 直接被髮送到佇列<queueName>中。

  • 4./topic/<topicName>
    對於 SUBCRIBE frame，destination 創建出自動刪除的、非持久的 queue 並根據 routingkey 為<topicName>繫結到 amq.topic exchange 上，同時實現對該 queue 的訂閱。
    對於 SEND frame，訊息會被髮送到 amq.topic exchange 中，routingKey 為<topicName>。
    執行java類

Paste_Image.png

瀏覽器訪問效果

Paste_Image.png

WebSocket 作為 HTML5 提供的新一代客戶端-伺服器非同步通訊方法，能夠輕鬆完成前端與後臺的雙向通訊。RabbitMQ 服務提供了一個 STOMP 外掛，能夠實現與 WebSocket 的橋接，這樣既能夠實現訊息的主動推送，同時也能夠實現訊息的非同步處理。在傳統的 Web 開發中存在許多狀態變更實時性的需求，比如資源被佔用後需要廣播它的實時狀態，利用本文提出的解決方案，可以方便將其推送到所有監聽的客戶端。因此在新 J2EE 開發專案中，建議使用本文提出的方案替代原來 ajax 輪詢方法重新整理狀態。

實現伺服器端推送的幾種方式

Web 應用都是基於 HTTP 協議的請求/響應模式，無法像 TCP 協議那樣保持長連線，因此 Web 應用就很難像手機那樣實現實時的訊息推送。就目前來看，Web 應用的訊息推送方式主要有以下幾種

  • Ajax 短輪詢
    Ajax 輪詢主要通過頁面端的 JS 定時非同步重新整理任務來實現資料的載入，但這種方式實時效果較差，而且對服務端的壓力也較大。

  • 長輪詢
    長輪詢主要也是通過 Ajax 機制，但區別於傳統的 Ajax 應用，長輪詢的伺服器端會在沒有資料時阻塞請求直到有新的資料產生或者請求超時才返回，之後客戶端再重新建立連接獲取資料，具體實現方式見圖 1 所示。但長輪詢服務端會長時間地佔用資源，如果訊息頻繁傳送的話會給服務端帶來較大的壓力。

  • WebSocket 雙向通訊
    WebSocket 是 HTML5 中一種新的通訊協議，能夠實現瀏覽器與伺服器之間全雙工通訊。如果瀏覽器和服務端都支援 WebSocket 協議的話，該方式實現的訊息推送無疑是最高效、簡潔的。並且最新版本的 IE、Firefox、Chrome 等瀏覽器都已經支援 WebSocket 協議，Apache Tomcat 7.0.27 以後的版本也開始支援 WebSocket。

相關推薦

RabbitMQ訊息推送總結

藉助於 RabbitMQ 的 Web STOMP 外掛，實現瀏覽器與服務端的全雙工通訊。從本質上說，RabbitMQ 的 Web STOMP 外掛也是利用 WebSocket 對 STOMP 協議進行了一次橋接，從而實現瀏覽器與服務端的雙向通訊。 安裝 RabbitMQ 服務

Linux基礎練習總結-訊息推送部署

1、錯誤排查 [[email protected] ~]# python hello.py  Traceback (most recent call last):   File "hello.py", line 4, in <module>   &

（七）RabbitMQ訊息佇列-通過fanout模式將訊息推送到多個Queue中

前面第六章我們使用的是direct直連模式來進行訊息投遞和分發。本章將介紹如何使用fanout模式將訊息推送到多個佇列。 有時我們會遇到這樣的情況，多個功能模組都希望得到完整的訊息資料。例如一個log的訊息，一個我們希望輸出在螢幕上實時監控，另外一個使用者持久化日誌。這時就可以使用fano

rabbitmq本身支援訊息推送，不僅僅是poll模式

1.  poll方式接受訊息       QueueingConsumer consumer = new QueueingConsumer(channel);       channel.basicConsume(queueName, true, consumer);

未讀訊息（小紅點），前端與 RabbitMQ實時訊息推送實踐，賊簡單~

前幾天粉絲群裡有個小夥伴問過：`web` 頁面的未讀訊息（小紅點）怎麼實現比較簡單，剛好本週手頭有類似的開發任務，索性就整理出來供小夥伴們參考，沒準哪天就能用得上呢。 之前在 [《springboot + rabbitmq 做智慧家居》](https://mp.weixin.qq.com/s?__biz=M

Android 基於Netty的訊息推送方案之物件的傳遞（四）

在上一篇文章中《Android 基於Netty的訊息推送方案之字串的接收和傳送（三）》我們介紹了Netty的字串傳遞，我們知道了Netty的訊息傳遞都是基於流，通過ChannelBuffer傳遞的，那麼自然，Object也需要轉換成ChannelBuffer來傳遞。好在Netty本身已經給我們寫好了

Android 基於Netty的訊息推送方案之字串的接收和傳送（三）

在上一篇文章中《Android 基於Netty的訊息推送方案之概念和工作原理(二)》 ，我們介紹過一些關於Netty的概念和工作原理的內容，今天我們先來介紹一個叫做ChannelBuffer的東東。 ChannelBuffer  Netty中的訊息傳遞，都必須以位元

Android 基於Netty的訊息推送方案之概念和工作原理(二)

上一篇文章中我講述了關於訊息推送的方案以及一個基於Netty實現的一個簡單的Hello World，為了更好的理解Hello World中的程式碼，今天我來講解一下關於Netty中一些概念和工作原理的內容，如果你覺得本篇文章有些枯燥，請先去閱讀《Android 基於Netty的訊息推送方案之Hell

IOS APNS訊息推送框架介紹(pushy)以及詳細使用方法

最近公司需要做IOS訊息推送的功能，我負責後臺推送，IOS端資料處理以及回撥我不負責，本篇文章主要介紹基於java的apns訊息推送，使用框架為pushy。 宣告：我先前也沒有接觸過這個IOS推送，自己研究了兩天，通過百度，對比各個框架的優缺點，最後選擇了這個框架，有說的不對的地方，還

android 實現mqtt訊息推送，以及不停斷線重連的問題解決

前段時間專案用到mqtt的訊息推送，整理一下程式碼，程式碼的原型是網上找的，具體哪個地址已經忘記了。 程式碼的實現是新建了一個MyMqttService，全部功能都在裡面實現，包括連伺服器，斷線重連，訂閱訊息，處理訊息，釋出訊息等基本操作。 首先新增依賴： dependencies { &

基於Netty實現的Android 訊息推送(即時通訊)的解決方案

根據Netty框架實現訊息推送(即時聊天)功能. Netty框架，TCP長連線，心跳，阻塞訊息佇列，執行緒池處理訊息傳送, 基於Google ProtoBuf自定義的訊息協議， TCP粘包/拆包.... 客戶端通過TCP連線到伺服器，並建立TCP長連線；當伺服器端收到新訊息後通過TCP連線推送給

百度雲訊息推送

import com.baidu.yun.core.log.YunLogEvent; import com.baidu.yun.core.log.YunLogHandler; import com.baidu.yun.push.auth.PushKeyPair; import com.baidu.yun

微信公總號訊息推送

                   * 任務完成給製作者推送訊息          &nbs

Android UDP - 簡單訊息推送功能

近期接觸了幾個小的Android開發專案，根據需求要利用到網路傳輸中的UDP協議方式傳輸，遇到不少坑，在此分享一下在Android應用中UDP的一些簡單技術功能實現，希望能幫到用得到的同僚。 需(wo)求(yao)分(gan)析(ma)： 從PC上輸入一串亂七八糟，然後能在我手機（某為pow

[轉]spring boot下WebSocket訊息推送

原文連結：spring boot下WebSocket訊息推送， 修正部分錯別字 WebSocket協議 WebSocket是一種在單個TCP連線上進行全雙工通訊的協議。WebSocket通訊協議於2011年被IETF定為標準RFC 6455，並由RFC7936補充規範。WebSocket A

Django——微信訊息推送

前言 微信公眾號的分類 微信訊息推送 公眾號 已認證公眾號 服務號 已認證服務號 企業號 基於：微信認證服務號 主動推送微信訊息。 前提：關注服務號 環境：沙箱環境 沙箱環境地址： https:/

iOS訊息推送機制

  Little_Dragon  關注 2015.10.05 22:33*  字數 1253  閱讀 17337 評論 36 喜歡 151 推送

使用websocket進行訊息推送服務

Websocket主要做訊息推送，簡單，輕巧，比comet好用 入門瞭解：https://www.cnblogs.com/xdp-gacl/p/5193279.html   /** * A Web Socket session represents a conversation bet

nodejs訊息推送之socket.io 與 ws對比

node.js的websocket庫目前比較熱門的是ws和socket.io。我們對比一下這兩個庫。 一、筆者寫這篇文章時，ws的周下載量是4百多萬，最近一次更新是11天前，總共98個版本。 socket.io周下載量接近2百萬，最近一次更新是三個月前。總共110個版本 從n

nodejs訊息推送之ws

nodejs伺服器端，如何訊息的型別為test就返回訊息，反之則群發： const WebSocket = require('ws'); const wss = new WebSocket.Server({port:3030}) const connection = {} wss.on('co