RabbitMQ作為AMQP的代表性產(chǎn)品，在項(xiàng)目中大量使用。結(jié)合現(xiàn)在主流的spring boot，極大簡(jiǎn)化了開(kāi)發(fā)過(guò)程中所涉及到的消息通信問(wèn)題。
首先正確的安裝RabbitMQ及運(yùn)行正常。
RabbitMQ需啊erlang環(huán)境，所以首先安裝對(duì)應(yīng)版本的erlang，可在RabbitMQ官網(wǎng)下載
|
1
|
# rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm |
使用yum安裝RabbitMQ,避免缺少依賴包引起的安裝失敗
|
1
|
# yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm |
啟動(dòng)RabbitMQ
|
1
|
# /sbin/service rabbitmq-server start |
由于RabbitMQ默認(rèn)提供的guest用戶只能本地訪問(wèn)，所以額外創(chuàng)建用戶用于測(cè)試
|
1
2
|
# /sbin/rabbitmqctl add_user test test123用戶名：test，密碼：test123 |
開(kāi)啟web管理插件
|
1
|
# rabbitmq-plugins enable rabbitmq_management |
并使用之前創(chuàng)建的用戶登錄,并設(shè)置該用戶為administrator,虛擬主機(jī)地址為/
spring boot 引入相關(guān)依賴
|
1
2
3
4
5
6
|
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency></dependencies> |
消息生產(chǎn)者
application.properties添加一下配置
|
1
2
3
4
5
6
7
|
spring.rabbitmq.host=192.168.1.107spring.rabbitmq.port=5672spring.rabbitmq.username=testspring.rabbitmq.password=test123spring.rabbitmq.publisher-confirms=truespring.rabbitmq.publisher-returns=truespring.rabbitmq.template.mandatory=true |
spring boot配置類，作用為指定隊(duì)列，交換器類型及綁定操作
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.Queue;import org.springframework.amqp.core.TopicExchange;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class RabbitConfig { //聲明隊(duì)列 @Bean public Queue queue1() { return new Queue("hello.queue1", true); // true表示持久化該隊(duì)列 } @Bean public Queue queue2() { return new Queue("hello.queue2", true); } //聲明交互器 @Bean TopicExchange topicExchange() { return new TopicExchange("topicExchange"); } //綁定 @Bean public Binding binding1() { return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1"); } @Bean public Binding binding2() { return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#"); }} |
共聲明了2個(gè)隊(duì)列，分別是hello.queue1，hello.queue2，交換器類型為TopicExchange,并與hello.queue1，hello.queue2隊(duì)列分別綁定。
生產(chǎn)者類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
import java.util.UUID;import javax.annotation.PostConstruct;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;import org.springframework.amqp.rabbit.support.CorrelationData;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback { @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init() { rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnCallback(this); } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { System.out.println("消息發(fā)送成功:" + correlationData); } else { System.out.println("消息發(fā)送失敗:" + cause); } } @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println(message.getMessageProperties().getCorrelationIdString() + " 發(fā)送失敗"); } //發(fā)送消息，不需要實(shí)現(xiàn)任何接口，供外部調(diào)用。 public void send(String msg){ CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString()); System.out.println("開(kāi)始發(fā)送消息 : " + msg.toLowerCase()); String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString(); System.out.println("結(jié)束發(fā)送消息 : " + msg.toLowerCase()); System.out.println("消費(fèi)者響應(yīng) : " + response + " 消息處理完成"); }} |
要點(diǎn)：
1.注入RabbitTemplate
2.實(shí)現(xiàn)RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口（非必須）。
ConfirmCallback接口用于實(shí)現(xiàn)消息發(fā)送到RabbitMQ交換器后接收ack回調(diào)。ReturnCallback接口用于實(shí)現(xiàn)消息發(fā)送到RabbitMQ交換器，但無(wú)相應(yīng)隊(duì)列與交換器綁定時(shí)的回調(diào)。
3.實(shí)現(xiàn)消息發(fā)送方法。調(diào)用rabbitTemplate相應(yīng)的方法即可。rabbitTemplate常用發(fā)送方法有
|
1
2
3
|
rabbitTemplate.send(message); //發(fā)消息，參數(shù)類型為org.springframework.amqp.core.MessagerabbitTemplate.convertAndSend(object); //轉(zhuǎn)換并發(fā)送消息。 將參數(shù)對(duì)象轉(zhuǎn)換為org.springframework.amqp.core.Message后發(fā)送rabbitTemplate.convertSendAndReceive(message) //轉(zhuǎn)換并發(fā)送消息,且等待消息者返回響應(yīng)消息。 |
針對(duì)業(yè)務(wù)場(chǎng)景選擇合適的消息發(fā)送方式即可。
消息消費(fèi)者
application.properties添加一下配置
|
1
2
3
4
5
6
7
|
spring.rabbitmq.host=192.168.1.107spring.rabbitmq.port=5672spring.rabbitmq.username=testspring.rabbitmq.password=test123spring.rabbitmq.listener.concurrency=2 //最小消息監(jiān)聽(tīng)線程數(shù)spring.rabbitmq.listener.max-concurrency=2 //最大消息監(jiān)聽(tīng)線程數(shù) |
消費(fèi)者類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class Receiver { @RabbitListener(queues = "hello.queue1") public String processMessage1(String msg) { System.out.println(Thread.currentThread().getName() + " 接收到來(lái)自hello.queue1隊(duì)列的消息：" + msg); return msg.toUpperCase(); } @RabbitListener(queues = "hello.queue2") public void processMessage2(String msg) { System.out.println(Thread.currentThread().getName() + " 接收到來(lái)自hello.queue2隊(duì)列的消息：" + msg); }} |
由于定義了2個(gè)隊(duì)列，所以分別定義不同的監(jiān)聽(tīng)器監(jiān)聽(tīng)不同的隊(duì)列。由于最小消息監(jiān)聽(tīng)線程數(shù)和最大消息監(jiān)聽(tīng)線程數(shù)都是2，所以每個(gè)監(jiān)聽(tīng)器各有2個(gè)線程實(shí)現(xiàn)監(jiān)聽(tīng)功能。
要點(diǎn):
1.監(jiān)聽(tīng)器參數(shù)類型與消息實(shí)際類型匹配。在生產(chǎn)者中發(fā)送的消息實(shí)際類型是String，所以這里監(jiān)聽(tīng)器參數(shù)類型也是String。
2.如果監(jiān)聽(tīng)器需要有響應(yīng)返回給生產(chǎn)者，直接在監(jiān)聽(tīng)方法中return即可。
運(yùn)行測(cè)試
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
import java.util.Date;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import com.sam.demo.rabbitmq.Application;import com.sam.demo.rabbitmq.sender.Sender;@RunWith(value=SpringJUnit4ClassRunner.class)@SpringBootTest(classes = Application.class)public class RabbitTests { @Autowired private Sender sender; @Test public void sendTest() throws Exception { while(true){ String msg = new Date().toString(); sender.send(msg); Thread.sleep(1000); } }} |
輸出:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
開(kāi)始發(fā)送消息 : wed mar 29 23:20:52 cst 2017SimpleAsyncTaskExecutor-1 接收到來(lái)自hello.queue2隊(duì)列的消息：Wed Mar 29 23:20:52 CST 2017SimpleAsyncTaskExecutor-2 接收到來(lái)自hello.queue1隊(duì)列的消息：Wed Mar 29 23:20:52 CST 2017結(jié)束發(fā)送消息 : wed mar 29 23:20:52 cst 2017消費(fèi)者響應(yīng) : WED MAR 29 23:20:52 CST 2017 消息處理完成------------------------------------------------消息發(fā)送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886]開(kāi)始發(fā)送消息 : wed mar 29 23:20:53 cst 2017SimpleAsyncTaskExecutor-1 接收到來(lái)自hello.queue1隊(duì)列的消息：Wed Mar 29 23:20:53 CST 2017SimpleAsyncTaskExecutor-2 接收到來(lái)自hello.queue2隊(duì)列的消息：Wed Mar 29 23:20:53 CST 2017結(jié)束發(fā)送消息 : wed mar 29 23:20:53 cst 2017消費(fèi)者響應(yīng) : WED MAR 29 23:20:53 CST 2017 消息處理完成------------------------------------------------消息發(fā)送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34]開(kāi)始發(fā)送消息 : wed mar 29 23:20:54 cst 2017SimpleAsyncTaskExecutor-2 接收到來(lái)自hello.queue1隊(duì)列的消息：Wed Mar 29 23:20:54 CST 2017SimpleAsyncTaskExecutor-1 接收到來(lái)自hello.queue2隊(duì)列的消息：Wed Mar 29 23:20:54 CST 2017結(jié)束發(fā)送消息 : wed mar 29 23:20:54 cst 2017消費(fèi)者響應(yīng) : WED MAR 29 23:20:54 CST 2017 消息處理完成------------------------------------------------ |
如果需要使用的其他的交換器類型，spring中都已提供實(shí)現(xiàn)，所有的交換器均實(shí)現(xiàn)org.springframework.amqp.core.AbstractExchange接口。
常用交換器類型如下：
Direct(DirectExchange)：direct 類型的行為是"先匹配, 再投送". 即在綁定時(shí)設(shè)定一個(gè) routing_key, 消息的routing_key完全匹配時(shí), 才會(huì)被交換器投送到綁定的隊(duì)列中去。
Topic(TopicExchange)：按規(guī)則轉(zhuǎn)發(fā)消息（最靈活）。
Headers(HeadersExchange)：設(shè)置header attribute參數(shù)類型的交換機(jī)。
Fanout(FanoutExchange)：轉(zhuǎn)發(fā)消息到所有綁定隊(duì)列。
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持服務(wù)器之家。
原文鏈接：http://www.jianshu.com/p/e1258c004314#
