消息隊列----RabbitMQ

時間 2019-12-06

原文原文鏈接

1、RabbitMQ概覽html

1.什麼是消息隊列？java

「消息隊列」是在消息的傳輸過程當中保存消息的容器。消息隊列管理器在將消息從它的源中繼到它的目標時充當中間人。隊列的主要目的是提供路由並保證消息的傳遞；若是發送消息時接收者不可用，消息隊列會保留消息，直到能夠成功地傳遞它。web

2.爲何要使用消息隊列？數據庫

主要是因爲在高併發環境下，因爲來不及同步處理，請求每每會發生堵塞。如：大量的insert，update等請求同時到達數據庫，致使無數的行鎖表鎖，因請求會堆積過多，而觸發too many connections錯誤。經過使用消息隊列，咱們能夠異步處理請求，從而緩解系統的壓力。windows

3.RabbitMQ瀏覽器

　　RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現。（官網：http://www.rabbitmq.com/）服務器

　　RabbitMQ是一個消息代理：它接受並轉發消息。併發

4.RabbitMQ的安裝（windows平臺）異步

　　1）先決條件：下載erlang語言環境。version:20.1 （http://www.erlang.org/downloads）ide

　　安裝完成後，配置E環境變量RLANG_HOME爲安裝的根目錄，並把%ERLANG_HOME%\bin追加到PATH環境變量

　　2）下載RabbitMQ 。version:3.6.12（http://www.rabbitmq.com/install-windows.html）

　　執行sbin目錄下的rabbitmq-server.bat來啓動服務，默認端口：5672

　　　　注意：elang 和 RabbitMQ的版本對應要求。

　　3）開啓管理插件

　　　　Rabbitmq-management 插件提供了用於管理、監控rabbitmq server的http 應用程序接口，基於瀏覽器的用戶界面和命令行工具rabbitmqadmin。

　　　　開啓rabbitmq管理插件：

　　　　rabbitmq-plugins enable rabbitmq_management

　　4）啓動 rabbitmq 服務，並使用默認用戶guest登陸web 用戶界面。

　　　　（使用windows系統服務或者rabbit root/sbin/rabbitmq-server.bat啓動）:

　　　　登入

　　5）RabbitMQ Hello word 之（消息發佈）

　　本文以JAVA語言爲例進行實現。官網提供了PHP、C#、JAVASCRIPT等多種語言的實現代碼。

　　注：須要引入rabbitmq java client包，Maven倉庫信息：groupId: com.rabbitmq artifactId: amqp-client

package com.example.mq;

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 Sender_1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        //對於全部的參數，rabbitmq server在本地都有一個合適的默認值
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        //可以使用鏈接字符串代替
        //factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");
        Connection connection=factory.newConnection();

        //使用鏈接打開一個通道。Channel能夠用於發送、接收信息
        Channel channel=connection.createChannel();


        //生命 一個 Queues用於發送信息。
        String queueName="queue_1";
        channel.queueDeclare(queueName,false,false,false,null);
        String message="Hello world 1!";
        channel.basicPublish("",queueName,null,message.getBytes());
        System.out.println("[x] Sent '"+message+"'");

        channel.close();
        connection.close();
    }
}

　　　　執行消息發佈程序，分別以mesaage="Hello world 1!"； mesaage="Hello world 2!"；mesaage="Hello world 3!"；發送三條消息。

　　　　發送成功後，能夠在管理頁面看到rabbitmq server 上有一個名爲queue_1的消息隊列，其中包含三條消息。

　　　　也可使用命令查看隊列和其包含的消息：rabbitmqctl.bat list_queues

　　6）RabbitMQ Hello word 之（消息接收）

package com.example.mq;

import com.rabbitmq.client.*;

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

public class Receiver_1 {
    private final static String queueName="queue_1";
    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.queueDeclare(queueName,false,false,false,null);
        System.out.println("[*]waiting for message .to exit press ctrl+c...");

        Consumer consumer=new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws UnsupportedEncodingException {
                String message =new String(body,"UTF-8");
                System.out.println("[x] Received '"+message+"'");
            }
        };
        channel.basicConsume(queueName,true,consumer);
    }
}