RabbitMQ簡單模式–直連，相當于點對點連接一樣

在上圖的模型中，有以下概念：

P：生產者，也就是要發送消息的程序
C：消費者：消息的接受者，會一直等待消息到來。
queue：消息隊列，圖中紅色部分。類似一個郵箱，
可以緩存消息；生產者向其中投遞消息，消費者從其中取出消息。
應用場景我們可以應用于用戶的登錄，登錄以后向消息隊列中發送一條登錄消息，后臺讀取到消息可以對用戶登錄積分加一

代碼如下：
連接消息隊列的工具類

package com.zuoan.utils;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;import java.io.IOException; import java.util.concurrent.TimeoutException;/*** @Description: TODO* @Author: 黃石軍* @CreateTime: 2022/4/9 16:56* @Company:*/ public class RabbitMQUtils {//創建連接工廠，在類加載時創建，只創建一次private static ConnectionFactory connectionFactory;static {connectionFactory = new ConnectionFactory();//設置端口號connectionFactory.setPort(5672);//設置主機名稱connectionFactory.setVirtualHost("/ems");//設置訪問虛擬主機的用戶名和密碼connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");}//提供一個創建連接的方法public static Connection getConnection(String host) {try {// 設置連接rabbitmq主機connectionFactory.setHost(host);return connectionFactory.newConnection();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}return null;}public static void closeChannelAndConnection(Channel channel, Connection connection) {if (channel != null) {try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}if (connection != null) {try {connection.close();} catch (IOException e) {e.printStackTrace();}}} }

生產者

package com.zuoan;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.zuoan.utils.RabbitMQUtils; import org.junit.Test;import java.io.IOException; import java.util.concurrent.TimeoutException;/*** @Description: TODO* @Author: 黃石軍* @CreateTime: 2022/4/9 15:53* @Company:*/ public class Provider {@Testpublic void testSendMessage() throws IOException, TimeoutException {Connection connection = RabbitMQUtils.getConnection("192.168.128.100");//獲取連接中通道Channel channel = connection.createChannel();//將通道與消息隊列綁定//參數一：隊列名稱,參數二：是否持久化,參數三：是否獨占隊列，參數四：是否在消費完成后自動刪除隊列，channel.queueDeclare("hello",false,false,false,null);//發布消息//參數一：交換機名稱，參數二：隊列名稱，參數三：消息額外屬性，參數四：消息內容channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());RabbitMQUtils.closeChannelAndConnection(channel,connection);}}

消費者

package com.zuoan;import com.rabbitmq.client.*; import com.zuoan.utils.RabbitMQUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;/*** @Description: TODO* @Author: 黃石軍* @CreateTime: 2022/4/9 16:14* @Company:*/ public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtils.getConnection("192.168.128.100");//創建連接通道Channel channel = connection.createChannel();//通道綁定對象channel.queueDeclare("hello",false,false,false,null);//消費消息//參數一：隊列名稱 參數二：開始消費時自動確認機制 參數三：消費時的回調接口channel.basicConsume("hello",true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("new String(body) = "+ new String(body));}});// channel.close(); // connection.close();} }

每運行一次provider都會向消息隊列中添加一條消息，consumer一直運行著不會停止，取出隊列中的消息

總結

以上是生活随笔為你收集整理的RabbitMQ第一种消息模型--直连模型的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。