Spring?boot?整合RabbitMQ實(shí)現(xiàn)通過RabbitMQ進(jìn)行項(xiàng)目的連接
什么是RabbitMQ
一款基于AMQP用于軟件之間通信的中間件 。消費(fèi)并不需要確保提供方存在,實(shí)現(xiàn)了服務(wù)之間的高度解耦
AMQP，即Advanced Message Queuing Protocol，高級消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、隊(duì)列、路由（包括點(diǎn)對點(diǎn)和發(fā)布/訂閱）、可靠性、安全。
RabbitMQ是一個(gè)開源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。
基本組成：Queue：消息隊(duì)列，存儲消息的隊(duì)列，消息到達(dá)隊(duì)列并轉(zhuǎn)發(fā)給指定的消費(fèi)方;
Exchange消息隊(duì)列交換機(jī)，按一定的規(guī)則將消息路由轉(zhuǎn)發(fā)到某個(gè)隊(duì)列，對消息進(jìn)行過慮;
消費(fèi)者：消費(fèi)信息;
提供者：發(fā)送消息;
官網(wǎng) Messaging that just works — RabbitMQ
消息隊(duì)列：接受并轉(zhuǎn)發(fā)消息，類似于快遞公司
product : 消息的發(fā)送者,生產(chǎn)者
consumer：消息的消費(fèi)者，從隊(duì)列獲取消息，并且使用
queue :先進(jìn)先出，一個(gè)queue可以對應(yīng)多個(gè)consumer
消息隊(duì)列的優(yōu)點(diǎn)
代碼解耦,提高系統(tǒng)穩(wěn)定性
應(yīng)對流量高峰,降低流量沖擊，面對秒殺這種情況時(shí)，請求進(jìn)來先去排隊(duì)，可以保證系統(tǒng)的穩(wěn)定
異步執(zhí)行,提高系統(tǒng)響應(yīng)速度
消息隊(duì)列的特性
性能好
它是一種基礎(chǔ)組件
支持消息確認(rèn)，為了防止數(shù)據(jù)丟失以及應(yīng)對特殊情況，在數(shù)據(jù)沒有處理完，沒有確認(rèn)之前消息不會(huì)丟掉。
RabbitMQ特點(diǎn)
路由能力靈活強(qiáng)大
開源免費(fèi)
支持編程語言多
應(yīng)用廣泛，社區(qū)活躍
有開箱即用的監(jiān)控和管理后臺
RabbitMQ核心概念
生產(chǎn)者數(shù)量是不限制的，生產(chǎn)者生產(chǎn)的消息Message進(jìn)入交換機(jī)，交換一可以連接多個(gè)隊(duì)列也可以僅連接一個(gè)對聯(lián)，交換機(jī)與隊(duì)列的關(guān)系是不固定的，交換機(jī)會(huì)綁定到隊(duì)列上（Binding）根據(jù)的規(guī)則就是Routing Key路由鍵用來確定交換機(jī)與隊(duì)列如何進(jìn)行綁定 ，消息經(jīng)過交換機(jī)經(jīng)過連接發(fā)送個(gè)消費(fèi)者，在連接中多多個(gè)信道，數(shù)據(jù)都是在信道中進(jìn)行讀寫的，消費(fèi)者從中提取想要的消息進(jìn)行處理。Broker(服務(wù)實(shí)例)也就是服務(wù)端，Virtual Host (虛擬主機(jī))同一個(gè)RabbitMQ可能給多個(gè)服務(wù)進(jìn)行使用，服務(wù)與服務(wù)之間想要隔離開就可以使用虛擬主機(jī)進(jìn)行隔離。
Producer :消息生產(chǎn)者
Message :消息
Exchange :交換機(jī)
Binding :綁定交換機(jī)和隊(duì)列
Routing key :路由鍵,決定路由規(guī)則
Queue :隊(duì)列,存儲消息
Connection :連接服務(wù)端
Channel :信道,讀寫數(shù)據(jù).
Consumer :消費(fèi)者
Broker :服務(wù)實(shí)例
Virtual host :虛擬主機(jī)，用于區(qū)分不同服務(wù),類似于不同域名,不會(huì)相互影響
安裝RabbitMQ
LINUX環(huán)境下安裝3.8.2 使用Xshell
先進(jìn)行環(huán)境配置
連接成功以后輸入
echo "export LC_ALL=en_US.UTF-8" >> /etc/profile 把編碼設(shè)置成utf-8
source /etc/profile 使設(shè)置生效
輸入curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash 配置RabbitMQ源
看到這個(gè)命令就可以進(jìn)行下一步了
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash
配置erlang環(huán)境
看到這個(gè)命令進(jìn)行下一步
sudo yum install rabbitmq-server-3.8.2-1.el7.noarch
輸入y
常用命令
開啟web管理界面
rabbitmq-plugins enable rabbitmq_management
停止RabbitMQ
rabbitmqctl stop
設(shè)置開機(jī)啟動(dòng)
systemctl enable rabbitmq-server
啟動(dòng)RabbitMQ
systemctl start rabbitmq-server
看看端口有沒有起來，查看狀態(tài)
rabbitmqctl status
要檢查RabbitMQ服務(wù)器的狀態(tài)，請運(yùn)行：
systemctl status rabbitmq-server
Windows
先安裝erlang并配置環(huán)境，安裝RabbitMQ
鏈接：https://pan.baidu.com/s/1S4D2zh-NSoXh-QPQVNBi-w
提取碼：1111
這里直接放上鏈接，erlang安裝好后要去配置環(huán)境
解壓縮后sbin目錄下，rabbitmq-server.bat 這個(gè)文件就是啟動(dòng)
用終端cmd輸入：
cd d:\你的RabbitMQ按照地址\sbin
rabbitmq-plugins enable rabbitmq_management
rabbitmq-server
然后就可以用guest訪問http://127.0.0.1:15672/#/
賬號密碼都是guest
RabbitMQ實(shí)操分布了解
1 生產(chǎn)者
這里的前提是你有個(gè)云服務(wù)器，并且已經(jīng)完成了配置，為了操作簡便這里就用本機(jī)了哈
我們要有一個(gè)管理者啊在sbin目錄輸入
rabbitmqctl add_user newadmin newpassword
rabbitmqctl set_user_tags newadmin administrator
rabbitmqctl set_permissions -p / newadmin ".*" ".*" ".*"//這一步已經(jīng)把在虛擬主機(jī)上把權(quán)限配置了
賬號test 密碼123456
新建一個(gè)mavene項(xiàng)目，
2 引入依賴
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<!-- 記錄日志-->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-nop</artifactId>
<version>1.7.29</version>
</dependency>
</dependencies>/**
* 描述 發(fā)送類 連接到服務(wù)端 發(fā)送 退出
*/
public class Send {
//設(shè)置隊(duì)列的名字
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//創(chuàng)建連接工廠
ConnectionFactory factory = new ConnectionFactory();
//設(shè)置RabbitMQ地址
factory.setHost("127.0.0.1");
factory.setUsername("test");
factory.setPassword("123456");
//建立連接
Connection connection = factory.newConnection();
//獲得信道
Channel channel = connection.createChannel();
//聲明隊(duì)列
// queueName 持久存在？ 獨(dú)有？ 自動(dòng)刪除？
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//發(fā)布消息
String message = "Hello World! ";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("發(fā)送了消息：" + message);
//關(guān)閉連接
channel.close();
connection.close();
}
}運(yùn)行一下
發(fā)送成功了 如果我么連接不到RabbitMQ是無法正常發(fā)送的
2 消費(fèi)者
我么要做的就是把剛剛發(fā)送的存儲在隊(duì)列里的消息拿到并打印出來
**
* 描述： 接收消息，并打印，持續(xù)運(yùn)行
*/
public class Recvice {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//創(chuàng)建連接工廠
ConnectionFactory factory = new ConnectionFactory();
//設(shè)置RabbitMQ地址
factory.setHost("127.0.0.1");
factory.setUsername("test");
factory.setPassword("123456");
//建立連接
Connection connection = factory.newConnection();
//獲得信道
Channel channel = connection.createChannel();
//聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//接收消息并消費(fèi) queueName 自動(dòng)簽收 處理消息
channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("收到消息：" + message);
}
});
}
}
可以看到Receive是（打錯(cuò)了，尬）一直運(yùn)行的，我么把發(fā)送的消息改一下再發(fā)送試試
我們之前設(shè)置的是自動(dòng)接收消息們可以看到運(yùn)行時(shí)成功的
去web控制臺也能看到是有hello這個(gè)隊(duì)列的 還有更多的功能就靠你們自己去探索了
Springboot 整合RabbitMQ代碼實(shí)操
1 新建兩個(gè)Spring項(xiàng)目 一個(gè)生產(chǎn)者，一個(gè)消費(fèi)者不需要引入依賴一會(huì)兒手動(dòng)加
主要關(guān)鍵是定義隊(duì)列 queue 定義routingKey
生產(chǎn)者
配置文件
guest是默認(rèn)的用戶只能本機(jī)時(shí)使用
server.port=8080 spring.application.name=producer spring.rabbitmq.addresses=127.0.0.1:5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.virtual-host=/ spring.rabbitmq.connection-timeout=15000
依賴
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>spring-boot-rabbirmq-producer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-boot-rabbirmq-producer</name>
<description>spring-boot-rabbirmq-producer</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>我們只在原基礎(chǔ)上加了一個(gè)依賴
spring-boot-starter-amqp
啟動(dòng)類
@SpringBootApplication
public class SpringBootRabbirmqProducerApplication {
public static void main(String[] args) {
SpringApplication.run(SpringBootRabbirmqProducerApplication.class, args);
}
}發(fā)送消息類
/**
* 描述： 發(fā)送消息
*/
@Component
public class MessageSender {
@Autowired
private AmqpTemplate rabbitmqTemplate;
public void send1() {
String message = "This is message 1, routing key is hello.sayHello";
System.out.println("發(fā)送了："+message);
// 交換機(jī) key 內(nèi)容
this.rabbitmqTemplate.convertAndSend("bootExchange", "hello.sayHello", message);
}
public void send2() {
String message = "This is message 2, routing key is hello.sayNothing";
System.out.println("發(fā)送了："+message);
this.rabbitmqTemplate.convertAndSend("bootExchange", "hello.sayNothing", message);
}
}配置類
/**
* 描述： rabbitmq配置類
*/
@Configuration
public class TopicRabbitConfig {
//定義隊(duì)列 注意類型：import org.springframework.amqp.core.Queue;
@Bean
public Queue queue1() {
return new Queue("queue1");
}
@Bean
public Queue queue2() {
return new Queue("queue2");
}
//交換機(jī)
@Bean
TopicExchange exchange() {
return new TopicExchange("bootExchange");
}
//將隊(duì)列綁定到交換機(jī)
@Bean
Binding bingdingExchangeMessage1(Queue queue1, TopicExchange exchange) {
return BindingBuilder.bind(queue1).to(exchange).with("hello.sayHello");
}
@Bean
Binding bingdingExchangeMessage2(Queue queue2, TopicExchange exchange) {
return BindingBuilder.bind(queue2).to(exchange).with("hello.#");
}
}這里注意第一個(gè)消息的routingkey是跟配置類一樣的hello.sayHello 就代表 我們這個(gè)交換機(jī)是僅能識別hello.sayHello的
第二個(gè)交換機(jī)的routingkey是hello.# 那就意味著只要key是hello.()類型我們都能識別到也就是第一個(gè)和第二個(gè)消息都能識別到
編寫測試類用來發(fā)送消息
@SpringBootTest
class SpringBootRabbirmqProducerApplicationTests {
@Autowired
MessageSender messageSender;
@Test
public void send1(){
messageSender.send1();
}
@Test
public void send2(){
messageSender.send2();
}
}生產(chǎn)者就編寫完成
消費(fèi)者
配置文件，大體一樣，用戶我用的管理者權(quán)限的用戶test 端口號不能一樣
server.port=8081 spring.application.name=consumer spring.rabbitmq.addresses=127.0.0.1:5672 spring.rabbitmq.username=test spring.rabbitmq.password=123456 spring.rabbitmq.virtual-host=/ spring.rabbitmq.connection-timeout=15000
依賴 與生產(chǎn)者一樣只用加一個(gè)
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>spring-boot-rabbitmq-consumer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-boot-rabbitmq-consumer</name>
<description>spring-boot-rabbitmq-consumer</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>啟動(dòng)類
@SpringBootApplication
public class SpringBootRabbitmqConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(SpringBootRabbitmqConsumerApplication.class, args);
}
}消費(fèi)者1 消費(fèi)者一綁的隊(duì)列是queue1 接收消息是要通過交換機(jī)-> 隊(duì)列-> 信道 那就意味著隊(duì)列1中將有hello.sayHello
/**
* 描述： 消費(fèi)者1
*/
@Component
@RabbitListener(queues = "queue1")
public class Receiver1 {
//處理方法
@RabbitHandler
public void process(String message) {
System.out.println("Receiver1: " + message);
}
}消費(fèi)者2
/**
* 描述： 消費(fèi)者2
*/
@Component
@RabbitListener(queues = "queue2")
public class Receiver2 {
@RabbitHandler
public void process(String message) {
System.out.println("Receiver2: " + message);
}
}運(yùn)行結(jié)果
這本身是兩個(gè)獨(dú)立的項(xiàng)目，但是通過RabbitMQ使兩個(gè)項(xiàng)目產(chǎn)生了連接，Springboot完成了對RabbitMQ的整合。
到此這篇關(guān)于Springboot整合RabbitMQ實(shí)現(xiàn)通過RabbitMQ進(jìn)行項(xiàng)目的連接的文章就介紹到這了,更多相關(guān)Springboot整合RabbitMQ內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
java實(shí)現(xiàn)順序結(jié)構(gòu)線性列表的函數(shù)代碼
java實(shí)現(xiàn)順序結(jié)構(gòu)線性列表的函數(shù)代碼。需要的朋友可以過來參考下，希望對大家有所幫助2013-10-10
linux部署出現(xiàn)java文件操作報(bào)錯(cuò):java.io.FileNotFoundException解決辦法
這篇文章主要g介紹了linux部署出現(xiàn)java文件操作報(bào)錯(cuò):java.io.FileNotFoundException解決的相關(guān)資料,這個(gè)錯(cuò)誤通常表示你的Spring Boot應(yīng)用程序無法找到指定的文本文件,需要的朋友可以參考下2023-12-12
這篇文章主要給大家介紹了關(guān)于Java面試之如何獲取客戶端真實(shí)IP的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家學(xué)習(xí)或者使用Java具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09
這篇文章主要介紹了Java使用線程池的優(yōu)勢有哪些，幫助大家更好的理解和掌握java池化技術(shù)，感興趣的朋友可以了解下2020-09-09
這篇文章主要為大家詳細(xì)介紹了Eclipse安裝Free marker插件教程,具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-11-11
分頁技術(shù)原理與實(shí)現(xiàn)之Java+Oracle代碼實(shí)現(xiàn)分頁(二)
這篇文章主要介紹了分頁技術(shù)原理與實(shí)現(xiàn)的第二篇：Java+Oracle代碼實(shí)現(xiàn)分頁，感興趣的小伙伴們可以參考一下2016-06-06
Java中如何使用?byte?數(shù)組作為?Map?的?key
本文將討論在使用HashMap時(shí)，當(dāng)byte數(shù)組作為key時(shí)所遇到的問題及其解決方案，介紹使用String和List這兩種數(shù)據(jù)結(jié)構(gòu)作為臨時(shí)解決方案的方法，感興趣的朋友跟隨小編一起看看吧2023-06-06
這篇文章主要介紹了Java的MoreSuppliers工具類方法解析,MoreSuppliers類是一個(gè)Java工具類,它提供了一些增強(qiáng)的Supplier函數(shù),使得Supplier執(zhí)行的結(jié)果可以被緩存,真正的調(diào)用只執(zhí)行一次,需要的朋友可以參考下2024-01-01
這篇文章主要介紹了java操作Apache druid的實(shí)例代碼,本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-11-11
這篇文章主要介紹了Java 關(guān)系運(yùn)算符詳情及案例實(shí)現(xiàn)，Java 也提供了許多類型的運(yùn)算符，可以根據(jù)需要使用它們來執(zhí)行各種計(jì)算和函數(shù)，包括邏輯、算術(shù)、關(guān)系等。它們根據(jù)它們提供的功能進(jìn)行分類，下面將詳細(xì)介紹該內(nèi)容,需要的朋友可以參考一下2021-12-12