SpringBoot和RabbitMQ集成

红太狼 2022-04-25 01:42 159阅读 0赞

SpringBoot和RabbitMQ的集成：

## 步骤 ##

![AH1g2V.png][]

## 自动配置 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>@ConfigurationProperties(prefix = "spring.rabbitmq") public class RabbitProperties {
 	//地址 private String host = "localhost"; 	//端口 private int port = 5672; 	//账号 private String username; 	//密码 private String password; 	//SSL配置 private final Ssl ssl = new Ssl(); //虚拟主机 private String virtualHost; //地址 private String addresses; 	//请求心跳超时，以秒为单位; 零，没有。 private Integer requestedHeartbeat; 	//Publisher Confirms and Returns机制 private boolean publisherConfirms; private boolean publisherReturns; 	//连接超时时间 private Integer connectionTimeout; 	//缓存 private final Cache cache = new Cache(); 	//监听容器配置 private final Listener listener = new Listener(); private final Template template = new Template(); private List&lt;Address&gt; parsedAddresses; public String getHost() {
 return this.host; } </pre></td> 
 </tr> 
 </tbody> 
</table>

RabbitProperties封装了RabbitMQ发送和接收消息。

RabbitTemplate给RabbitMQ发送和接收消息。

AmqpAdmin,RabbitMQ系统管理功能组件。

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>@Bean 		@ConditionalOnSingleCandidate(ConnectionFactory.class) 		@ConditionalOnMissingBean(RabbitTemplate.class) 		public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
 //生成rabbitTemplate来操作rabbitmq 			RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); 			MessageConverter messageConverter = this.messageConverter.getIfUnique(); //如果messageConverter不为空设置我们自己的messageConverter 			if (messageConverter != null) {
 				rabbitTemplate.setMessageConverter(messageConverter); 			} 			rabbitTemplate.setMandatory(determineMandatoryFlag()); 			RabbitProperties.Template templateProperties = this.properties.getTemplate(); 			RabbitProperties.Retry retryProperties = templateProperties.getRetry(); 			if (retryProperties.isEnabled()) {
 				rabbitTemplate.setRetryTemplate(createRetryTemplate(retryProperties)); 			} 			if (templateProperties.getReceiveTimeout() != null) {
 				rabbitTemplate.setReceiveTimeout(templateProperties.getReceiveTimeout()); 			} 			if (templateProperties.getReplyTimeout() != null) {
 				rabbitTemplate.setReplyTimeout(templateProperties.getReplyTimeout()); 			} 			return rabbitTemplate; 		} 		@Bean 		@ConditionalOnSingleCandidate(ConnectionFactory.class) 		@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic", 				matchIfMissing = true) 		@ConditionalOnMissingBean(AmqpAdmin.class) 		public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
 			return new RabbitAdmin(connectionFactory); 		} 	} </pre></td> 
 </tr> 
 </tbody> 
</table>

## P2P发送 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>package com.hph.amqp; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import java.util.Arrays; import java.util.HashMap; import java.util.Map; @RunWith(SpringRunner.class) @SpringBootTest public class SpringBootAmqpApplicationTests {
 @Autowired RabbitTemplate rabbitTemplate; /** * 单播 P2P */ @Test public void p2p() {
 Map&lt;String, Object&gt; map = new HashMap&lt;&gt;(); map.put("msg","这是第1个消息"); map.put("data", Arrays.asList("Hello Rabitmq",123456, true)); 		//对象默认被序列化以后发送出去 rabbitTemplate.convertAndSend("exchange.direct", "phh.news",map); } } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHdZeP.png][]

这是因为默认使用的是application/x-java-serialized-object的序列化

## 获取消息 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 </pre></td> 
 <td><pre>@Test public void receive() {
 Object o = rabbitTemplate.receiveAndConvert("hph.news"); System.out.println(o.getClass()); System.out.println(o); } </pre></td> 
 </tr> 
 </tbody> 
</table>

## 转为Json ##

由于是RabbitTemplate操作Rabbit的在RabbitTemplate中RabbitTemplate为默认的序列化器

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>private volatile MessageConverter messageConverter = new SimpleMessageConverter(); </pre></td> 
 </tr> 
 </tbody> 
</table>

MessageConverter又一下实现类我们使用的是Jackson2JsonMessageConverter的序列化器

![AHwo34.png][]

在设置我们自己的MessageConverter

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>if (messageConverter != null) {
 				rabbitTemplate.setMessageConverter(messageConverter); 			} </pre></td> 
 </tr> 
 </tbody> 
</table>

再次发送消息

![AHB9iT.png][]

## 自定义发送 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>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 </pre></td> 
 <td><pre>package com.hph.amqp.bean; public class Person {
 private String name; private Integer age; public Person() {
 } public Person(String name, Integer age) {
 this.name = name; this.age = age; } public String getName() {
 return name; } public void setName(String name) {
 this.name = name; } public Integer getAge() {
 return age; } public void setAge(Integer age) {
 this.age = age; } @Override public String toString() {
 return "Person{" + "name='" + name + '\'' + ", age=" + age + '}'; } } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHBqk6.png][]

## 反序列化 ##

![AHcEOe.png][]

## 广播发送 ##

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>@Test public void sendMessages() {
 rabbitTemplate.convertAndSend("exchange.fanout", "hph.news", new Person("清风笑丶",18)); } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHrePK.png][]

## 监听消息队列 ##

启动SpringBoot然后运行sendMessage任务。

![AHc1l8.png][]

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 </pre></td> 
 <td><pre> @RabbitListener(queues = "hph") public void receive02(Message message){
 System.out.println(message.getBody()); System.out.println(message.getMessageProperties()); } } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHcWfx.png][]

消息头信息。

## 管理 ##

在SpringBoot中消息队列的管理使用到了amqpAdmin

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>@ConditionalOnMissingBean(AmqpAdmin.class) 	public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
 		return new RabbitAdmin(connectionFactory); 	} </pre></td> 
 </tr> 
 </tbody> 
</table>

在RabbitAutoConfiguration

![AHgiAs.png][]

!\[

AHg0UA.png\]([https://s2.ax1x.com/2019/04/11/AHg0UA.png][https_s2.ax1x.com_2019_04_11_AHg0UA.png])

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>@Test public void createExchange(){
 amqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange")); System.out.println("创建完成"); } </pre></td> 
 </tr> 
 </tbody> 
</table>

运行该方法。

![AH2Yin.png][]

### 创建exchange ###

### 创建Queue ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 </pre></td> 
 <td><pre>@Test public void createQueue() {
 amqpAdmin.declareQueue(new Queue("amqpadmin.queue", true)); System.out.println("创建队列成功"); } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHRUkd.png][]

### 绑定exchange ###

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 </pre></td> 
 <td><pre>public Binding(String destination, DestinationType destinationType, String exchange, String routingKey, Map&lt;String, Object&gt; arguments) {
 		this.destination = destination; 		this.destinationType = destinationType; 		this.exchange = exchange; 		this.routingKey = routingKey; 		this.arguments = arguments; 	} </pre></td> 
 </tr> 
 </tbody> 
</table>

之前尚未绑定

![AHR5cV.png][]

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>@Test public void bindExchange() {
 amqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE, "amqpadmin.exchange", "amqp.bind", null)); } </pre></td> 
 </tr> 
 </tbody> 
</table>

![AHRqAJ.png][]

绑定成功

[AH1g2V.png]: /images/20220210/60a8e0dec65542ebaa6f4e2c1ed7da19.png
[AHdZeP.png]: /images/20220210/95798c79762d41b7a200c495a9f4e984.png
[AHwo34.png]: /images/20220210/83528e22d2824dbaacd0786d9db511f0.png
[AHB9iT.png]: /images/20220210/9fdeb37838064970b6c800eb94ec781b.png
[AHBqk6.png]: /images/20220210/d91b75200c8844b38c9f14f571e78a16.png
[AHcEOe.png]: /images/20220210/4e3903413c3e4744b4add30ac3db34eb.png
[AHrePK.png]: /images/20220210/065df2feb8b0412c9906494a62ea48f2.png
[AHc1l8.png]: /images/20220210/051f0befa6594a01b1b053d2fde2aebd.png
[AHcWfx.png]: /images/20220210/fec0287487c4422fa24a0f62455f967d.png
[AHgiAs.png]: /images/20220210/c6ba3ec266ed48a29087b36076e4d792.png
[https_s2.ax1x.com_2019_04_11_AHg0UA.png]: https://s2.ax1x.com/2019/04/11/AHg0UA.png
[AH2Yin.png]: /images/20220210/b06fcd968db74325ba538aeb3f891e98.png
[AHRUkd.png]: /images/20220210/83de4a7f9587450a99dae9a9c7cfdfbd.png
[AHR5cV.png]: /images/20220210/ff8dd908adce493ab4141b850de8efa6.png
[AHRqAJ.png]: /images/20220210/4c0732f6410c491f9b23798073c70390.png