1. 程式人生 > >RabbitMQ系列之---初識RabbitMQ

RabbitMQ系列之---初識RabbitMQ

為什麼要使用RabbitMQ?

訊息佇列的作用

  • 非同步呼叫
  • 系統解耦
  • 削峰限流
  • 訊息通訊

訊息佇列的缺點

  • 系統可用性降低
  • 系統穩定性降低
  • 分散式一致性問題(可靠訊息最終一致性的分散式事務方案解決)

RabbitMQ的優勢

  • 支援高併發、高吞吐、效能好
  • 有完善的後臺管理介面
  • 它還支援叢集化、高可用部署架構、訊息高可靠支援
  • RabbitMQ的開源社群很活躍,較高頻率的迭代版本,來修復發現的bug以及進行各種優化
  • 最重要的是它是開源免費的。

RabbitMQ的缺點

  • 它是基於erlang語言開發的,所以導致較為難以分析裡面的原始碼,也較難進行深層次的原始碼定製和改造,畢竟需要較為紮實的erlang語言功底才可以。

核心概念

  • Server:又稱Broker,接受客戶端的連線,實現AMQP實體服務;
  • Connection:連線,應用程式與Broker的網路連線;
  • Channel:網路通道,也稱通道,幾乎所有的通道都在Channel中進行的,Channel是進行訊息讀寫的通道。客戶端可建立多個Channel,每個Channel代表一個會話任務;是建立在“真實的”TCP連線內的虛擬連線。AMQP命令都是通過通道傳送出去的。那麼我們為什麼需要通道呢?為什麼不直接通過TCP連線傳送AMQP命令呢?因為作業系統建立和銷燬TCP會話是非常昂貴的開銷,而且作業系統只能建立數量不多的TCP連線,很快就達到效能瓶頸,無法滿足高效能的需求。
  • Message:訊息,伺服器和應用程式之間傳送的資料,由Properties和Body組成。Properties可以對訊息進行修飾,比如訊息的優先順序、等高階特性;Body則就是訊息體的內容;
  • Virtual host:虛擬地址,用於進行邏輯隔離,最上層的訊息路由,並非物理概念。一個Virtual Host裡面有若干個Exchange和Queue,同一個Virtual Host裡面不能有相同名稱的Exchange或者Queue;
  • Exchange:交換機,接收訊息,根據路由鍵轉發訊息到繫結的佇列;
    交換機的型別:direct:直連,fanout:廣播,headers:以...開頭,topic:主題
    參考:Spring Boot RabbitMQ 四種交換器
  • Binding:Exchange和Queue之間的虛擬連線,binding中可以包含routing key;
  • Routing Key:一個路由規則,虛擬機器可用它來確定如何路由一個特定的訊息;
  • Queue:也稱為Message Queue,訊息佇列,儲存訊息並將它們轉發給消費者。

幾種交換機

  • Direct:直連交換機

    所有傳送到Direct Exchange的訊息被轉發到RouteKey中指定的Queue
    注意:Direct模式可以直接使用RabbitMQ自帶的Exchange:default,並以佇列名作為路由鍵。 Exchange,所以不需要將Exchange進行任何繫結(binding)操作,訊息傳遞時,RouteKey必須完全匹配才會被佇列接收,否則該訊息會被拋棄。
public class Producer4DirectExchange {
    public static void main(String[] args) throws IOException, TimeoutException {
        ...
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_direct_exchange";
        String routingKey = "test.direct";
        String msg = "Hello World RabbitMQ 4 Direct Exchange Message...";

        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
        ...
    }
}
public class Consumer4DirectExchange {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
       ...
        Channel channel = connection.createChannel();

        String exchangeName = "test_direct_exchange";
        String exchangeType = "direct";
        String queueName = "test_direct_queue";
        String routingKey = "test.direct";
        
        // 宣告交換機
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
        // 宣告佇列
        channel.queueDeclare(queueName, false, false, false, null);
        // 交換機與佇列繫結
        channel.queueBind(queueName, exchangeName, routingKey);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName,true,queueingConsumer);

        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("收到訊息: " + msg);
        }
    }
}
  • Topic:主題交換機

    所有傳送到Topic Exchange的訊息被轉發到所有關心RouteKey中指定的Queue上,Exchange將RouteKey和某個Topic進行模糊匹配,此時佇列需要繫結一個Topic。
    注意:可以使用萬用字元進行模糊匹配
"#" 匹配一個或多個詞
"*" 匹配一個詞
"." 是單詞的分隔符

例如:
"log.#":能夠匹配到"log.info.oa
"log.*":只會匹配到"log.error"
public class Producer4TopicExchange {
    public static void main(String[] args) throws IOException, TimeoutException {
        ...
        Channel channel = connection.createChannel();

        String exchangeName = "test_topic_exchange";
        String routingKey1 = "user.save";
        String routingKey2 = "user.update";
        String routingKey3 = "user.delete.ok";

        String msg = "Hello World RabbitMQ 4 Topic Exchange Message...";

        channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
        channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());
        channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());
        ...
    }
}
public class Consumer4TopicExchange {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ...
        Channel channel = connection.createChannel();

        String exchangeName = "test_topic_exchange";
        String exchangeType = "topic";
        String queueName = "test_topic_queue";
        String routingKey = "user.*";

        channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName, true, queueingConsumer);

        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("收到訊息: " + msg);
        }
    }
}
  • Fanout Exchange:廣播交換機

    不處理路由鍵,只需要簡單的將佇列繫結到交換機;
    傳送到交換機的訊息都會被轉發到與該交換機繫結的所有佇列上;
    Fanout交換機轉發訊息是最快的。
public class Producer4FanoutExchange {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ...
        Channel channel = connection.createChannel();

        String exchangeName = "test_fanout_exchange";
        String routingKey = "";

        String msg = "Hello World RabbitMQ 4 Fanout Exchange Message...";

        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
        ...
    }
}
public class Consumer4FanoutExchange {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ...
        Channel channel = connection.createChannel();

        String exchangeName = "test_fanout_exchange";
        String exchangeType = "fanout";
        String queueName = "test_fanout_queue";
        String routingKey = "";

        channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName, true, queueingConsumer);

        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("收到訊息: " + msg);
        }
    }
}

參考資料:

  1. 石杉大神訊息中介軟體系列文章
  2. 慕課網《RabbitMQ訊息中介軟體技術精講》
  3. 書《RabbitMQ實戰-高效部署分散式訊息佇列》