RabbitMQ簡(jiǎn)單模式–直連，相當(dāng)于點(diǎn)對(duì)點(diǎn)連接一樣

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

P：生產(chǎn)者，也就是要發(fā)送消息的程序
C：消費(fèi)者：消息的接受者，會(huì)一直等待消息到來(lái)。
queue：消息隊(duì)列，圖中紅色部分。類似一個(gè)郵箱，
可以緩存消息；生產(chǎn)者向其中投遞消息，消費(fèi)者從其中取出消息。
應(yīng)用場(chǎng)景我們可以應(yīng)用于用戶的登錄，登錄以后向消息隊(duì)列中發(fā)送一條登錄消息，后臺(tái)讀取到消息可以對(duì)用戶登錄積分加一

代碼如下：
連接消息隊(duì)列的工具類

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 {//創(chuàng)建連接工廠，在類加載時(shí)創(chuàng)建，只創(chuàng)建一次private static ConnectionFactory connectionFactory;static {connectionFactory = new ConnectionFactory();//設(shè)置端口號(hào)connectionFactory.setPort(5672);//設(shè)置主機(jī)名稱connectionFactory.setVirtualHost("/ems");//設(shè)置訪問(wèn)虛擬主機(jī)的用戶名和密碼connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");}//提供一個(gè)創(chuàng)建連接的方法public static Connection getConnection(String host) {try {// 設(shè)置連接rabbitmq主機(jī)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();}}} }

生產(chǎn)者

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();//將通道與消息隊(duì)列綁定//參數(shù)一：隊(duì)列名稱,參數(shù)二：是否持久化,參數(shù)三：是否獨(dú)占隊(duì)列，參數(shù)四：是否在消費(fèi)完成后自動(dòng)刪除隊(duì)列，channel.queueDeclare("hello",false,false,false,null);//發(fā)布消息//參數(shù)一：交換機(jī)名稱，參數(shù)二：隊(duì)列名稱，參數(shù)三：消息額外屬性，參數(shù)四：消息內(nèi)容channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());RabbitMQUtils.closeChannelAndConnection(channel,connection);}}

消費(fèi)者

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");//創(chuàng)建連接通道Channel channel = connection.createChannel();//通道綁定對(duì)象channel.queueDeclare("hello",false,false,false,null);//消費(fèi)消息//參數(shù)一：隊(duì)列名稱 參數(shù)二：開(kāi)始消費(fèi)時(shí)自動(dòng)確認(rèn)機(jī)制 參數(shù)三：消費(fèi)時(shí)的回調(diào)接口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();} }

每運(yùn)行一次provider都會(huì)向消息隊(duì)列中添加一條消息，consumer一直運(yùn)行著不會(huì)停止，取出隊(duì)列中的消息

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。