一、概述

RabbitMQ 有很多的使用模式,具體的可以去參考官網,本次我們將簡單的搭建一個 Hello World,該模式下只有一個 Producer(生產者)、一個佇列(Queue)、一個消費者(Consumer)

二、RabbitMQ 核心工作原理

三、引入對應的依賴

<dependencies>   
    <!--rabbitmq 客戶端依賴-->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.8.0</version>
    </dependency>
</dependencies>

<build>
    <plugins>
    <!--指定編譯的版本-->    
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>8</source>
                <target>8</target>
            </configuration>
        </plugin>
    </plugins>
</build>
 

四、生產者

public class Producer {
    private static final String QUEUE_NAME = "helloWorld";
    private static final String HOST_ADDRESS = "192.168.59.130";
    private static final String USER_NAME = "admin";
    private static final String PASSWORD = "admin123";

    public static void main(String[] args) throws Exception {
        // 建立連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 設定主機地址、使用者名稱、密碼
        factory.setHost(HOST_ADDRESS);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        // 使用連線工廠建立連線物件
        Connection connection = factory.newConnection();
        // 使用連線物件建立通道
        Channel channel = connection.createChannel();
        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, true, null);
        // 定義要傳送的訊息
        String message = "hello rabbitmq...";
        // 傳送訊息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));

        System.out.println("producer send message successfully...");
    }
}
 

五、消費者

public class Consumer {
    private static final String QUEUE_NAME = "helloWorld";
    private static final String HOST_ADDRESS = "192.168.59.130";
    private static final String USER_NAME = "admin";
    private static final String PASSWORD = "admin123";

    public static void main(String[] args) throws Exception {
        // 建立連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 設定主機地址、使用者名稱、密碼
        factory.setHost(HOST_ADDRESS);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        // 連線工廠建立連線物件
        Connection connection = factory.newConnection();
        // 連線物件建立通道
        Channel channel = connection.createChannel();
        // 宣告佇列
        // exclusive:是否宣告一個排它的佇列 true:是 false:否
        // autoDelete:當最後一個消費者斷開連線之後,該佇列會自動刪除
        channel.queueDeclare(QUEUE_NAME, false, false, true, null);

        // 訊息傳送時會回撥該介面
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };
        // 取消消費時會回撥該介面
        CancelCallback cancelCallback = (x) -> {
            System.out.println("刪除訊息時的回撥.....");
        };

        // 接收訊息
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

        System.out.println("訊息消費完成");
    }
}
 

先啟動生產者傳送訊息,然後再啟動消費者便可以消費具體的訊息了