2. 程式人生 > >RabbitMQ札記-訊息分發策略

RabbitMQ札記-訊息分發策略

阿新 • • 發佈：2018-12-12

今天來學習RabbitMQ的訊息分發策略。

在RabbitMQ札記-RabbitMQ入門一文中，我們曾學習過RabbitMQ的概念模型，其中就介紹過訊息分發策略。

Exchange表示交換器，用來接收生產者傳送的訊息並將這些訊息路由給佇列。從圖中可以看出，Procuder釋出的Message進入了Exchange。Exchange通過Routing key與Queue繫結在一起。通過Routing key， RabbitMQ可以得知應該把這個Message放到哪個Queue裡。Exchange分發訊息時根據型別的不同分發策略有區別，目前共五種型別：fanout、direct、topic、headers、x-delayed-message。

fanout。每個發到 fanout 型別交換器的訊息都會分到所有繫結的佇列上去。 direct。如果Routing key匹配, 那麼Message就會被傳遞到相應的Queue中。比如key可以傳遞到Routing key為key的Queue。 topic。對Routing key進行模式匹配，比如key*可以傳遞到Routing key為key1、key2、key3的Queue。

為了理解訊息分發策略，我們將建立一個簡單的日誌系統。它將包含兩種程式，一個Producer和兩個Consumer。Producer將傳送日誌訊息，Consumer將接收並列印訊息。

fanoutx

在前面的文章中，每個Message都是deliver給一個Consumer。今天我們學習如何將一個Message傳遞給多個Consumer，這種模式被稱為“釋出/訂閱

”模式。

分發策略中的fanout就是廣播模式，每個發到 fanout 型別Exchange的訊息都會發到所有繫結的佇列上去。讓我們建立一個這種型別的Exchange，並將其命名為logs：

channel.exchangeDeclare（“logs”，“fanout”）;

現在，我們可以釋出Message到名為logs的Exchange了：

channel.basicPublish("logs","",null,message.getBytes());
臨時佇列

截至現在，我們用的queue都是有名字的，如testQueue。但是對於我們將要構建的日誌系統，並不需要有名字的queue。當我們不給queueDeclare()提供引數時，將用一個自動生成的名稱建立一個非持久的，獨佔的自動刪除佇列：

String queueName = channel.queueDeclare().getQueue();

此時queueName是一個隨機的佇列名稱，它可能看起來像amq.gen-JzTY20BRgKO-HjmUJj0wLg。

繫結

我們已經學習瞭如何建立fanout 型別的Exchange和臨時佇列。現在我們需要將兩者繫結起來。

channel.queueBind(queueName,"logs","");
最終版本

Producer.java

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

public class Producer {

	private final static String EXCHANGE_NAME = "logs";

	public static void main(String[] args) throws IOException, TimeoutException {
		// 建立一個到伺服器的連線
		ConnectionFactory factory = new ConnectionFactory();
		factory.setUsername("yst");
		factory.setPassword("yst");
		factory.setHost("192.168.17.64");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();

		//釋出訊息到我們的exchange，而不是佇列
		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

		// 傳送的訊息
		String message = "Hello World.";
		// 發訊息
		channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
		System.out.println("Sent：" + message);

		// 關閉渠道和連線;
		channel.close();
		conn.close();
	}
}

Consumer.java

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

	private final static String EXCHANGE_NAME = "logs";

	public static void main(String[] args) throws IOException, TimeoutException {
		// 建立一個到伺服器的連線
		ConnectionFactory factory = new ConnectionFactory();
		factory.setUsername("yst");
		factory.setPassword("yst");
		factory.setHost("192.168.17.64");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();

		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
		//獲取channel繫結的佇列的名字
		String queueName = channel.queueDeclare().getQueue();
		//將queue與exchange繫結
		channel.queueBind(queueName, EXCHANGE_NAME, "");

		System.out.println("Waiting for messages.");

		// 建立佇列消費者
		final DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				System.out.println("Received：" + message);
			}
		};
		channel.basicConsume(queueName, true, consumer);
	}
}

啟動Consumer的兩個例項C1和C2，再執行Producer傳送訊息，檢視結果 C1：

Waiting for messages.
Received：Hello World.

C2：

Waiting for messages.
Received：Hello World.

從結果中可以看出，兩個Consumer都收到了訊息。從管理平臺中可以看到，logs交換器確實綁定了兩個queue。 MarkdownPhotos/master/CSDNBlogs/RabbitMQ/pubsubDemoResult.png

direct

在上文中，我們建立了一個簡單的日誌系統，它可以將訊息廣播給多個Consumer。在本文中，我們將通過direct模式為其新增一個功能。Consumer將只能訂閱一部分訊息。

direct意為如果Routing key匹配, 那麼Message就會被傳遞到相應的Queue中。比如Routing key為key的Exchange可以分發訊息到Routing key為key的Queue。 MarkdownPhotos/master/CSDNBlogs/RabbitMQ/directExchangeModel.png 從圖中我們可以看到兩個佇列繫結的直接交換。第一個佇列用繫結鍵key1繫結，第二個佇列有兩個繫結，一個繫結鍵為key2，另一個為key3。通過路由鍵key1釋出到Exchange的訊息將被路由到佇列Queue1。通過路由鍵key2或key3釋出到Exchange的訊息將被路由到佇列Queue2。其他訊息將被丟棄。

繫結

繫結是exchange和queue之間的關係。在前面的例子中，我們已經建立了繫結，queueName為佇列的名字，“logs”為Exchange的名字。

channel.queueBind(queueName,"logs","");

繫結可以採用額外的routingKey引數。

channel.queueBind（queueName，"logs"，routingKey）;

對於fanout的exchange來說，這個引數是被忽略的。

多個繫結

繫結多個佇列是完全合法的。 MarkdownPhotos/master/CSDNBlogs/RabbitMQ/multiBindExchangeModel.png 在這種情況下，直接交換就像廣播一樣，將訊息廣播到所有的匹配佇列。

最終版本

Producer.java

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

public class Producer {

	private final static String EXCHANGE_NAME = "direct_logs";
	private final static String ROUTING_KEY = "key1";

	public static void main(String[] args) throws IOException, TimeoutException {
		// 建立一個到伺服器的連線
		ConnectionFactory factory = new ConnectionFactory();
		factory.setUsername("yst");
		factory.setPassword("yst");
		factory.setHost("192.168.17.64");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();

		//釋出訊息到我們的exchange，而不是佇列
		channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

		// 傳送的訊息
		String message = "Hello World.";
		// 發訊息
		channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());
		System.out.println("Sent：" + ROUTING_KEY + ":" + message);

		// 關閉渠道和連線;
		channel.close();
		conn.close();
	}
}

Consumer.java

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

	private final static String EXCHANGE_NAME = "direct_logs";
	private final static String ROUTING_KEY = "key1";

	public static void main(String[] args) throws IOException, TimeoutException {
		// 建立一個到伺服器的連線
		ConnectionFactory factory = new ConnectionFactory();
		factory.setUsername("yst");
		factory.setPassword("yst");
		factory.setHost("192.168.17.64");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();

		channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
		//獲取channel繫結的佇列的名字
		String queueName = channel.queueDeclare().getQueue();
		//將queue與exchange繫結
		channel.queueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);

		System.out.println("Waiting for messages.");

		// 建立佇列消費者
		final DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				System.out.println("Received：" + envelope.getRoutingKey() + "':'" + message);
			}
		};
		channel.basicConsume(queueName, true, consumer);
	}
}

Consumer2.java 將Consumer.java複製一份，命名為Consumer2.java。將其中的ROUTING_KEY改為“key2”。

執行兩個Consumer例項，稱其為C1和C2，執行一個Consumer2例項，稱其為C3。執行Producer傳送訊息。觀察執行結果 C1：

Waiting for messages.
Received：key1':'Hello World.

C2：

Waiting for messages.
Received：key1':'Hello World.

C3：

Waiting for messages.

結果驗證了上述直接交換和多個繫結中的內容。

topic

在上文中，我們改進了我們的日誌系統。我們使用了直接交換，使Consumer支援選擇性地接收日誌。儘管如此，它仍然有侷限性。為了提高日誌系統中的靈活性，我們需要了解更復雜的話題交換。

topic可以對Routing key進行模式匹配，比如key*可以傳遞到Routing key為key1、key2、key3的Queue。Routing key的值是有限制的，它必須是由“.”分隔的單詞列表，比如“ stock.usd.nyse ”，“ nyse.vmw ”，“ quick.orange.rabbit ”。Routing key最長不能超過255 bytes。

Routing key中可能有兩個特殊字元：

  • *可以代替一個字。
  • #可以代替零個或多個單詞。

請看下面一個例子 MarkdownPhotos/master/CSDNBlogs/RabbitMQ/topicExchangeModel.png 在這裡我們建立了兩個繫結： Queue1 的binding key 是".orange."； Queue2的binding key 是 “..rabbit” 和 “lazy.#”。

  • Queue1 對所有orange顏色的動物感興趣
  • Queue2 對所有的rabbits和所有的lazy的感興趣

比如Routing key是 “quick.orange.rabbit"將會發送到Queue1和Queue2中。訊息"lazy.orange.elephant” 也會發送到Queue1和Queue2。但是"quick.orange.fox" 會發送到Queue1；"lazy.brown.fox"會發送到Queue2。“lazy.pink.rabbit” 也會發送到Queue2，但是儘管兩個routing_key都匹配，它也只是傳送一次。“quick.brown.fox” 會被丟棄。

Topic exchange功能強大，可以轉化為其他的exchange。 如果binding_key 是 “#” ，它會接收所有的Message，不管Routing key是什麼，就像是fanout exchange。 如果 "*“和”#"沒有被使用，那麼topic exchange就變成了direct exchange。

參考資料：

相關推薦

RabbitMQ札記-訊息分發策略

今天來學習RabbitMQ的訊息分發策略。 在RabbitMQ札記-RabbitMQ入門一文中，我們曾學習過RabbitMQ的概念模型，其中就介紹過訊息分發策略。 Exchange表示交換器，用來接收生產者傳送的訊息並將這些訊息路由給佇列。從圖中可以看出，Pr

RabbitMQ(三):訊息持久化策略

一、前言 　　在正常的伺服器執行過程中，時常會面臨伺服器宕機重啟的情況，那麼我們的訊息此時會如何呢？很不幸的事情就是，我們的訊息可能會消失，這肯定不是我們希望見到的結果。所以我們希望AMQP伺服器崩潰了也可以將訊息恢復，這稱之為訊息持久化。RabbitMQ自然存在這種策略可以幫助我們完成這件事情。 二、持

kafka訊息分發策略分析

當我們使用kafka向指定Topic傳送訊息時，如果該Topic具有多個partition，無論消費者有多少，最終都會保證一個partition內的訊息只會被一個Consumer group中的一個Consumer消費，也就是說同一Consumer group中的多個Consumer自動會起到負載均衡的效果。

RabbitMQ訊息分發模式----"Publish/Subscribe"釋出/訂閱模式

介紹 我們都是基於一個佇列傳送和接受訊息。  前面講的幾種，不管是生產者端還是消費者端都必須知道一個指定的QueueName才能傳送、獲取訊息。  而RabbitMQ訊息模型的核心思想是生產者不會將訊息直接傳送給佇列。現在介紹一下完整的訊息傳遞模式： 如果同一個訊息，

RabbitMQ訊息分發模式----"Hello World"模式

"Hello World"模式：即最簡單的分發模式：一個訊息生產者，一個訊息消費者。 生產者端程式碼： <span style="font-size:14px;">public class MsgSender { private

rabbitmq(4)訊息佇列(公平分發)

1、目標 實現rabbitmq將訊息優先發給閒置的消費者。 2、程式碼 2.1、生產者 private final static String QUEUE_NAME = "hello";

輕鬆搞定RabbitMQ（二）——工作佇列之訊息分發機制

       上一篇博文中簡單介紹了一下RabbitMQ的基礎知識，並寫了一個經典語言入門程式——HelloWorld。本篇博文中我們將會建立一個工作佇列用來在工作者（consumer）間分發耗時任務。同樣是翻譯的官網例項。 工作佇列        在前一篇博文中，我們完

rabbitMq消費者角度：訊息分發、訊息應答（ACK）、公平分發

rabbitMq交換機簡介中介紹了rabbitMq的四種類型交換機；rabbitMq生產者角度一篇從生產者角度介紹如何保證訊息被正確傳送到伺服器，如果未正確傳送如何處理；本篇部落格將從消費者角度介紹三個問題：佇列分發訊息到消費者的規則、如何確保訊息一定被正確接受

RabbitMQ學習(四)——訊息分發機制

在前面的一篇博文中，我們對RabbitMQ中的交換機有了大致的瞭解，同時結合Spring Boot的例項，讓我們對RabbitMQ的用法有了更清晰的認識。如果忘記了可以去複習一下，RabbitMQ學習(三)——探索交換機(Exchange)，結合SpringBo

RabbitMQ消息分發輪詢

python我們首先下載pika，以及rabbitMQ，和ir語言，rabbitMQ是由ir語言編寫的消息隊列的使用過程大概如下：（1）客戶端連接到消息隊列服務器，打開一個channel。channel：消息通道，在客戶端的每個連接裏，可建立多個channel，每個channel代表一個會話任務。（2）客戶端

RabbitMQ消息分發輪詢和Message Acknowledgment

如果 實現 sage int cal llb received 一點 最終 一、消息分發 　　RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息並最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取消息並消費。 　　 　　多個消費

Python-RabbitMQ消息分發機制

連接 處的 code top exclusive exc 但是 現在 pika 上一篇中的例子是一個生產者對應一個消費者，那能不能一個生產者對應一個消費者呢？ 下面來測試一下，順便觀察一下它的分發策略。。。 步驟一：先編輯生產者代碼（rabbit_send.py） #top

Java使用RabbitMQ之公平分發

env 接受 col java catch conn ech exceptio don 發送消息： 1 package org.study.workfair; 2 3 import com.rabbitmq.client.Channel; 4 import co

RabbitMQ消費者訊息權重，

有的消費者處理速度快，有的消費者處理速度慢，我們想給處理快的多發訊息，處理慢的少發訊息， 怎麼辦呢？按照之前的輪詢模式，肯定不行的，這裡可以檢測訊息數量，如果消費者正在處理就不給他發 .... def callback(ch, method, properties, body): print

RabbitMQ（訊息中介軟體）在工作中的應用場景

  RabbitMQ（訊息中介軟體）在工作中的應用場景  1、跨系統的非同步通訊，所有需要非同步互動的地方都可以使用訊息佇列。就像我們除了打電話（同步）以外，還需要發簡訊，發電子郵件（非同步）的通訊方式。   2、多個應用之間的耦合，由於訊息是平臺無關和語言

RabbitMQ（訊息佇列）叢集配置與使用篇

介紹 MQ全稱為Message Queue, 訊息佇列(MQ)是一種應用程式對應用程式的通訊方法。應用程式通過讀寫出入佇列的訊息(針對應用程式的資料)來通訊，而無需專用連線來連結它們。訊息傳遞指的是程式之間通過在訊息中傳送資料進行通訊，而不是通過直接呼叫彼此來通訊，直接呼叫通常是用於諸如遠端過程呼叫的

Redis與RabbitMQ作為訊息佇列的對比

RabbitMQ RabbitMQ是實現AMQP（高階訊息佇列協議）的訊息中介軟體的一種，最初起源於金融系統，用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道

rabbitmq 傳送訊息格式注意

在rabbitmq的網頁版中，有publish ，用了傳送。 delivery_mode: 2 headers: content_type:

通過HTTP請求獲取RabbitMQ佇列訊息的方法

https://www.jianshu.com/p/9edf12944935 由於自動化測試的需要，現在要用一種簡單的方法可以獲取並驗證RabbitMQ中的訊息，而不是費勁通過程式再寫一個消費者。其實，RabbitMQ提供了HTTP API，可以幫助我們實現這個需求。 1. RabbitMQ

Kafka訊息保留策略

Kafka Broker預設的訊息保留策略是：要麼保留一定時間，要麼保留到訊息達到一定大小的位元組數。 當訊息達到設定的條件上限時，舊訊息就會過期並被刪除，所以，在任何時刻，可用訊息的總量都不會超過配置引數所指定的大小。 topic可以配置自己的保留策略，可以將訊息保留到不再使用他們為止。