this direct 還需要 添加屬性 創建 還需 topic start routing

前言

本篇主要講述Spring Boot與RabbitMQ的整合，內容非常簡單，純API的調用操作。 操作之間需要加入依賴Jar

[html] view plain copy

1. <dependency>
2. <groupId>org.springframework.boot</groupId>
3. <artifactId>spring-boot-starter-amqp</artifactId>
4. lt;/dependency>

消息生產者

不論是創建消息消費者或生產者都需要ConnectionFactory

ConnectionFactory配置

創建AmqpConfig文件AmqpConfig.java(後期的配置都在該文件中) [html] view plain copy

1. @Configuration
2. public class AmqpConfig {
4. public static final String EXCHANGE = "spring-boot-exchange";
5. public static final String ROUTINGKEY = "spring-boot-routingKey";
7. @Bean
8. public ConnectionFactory connectionFactory() {
9. CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
10. connectionFactory.setAddresses("127.0.0.1:5672");
11. connectionFactory.setUsername("guest");
12. connectionFactory.setPassword("guest");
13. connectionFactory.setVirtualHost("/");
14. connectionFactory.setPublisherConfirms(true); //必須要設置
15. return connectionFactory;
16. }
17. }

這裏需要顯示調用 [html] view plain copy

1. connectionFactory.setPublisherConfirms(true);

才能進行消息的回調。

RabbitTemplate

通過使用RabbitTemplate來對開發者提供API操作 [html] view plain copy

1. @Bean
2. @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
3. //必須是prototype類型
4. public RabbitTemplate rabbitTemplate() {
5. RabbitTemplate template = new RabbitTemplate(connectionFactory());
6. return template;
7. }

這裏設置為原型，具體的原因在後面會講到 在發送消息時通過調用RabbitTemplate中的如下方法 [html] view plain copy

1. public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData)

• exchange:交換機名稱

• routingKey:路由關鍵字

• object:發送的消息內容

• correlationData:消息ID

因此生產者代碼詳單簡潔

Send.java

[html] view plain copy

1. @Component
2. public class Send {
4. private RabbitTemplate rabbitTemplate;
6. /**
7. * 構造方法註入
8. */
9. @Autowired
10. public Send(RabbitTemplate rabbitTemplate) {
11. this.rabbitTemplate = rabbitTemplate;
12. }
14. public void sendMsg(String content) {
15. CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
16. rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);
17. }
20. }

如果需要在生產者需要消息發送後的回調，需要對rabbitTemplate設置ConfirmCallback對象，由於不同的生產者需要對應不同的ConfirmCallback，如果rabbitTemplate設置為單例bean，則所有的rabbitTemplate

實際的ConfirmCallback為最後一次申明的ConfirmCallback。

下面給出完整的生產者代碼：

[html] view plain copy

1. package com.lkl.springboot.amqp;
3. import java.util.UUID;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.amqp.rabbit.support.CorrelationData;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.stereotype.Component;
10. /**
11. * 消息生產者
12. *
13. * @author liaokailin
14. * @version $Id: Send.java, v 0.1 2015年11月01日 下午4:22:25 liaokailin Exp $
15. */
16. @Component
17. public class Send implements RabbitTemplate.ConfirmCallback {
19. private RabbitTemplate rabbitTemplate;
21. /**
22. * 構造方法註入
23. */
24. @Autowired
25. public Send(RabbitTemplate rabbitTemplate) {
26. this.rabbitTemplate = rabbitTemplate;
27. rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果為單例的話，那回調就是最後設置的內容
28. }
30. public void sendMsg(String content) {
31. CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
32. rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);
33. }
35. /**
36. * 回調
37. */
38. @Override
39. public void confirm(CorrelationData correlationData, boolean ack, String cause) {
40. System.out.println(" 回調id:" + correlationData);
41. if (ack) {
42. System.out.println("消息成功消費");
43. } else {
44. System.out.println("消息消費失敗:" + cause);
45. }
46. }
48. }

消息消費者

消費者負責申明交換機(生產者也可以申明)、隊列、兩者的綁定操作。

交換機

[html] view plain copy

1. /**
2. * 針對消費者配置
3. FanoutExchange: 將消息分發到所有的綁定隊列，無routingkey的概念
4. HeadersExchange ：通過添加屬性key-value匹配
5. DirectExchange:按照routingkey分發到指定隊列
6. TopicExchange:多關鍵字匹配
7. */
8. @Bean
9. public DirectExchange defaultExchange() {
10. return new DirectExchange(EXCHANGE);
11. }

在Spring Boot中交換機繼承AbstractExchange類

隊列

[html] view plain copy

1. @Bean
2. public Queue queue() {
3. return new Queue("spring-boot-queue", true); //隊列持久
5. }

綁定

[html] view plain copy

1. @Bean
2. public Binding binding() {
3. return BindingBuilder.bind(queue()).to(defaultExchange()).with(AmqpConfig.ROUTINGKEY);
4. }

完成以上工作後，在spring boot中通過消息監聽容器實現消息的監聽，在消息到來時執行回調操作。

消息消費

[html] view plain copy

1. @Bean
2. public SimpleMessageListenerContainer messageContainer() {
3. SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
4. container.setQueues(queue());
5. container.setExposeListenerChannel(true);
6. container.setMaxConcurrentConsumers(1);
7. container.setConcurrentConsumers(1);
8. container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設置確認模式手工確認
9. container.setMessageListener(new ChannelAwareMessageListener() {
11. @Override
12. public void onMessage(Message message, Channel channel) throws Exception {
13. byte[] body = message.getBody();
14. System.out.println("receive msg : " + new String(body));
15. channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //確認消息成功消費
16. }
17. });
18. return container;
19. }

下面給出完整的配置文件： [html] view plain copy

1. package com.lkl.springboot.amqp;
3. import org.springframework.amqp.core.AcknowledgeMode;
4. import org.springframework.amqp.core.Binding;
5. import org.springframework.amqp.core.BindingBuilder;
6. import org.springframework.amqp.core.DirectExchange;
7. import org.springframework.amqp.core.Message;
8. import org.springframework.amqp.core.Queue;
9. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
10. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
11. import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
12. import org.springframework.amqp.rabbit.core.RabbitTemplate;
13. import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
14. import org.springframework.beans.factory.config.ConfigurableBeanFactory;
15. import org.springframework.context.annotation.Bean;
16. import org.springframework.context.annotation.Configuration;
17. import org.springframework.context.annotation.Scope;
19. import com.rabbitmq.client.Channel;
21. /**
22. * Qmqp Rabbitmq
23. *
24. * http://docs.spring.io/spring-amqp/docs/1.4.5.RELEASE/reference/html/
25. *
26. * @author lkl
27. * @version $Id: AmqpConfig.java, v 0.1 2015年11月01日 下午2:05:37 lkl Exp $
28. */
30. @Configuration
31. public class AmqpConfig {
33. public static final String EXCHANGE = "spring-boot-exchange";
34. public static final String ROUTINGKEY = "spring-boot-routingKey";
36. @Bean
37. public ConnectionFactory connectionFactory() {
38. CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
39. connectionFactory.setAddresses("127.0.0.1:5672");
40. connectionFactory.setUsername("guest");
41. connectionFactory.setPassword("guest");
42. connectionFactory.setVirtualHost("/");
43. connectionFactory.setPublisherConfirms(true); //必須要設置
44. return connectionFactory;
45. }
47. @Bean
48. @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
49. //必須是prototype類型
50. public RabbitTemplate rabbitTemplate() {
51. RabbitTemplate template = new RabbitTemplate(connectionFactory());
52. return template;
53. }
55. /**
56. * 針對消費者配置
57. * 1. 設置交換機類型
58. * 2. 將隊列綁定到交換機
59. *
60. *
61. FanoutExchange: 將消息分發到所有的綁定隊列，無routingkey的概念
62. HeadersExchange ：通過添加屬性key-value匹配
63. DirectExchange:按照routingkey分發到指定隊列
64. TopicExchange:多關鍵字匹配
65. */
66. @Bean
67. public DirectExchange defaultExchange() {
68. return new DirectExchange(EXCHANGE);
69. }
71. @Bean
72. public Queue queue() {
73. return new Queue("spring-boot-queue", true); //隊列持久
75. }
77. @Bean
78. public Binding binding() {
79. return BindingBuilder.bind(queue()).to(defaultExchange()).with(AmqpConfig.ROUTINGKEY);
80. }
82. @Bean
83. public SimpleMessageListenerContainer messageContainer() {
84. SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
85. container.setQueues(queue());
86. container.setExposeListenerChannel(true);
87. container.setMaxConcurrentConsumers(1);
88. container.setConcurrentConsumers(1);
89. container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設置確認模式手工確認
90. container.setMessageListener(new ChannelAwareMessageListener() {
92. @Override
93. public void onMessage(Message message, Channel channel) throws Exception {
94. byte[] body = message.getBody();
95. System.out.println("receive msg : " + new String(body));
96. channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //確認消息成功消費
97. }
98. });
99. return container;
100. }
102. }

以上完成 Spring Boot與RabbitMQ的整合

自動配置

在Spring Boot中實現了RabbitMQ的自動配置，在配置文件中添加如下配置信息 [html] view plain copy

1. spring.rabbitmq.host=localhost
2. spring.rabbitmq.port=5672
3. spring.rabbitmq.username=test
4. spring.rabbitmq.password=test
5. spring.rabbitmq.virtualHost=test

後會自動創建ConnectionFactory以及RabbitTemplate對應Bean，為什麽上面我們還需要手動什麽呢? 自動創建的ConnectionFactory無法完成事件的回調，即沒有設置下面的代碼 [html] view plain copy

1. connectionFactory.setPublisherConfirms(true);

具體分析見後續文章的源碼解讀.

spring boot實戰(第十二篇)整合RabbitMQ