AMQP簡介
AMQP敬尺，即 Advanced Message Queuing Protocol枚尼，高級消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)砂吞，為面向消息的中間件設(shè)計(jì)署恍。消息中間件主要用于組件之間的解耦和通訊。
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)中存儲(chǔ)轉(zhuǎn)發(fā)消息为朋，具有很高的易用性和可用性。
在docker環(huán)境部署RabbitMQ
RabbitMQ是用 Erlang 編寫的厚脉，直接部署的話需要先部署 Erlang 環(huán)境习寸，比較麻煩。在 docker 環(huán)境下部署就比較簡單了傻工，直接使用rabbitmq官方提供的鏡像即可霞溪。
登錄 docker 節(jié)點(diǎn)，運(yùn)行 docker pull rabbitmq:management中捆，這里使用的是帶 web 管理插件的鏡像鸯匹。
啟動(dòng)容器：
docker run -d --name rabbitmq --publish 5671:5671 \
--publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 \
rabbitmq:management
容器啟動(dòng)之后就可以訪問web 管理端了 http://宿主機(jī)IP:15672，默認(rèn)創(chuàng)建了一個(gè) guest 用戶泄伪，密碼也是 guest殴蓬。
AMQP協(xié)議中的幾個(gè)重要概念
- Queue 是RabbitMQ的內(nèi)部對象，用于存儲(chǔ)消息蟋滴。RabbitMQ中的消息只能存儲(chǔ)在 Queue 中染厅，消費(fèi)者從 Queue 中獲取消息并消費(fèi)。
- Exchange 生產(chǎn)者將消息發(fā)送到 Exchange脓杉，由 Exchange 根據(jù)一定的規(guī)則將消息路由到一個(gè)或多個(gè) Queue 中（或者丟棄）糟秘。
- Binding RabbitMQ中通過 Binding 將 Exchange 與 Queue 關(guān)聯(lián)起來。
- Binding key 在綁定（Binding） Exchange 與 Queue 的同時(shí)球散，一般會(huì)指定一個(gè) binding key尿赚。
- Routing key 生產(chǎn)者在將消息發(fā)送給 Exchange 的時(shí)候，一般會(huì)指定一個(gè) routing key蕉堰，來指定這個(gè)消息的路由規(guī)則凌净。 Exchange 會(huì)根據(jù) routing key 和 Exchange Type 以及 Binding key 的匹配情況來決定把消息路由到哪個(gè) Queue。
-
Exchange Types RabbitMQ常用的Exchange Type有 fanout屋讶、 direct冰寻、 topic、 headers 這四種皿渗。
- fanout 這種類型的Exchange路由規(guī)則非常簡單斩芭，它會(huì)把所有發(fā)送到該Exchange的消息路由到所有與它綁定的Queue中轻腺，這時(shí) Routing key 不起作用。
- direct 這種類型的Exchange路由規(guī)則也很簡單划乖，它會(huì)把消息路由到那些 binding key 與 routing key完全匹配的Queue中贬养。
-
topic 這種類型的Exchange的路由規(guī)則支持 binding key 和 routing key 的模糊匹配，會(huì)把消息路由到滿足條件的Queue琴庵。 binding key 中可以存在兩種特殊字符
*與#误算，用于做模糊匹配，其中*用于匹配一個(gè)單詞迷殿，#用于匹配多個(gè)單詞（可以是零個(gè)）儿礼，單詞以.為分隔符。 - headers 這種類型的Exchange不依賴于 routing key 與 binding key 的匹配規(guī)則來路由消息庆寺，而是根據(jù)發(fā)送的消息內(nèi)容中的 headers 屬性進(jìn)行匹配蚊夫。
使用Spring AMQP收發(fā)消息
新建一個(gè)maven工程，修改pom.xml引入 spring amqp 依賴:
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.2.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
在 java 目錄中創(chuàng)建一個(gè)包 demo 懦尝，在包中創(chuàng)建啟動(dòng)入口 SpringAmqpApplication.java
public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(SpringAmqpApplication.class, args);
Sender sender = context.getBean("sender", Sender.class);
sender.sendMsg("測試Spring AMQP發(fā)送消息");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
context.close();
}
@Bean
CachingConnectionFactory myConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setHost("10.47.160.238");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
return connectionFactory;
}
@Bean
Exchange myExchange() {
return ExchangeBuilder.topicExchange("test.topic").durable().build();
}
@Bean
Queue myQueue() {
return QueueBuilder.durable("myQueue").build();
}
@Bean
public Binding myExchangeBinding(@Qualifier("myExchange") Exchange topicExchange,
@Qualifier("myQueue") Queue queue) {
return BindingBuilder.bind(queue).to(topicExchange).with("test.#").noargs();
}
@Bean
public RabbitTemplate myExchangeTemplate(CachingConnectionFactory myConnectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(myConnectionFactory);
rabbitTemplate.setExchange("test.topic");
rabbitTemplate.setRoutingKey("test.abc.123");
return rabbitTemplate;
}
在 demo 包下創(chuàng)建 Sender.java
@Component
public class Sender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMsg(String content) {
rabbitTemplate.convertAndSend(content);
System.out.println("發(fā)送消息: '" + content + "'");
}
}
在 demo 包下創(chuàng)建Receiver.java
@Component
public class Receiver {
@RabbitListener(queues = "myQueue")
public void processMessage(Message message) {
byte[] body = message.getBody();
System.out.println("收到消息: '" + new String(body) + "'");
}
}
回頭看下代碼这橙，在 SpringAmqpApplication.java 中創(chuàng)建了程序啟動(dòng)入口 main 方法，為了有時(shí)間把收到的消息打印出來导披，讓主線程 sleep 了1秒屈扎，配置了幾個(gè)和 RabbitMQ 相關(guān)的重要配置：
- RabbitMQ 的連接 CachingConnectionFactory
- 創(chuàng)建了一個(gè)名為 test.topic 并且類型為 topic 的 Exchange
- 創(chuàng)建了一個(gè)名為 myQueue 的 Queue
- 把上邊創(chuàng)建的 Queue 和 Exchange 進(jìn)行了綁定，并指定 binding key 為 test.#
- 最后還配置了一個(gè)spring封裝的模板工具類 rabbitTemplate撩匕，指定了 Exchange 和 Routing key鹰晨，用這個(gè) rabbitTemplate 發(fā)送消息， 會(huì)把消息發(fā)送到名為 test.topic 的 Exchange止毕，并且?guī)в?Routing key test.abc.123
Sender.java 和 Receiver.java 的代碼就比較簡單了模蜡，在 sendMsg 方法中使用 rabbitTemplate 發(fā)送消息。在 processMessage 方法上加了一個(gè) @RabbitListener(queues = "myQueue") 注解扁凛，指定從 myQueue 這個(gè)隊(duì)列中獲取消息忍疾。
運(yùn)行 main 方法啟動(dòng)工程，可以看到控制臺(tái)打印出了發(fā)送的消息和接收的消息谨朝。
最后
RabbitMQ在 spring cloud 中做為消息總線卤妒，負(fù)責(zé)傳遞和分發(fā)系統(tǒng)消息，是非常重要的一個(gè)角色字币，spring cloud bus 動(dòng)態(tài)加載配置就是使用消息總線则披，把重新拉去配置的消息分發(fā)到各個(gè)連接到消息總線的微服務(wù)。在 spring cloud steam 的消息驅(qū)動(dòng)模型中同樣使用了RabbitMQ洗出。
不僅如此瞧甩，RabbitMQ本身也是一個(gè)非常高效的消息服務(wù)器苍柏，可以用在服務(wù)之間異步調(diào)用揪罕，以及RPC遠(yuǎn)程調(diào)用（在消息頭中增加 Reply Queue 來監(jiān)聽調(diào)用返回信息）。
