RabbitMQ是使用Erlang語(yǔ)言開(kāi)發(fā)的消息中間件,其遵循了高級(jí)消息隊(duì)列協(xié)議(Advanced Message Queuing Protocol,AMQP)。

與Kafka等消息隊(duì)列相比，RabbitMQ最大的優(yōu)勢(shì)在于其較高的可靠性:

提供確認(rèn)(ACK)和重傳機(jī)制保證消息完成消費(fèi),消費(fèi)者異常不會(huì)導(dǎo)致消息丟失

提供消息持久化機(jī)制,broker崩潰不會(huì)導(dǎo)致消息丟失

集群模式下工作,保證高可用

因?yàn)榫哂休^高可靠性和一致性,RabbitMQ可以勝任訂單處理、秒殺等一致性要求較高的業(yè)務(wù)場(chǎng)景。

RabbitMQ概念與機(jī)制

RabbitMQ中的概念模型:

Broker:消息中間件實(shí)例,可能是單個(gè)節(jié)點(diǎn)也可能是運(yùn)行在多節(jié)點(diǎn)集群上的邏輯實(shí)體

消息(Message):消息由消息頭和消息體兩部分組成。消息頭中包括routing-key、priority等標(biāo)準(zhǔn)消息頭以及其它自定義消息頭，用于定義RabbitMQ對(duì)消息行為。消息體是字節(jié)流，包含消息內(nèi)容。

連接(Connection):客戶(hù)端與Broker之間的TCP連接

信道(Channel):Channel是建立在TCP連接上的邏輯(虛擬)連接。多個(gè)Channel復(fù)用同一個(gè)TCP連接,以避免建立TCP連接的巨大開(kāi)銷(xiāo)。RabbitMQ官方要求每個(gè)線程使用獨(dú)立的Channel,禁止多個(gè)線程共用Channel。

生產(chǎn)者(Publisher):發(fā)送消息的客戶(hù)端線程

消費(fèi)者(Consumer):處理消息的客戶(hù)端線程

交換機(jī)(Exchange):交換機(jī)負(fù)責(zé)將消息投遞到相應(yīng)的隊(duì)列

隊(duì)列(Queue):接收并保存交換機(jī)投遞的消息，直至被消費(fèi)者成功消費(fèi)。邏輯結(jié)構(gòu)遵循先進(jìn)先出FIFO。

綁定(Binding):將隊(duì)列(Queue)注冊(cè)到交換機(jī)(Exchange)的路由表

虛擬主機(jī)(Vhost):每個(gè)Broker下可建立多個(gè)vhost,每個(gè)vhost可建立獨(dú)立的Exchange、Queue、綁定及權(quán)限系統(tǒng)。同一個(gè)Broker下的vhost共享Connection、Channel和用戶(hù)系統(tǒng)，就是說(shuō)可以使用同一個(gè)用戶(hù)身份使用同一個(gè)Channel訪問(wèn)不同vhost。

交換機(jī)(Exchange)

生產(chǎn)者發(fā)送的消息會(huì)首先送到交換機(jī)(Exchange),交換機(jī)根據(jù)自身類(lèi)型和消息的routing-key等信息將消息投遞到綁定的消息隊(duì)列中。

RabbitMQ中的四種標(biāo)準(zhǔn)交換機(jī):

direct:如果消息的routing-key與隊(duì)列的binding-key完全相同，direct類(lèi)型的交換機(jī)則會(huì)將消息投遞到該隊(duì)列中。

多個(gè)隊(duì)列可以使用相同的binding-key綁定到同一個(gè)direct交換機(jī)，direct交換機(jī)會(huì)把消息投遞到所有binding-key與消息routing-key相同的隊(duì)列

topic:允許隊(duì)列的binding-key中包含通配符*和#,topic交換機(jī)會(huì)將消息投遞到binding-key與routing-key匹配的隊(duì)列中。

通配符按照關(guān)鍵字進(jìn)行匹配，如news.cn.a中的關(guān)鍵字是news、cn和a，即關(guān)鍵字按照.分割

#通配符匹配0個(gè)或多個(gè)關(guān)鍵字,news.#.a可以匹配news.a,news.cn.a和news.asia.cn.a等

*通配符匹配一個(gè)關(guān)鍵字,news.*.a匹配news.cn.a不匹配news.a、news.asia.cn.a

fanout:fanout交換機(jī)不進(jìn)行任何匹配,將消息投遞到所有綁定的隊(duì)列

header:header交換機(jī)根據(jù)消息頭進(jìn)行投遞，現(xiàn)在已較少使用

我們可以使用RabbitMQ的插件機(jī)制使用第三方交換機(jī)或自行開(kāi)發(fā)交換機(jī)。如實(shí)現(xiàn)延時(shí)投遞的delayed-message-exchange。

消息頭中的delivery-mode可以設(shè)置為persistent(持久化)或者transient(易失)。Exchange和Queue在處理持久化的消息時(shí)都會(huì)先將消息寫(xiě)入磁盤(pán)中再進(jìn)行下一步處理,即使RabbitMQ崩潰也不會(huì)丟失。

消費(fèi)者客戶(hù)端通常使用的channel.basicConsume使用推(push)模式投遞消息,即當(dāng)有新消息時(shí)Broker通過(guò)channel主動(dòng)向客戶(hù)端發(fā)送消息。客戶(hù)端也可以使用channel.basicGet從Broker拉取消息。

ACK機(jī)制

RabbitMQ提供了確認(rèn)送達(dá)(acknowledge)機(jī)制保證消息被正確處理不會(huì)丟失。

確認(rèn)送達(dá)的回執(zhí)有三種:

ACK:消息已被成功處理

NACK:消息處理異常,需要重新投遞

REJECT:消息非法,丟棄消息

RabbitMQ的Queue可以設(shè)置no_ack=true,則消息被投遞后即刪除不等待回執(zhí)。

channel.basicConsume可以指定auto_ack模式，若auto_ack=true當(dāng)客戶(hù)端收到完整消息后即會(huì)自動(dòng)發(fā)出ACK回執(zhí)，否則必須顯式的發(fā)出回執(zhí)。

Java代碼示例

首先安裝并啟動(dòng)RabbitMQ實(shí)例,Mac用戶(hù)可以使用Homebrew進(jìn)行安裝:

brew install rabbitmq

啟動(dòng)服務(wù):

brew services start rabbitmq

或者使用官方docker鏡像:

docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management

RabbitMQ官網(wǎng)提供了Ubuntu、RPM以及Windows等多種平臺(tái)安裝方式。

RabbitMQ默認(rèn)TCP端口為5672,Web控制臺(tái)默認(rèn)端口15672。

在Maven中添加依賴(lài):

<dependency>
 <groupId>com.rabbitmq</groupId>
 <artifactId>amqp-client</artifactId>
 <version>5.5.1</version>
</dependency>

編寫(xiě)生產(chǎn)者:

package rabbit;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @author finley
 */
public class RabbitProducer {

 public static void main(String[] args) throws IOException, TimeoutException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setUsername("guest");
  factory.setPassword("guest");
  factory.setHost("localhost");
  try (Connection conn = factory.newConnection();
    Channel channel = conn.createChannel()) {
   String exchangeName = "test-exchange";
   channel.exchangeDeclare(exchangeName, "direct", true);

   String routingKey = "hello";

   byte[] msg = "hello world".getBytes();
   AMQP.BasicProperties.Builder propsBuilder = new AMQP.BasicProperties.Builder();
   propsBuilder.deliveryMode(2); // persistent
   propsBuilder.priority(0); // normal
   propsBuilder.contentType("text/plain");
   channel.basicPublish(exchangeName, routingKey, propsBuilder.build(), msg);
  }
 }
}

編寫(xiě)消費(fèi)者:

package rabbit;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.*;

/**
 * @author finley
 */
public class RabbitConsumer {

 public static void main(String[] args) throws IOException, TimeoutException {
  ConnectionFactory factory = new ConnectionFactory();
  factory.setUsername("guest");
  factory.setPassword("guest");
  factory.setHost("localhost");
  try (Connection conn = factory.newConnection();
    Channel channel = conn.createChannel()) {
   String exchangeName = "test-exchange";
   channel.exchangeDeclare(exchangeName, "direct", true);

   String queueName = channel.queueDeclare().getQueue();
   String bindingKey = "hello";
   channel.queueBind(queueName, exchangeName, bindingKey);

   while(true) {
    channel.basicConsume(queueName, false, "", new DefaultConsumer(channel) {
     @Override
     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
      String routingKey = envelope.getRoutingKey();
      String contentType = properties.getContentType();
      String bodyStr = new String(body, "UTF-8");
      System.out.println("routingKey: " + routingKey + ", contentType: " + contentType + ", body: " + bodyStr);
      long deliveryTag = envelope.getDeliveryTag();
      channel.basicAck(deliveryTag, false);
     }
    });
   }
  }
 }

}

RabbitMQ是流行的開(kāi)源消息隊(duì)列系統(tǒng)，用erlang語(yǔ)言開(kāi)發(fā)，RabbitMQ是AMQP(高級(jí)消息隊(duì)列協(xié)議)的標(biāo)準(zhǔn)實(shí)現(xiàn)。采用該技術(shù)，我們可以實(shí)現(xiàn)異步處理、流量削峰、系統(tǒng)解耦；動(dòng)力節(jié)點(diǎn)RabbitMQ視頻教程，課程將講授RabbitMQ的環(huán)境搭建、消息的發(fā)送與接收、消息確認(rèn)、與SpringBoot集成等，讓大家快速掌握RabbitMQ技術(shù)，以適應(yīng)項(xiàng)目開(kāi)發(fā)的需要；

以上就是動(dòng)力節(jié)點(diǎn)Java培訓(xùn)機(jī)構(gòu)的小編針對(duì)“Rabbitmq消息中間件視頻，入門(mén)學(xué)習(xí)”的內(nèi)容進(jìn)行的回答，希望對(duì)大家有所幫助，如有疑問(wèn)，請(qǐng)?jiān)诰€咨詢(xún)，有專(zhuān)業(yè)老師隨時(shí)為你服務(wù)。