Springboot整合RabbitMq，看完这一篇就够了！

前言

RabbiMQ介绍

一、使用场景

RabbitMQ是一个消息中间件，所以最主要的作用就是：信息缓冲区，实现应用程序的异步和解耦。
RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。
AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。详细概念可以参考官方指南 RabbitMQ

二、相关概念

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

• 虚拟主机v-host：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机。
• 交换机：Exchange 用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
• 绑定：也就是交换机需要和队列相绑定，这其中如上图所示

交换机（Exchange）

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

• Direct：direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个 routingkey, 消息的routingkey 匹配时, 才会被交换器投送到绑定的队列中去.
• Topic：按规则转发消息（最灵活）
• Headers：设置header attribute参数类型的交换机
• Fanout：转发消息到所有绑定队列

Direct Exchange

Direct Exchange是RabbitMQ默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

第一个 X - Q1 就有一个 binding key，名字为 orange；X - Q2 就有 2 个 binding key，名字为 black 和 green。当消息中的 路由键 和 这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

Ps：为什么 X 到 Q2 要有 black，green，2个 binding key呢，一个不就行了吗？- 这个主要是因为可能又有 Q3，而Q3只接受 black 的信息，而Q2不仅接受black 的信息，还接受 green 的信息。

Topic Exchange

根据通配符转发消息到队列，在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

• *（星号）可以替代一个单词。
• ＃（hash）可以替换零个或多个单词。

Headers Exchange

headers 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

Fanout Exchange

消息广播的模式，也就是我们的发布订阅模式。Fanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing_key会被忽略。

消息确认

消息消费者如何通知 Rabbit 消息消费成功？

消息通过 ACK 确认是否被正确接收，每个 Message 都要被确认（acknowledged），可以手动去 ACK 或自动 ACK 自动确认会在消息发送给消费者后立即确认，但存在丢失消息的可能，如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息 如果消息已经被处理，但后续代码抛出异常，使用 Spring 进行管理的话消费端业务逻辑会进行回滚，这也同样造成了实际意义的消息丢失 如果手动确认则当消费者调用 ack、nack、reject 几种方法进行确认，手动确认可以在业务失败后进行一些操作，如果消息未被 ACK 则会发送到下一个消费者 如果某个服务忘记 ACK 了，则 RabbitMQ 不会再发送数据给它，因为 RabbitMQ 认为该服务的处理能力有限 ACK 机制还可以起到限流作用，比如在接收到某条消息时休眠几秒钟 消息确认模式有：

• AcknowledgeMode.NONE：自动确认
• AcknowledgeMode.AUTO：根据情况确认
• AcknowledgeMode.MANUAL：手动确认

SpringBoot集成RabbitMQ

1. 配置pom，主要添加spring-boot-starter-amqp支持，springboot基于2.1.4版本
2. <dependency>
3. <groupId>org.springframework.boot</groupId>
4. <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>
6. <dependency>
7. <groupId>org.springframework.boot</groupId>
8. <artifactId>spring-boot-starter-web</artifactId>
9. </dependency>
10. <dependency>
11. <groupId>org.springframework.boot</groupId>
12. <artifactId>spring-boot-starter-test</artifactId>
13. <scope>test</scope>
14. </dependency>
15. <dependency>
16. <groupId>org.projectlombok</groupId>
17. <artifactId>lombok</artifactId>
18. <optional>true</optional>
19. </dependency>

配置springboot的yaml文件
1.

markdown复制代码1. `server:`
2. `servlet:`
3. `context-path:/rabbitmq`
4. `port:9004`
5. `spring:`
6. `application:`
7. `name: rabbitmq`
8. `rabbitmq:`
9. `host: localhost`
10. `virtual-host:/crawl`
11. `username: xxxx`
12. `password: xxx`
13. `port:5672`
14. `# 消息失败返回，比如路由不到队列时触发回调`
15. `publisher-returns:true`
16. `# 消息正确发送确认`
17. `publisher-confirms:true`
18. `template:`
19. `retry:`
20. `enabled:true`
21. `initial-interval:2s`
22. `listener:`
23. `simple:`
24. `# 手动ACK 不开启自动ACK模式,目的是防止报错后未正确处理消息丢失 默认 为 none`
25. `acknowledge-mode: manual`

另外我们还要配置ACK确认回调的配置，通过实现RabbitTemplate.ConfirmCallback接口，消息发送到Broker后触发回调，也就是只能确认是否正确到达Exchange中。

1. import lombok.extern.slf4j.Slf4j;

2. import org.springframework.amqp.rabbit.connection.CorrelationData;

3. import org.springframework.amqp.rabbit.core.RabbitTemplate;

4. import org.springframework.beans.factory.annotation.Autowired;

5. import org.springframework.stereotype.Component;

6. import javax.annotation.PostConstruct;

7. /**

8. * @author lijianqing

9. * @version 1.0

10. * @ClassName RabbitTemplateConfirmCallback

11. * @date 2019/4/23 12:55

12. */

13. @Component

14. @Slf4j

15. public class RabbitTemplateConfirmCallback implements RabbitTemplate.ConfirmCallback {

16. @Autowired

17. private RabbitTemplate rabbitTemplate;

18. @PostConstruct

19. public void init() {

20. //指定 ConfirmCallback

21. kotlin复制代码    `rabbitTemplate.setConfirmCallback(this);`

22. }

23. @Override

24. public void confirm(CorrelationData correlationData, boolean ack, String cause) {

25. c复制代码    `log.info("消息唯一标识：{},确认结果：{},失败原因：{}", correlationData, ack, cause);`

26. }

27. }

消息失败返回，比如路由步到队列就会触发，如果西区奥西发送到交换器成功，但是没有匹配的队列就会触发回调

1. import lombok.extern.slf4j.Slf4j;

2. import org.springframework.amqp.core.Message;

3. import org.springframework.amqp.rabbit.core.RabbitTemplate;

4. import org.springframework.beans.factory.annotation.Autowired;

5. import org.springframework.stereotype.Component;

6. import javax.annotation.PostConstruct;

7. /**

8. * @author lijianqing

9. * @version 1.0

10. * @ClassName RabbitTemplateReturnCallback

11. * @date 2019/4/23 12:55

12. */

13. @Component

14. @Slf4j

15. public class RabbitTemplateReturnCallback implements RabbitTemplate.ReturnCallback {

16. @Autowired

17. private RabbitTemplate rabbitTemplate;

18. @PostConstruct

19. public void init() {

20. //指定 ReturnCallback

21. kotlin复制代码    `rabbitTemplate.setReturnCallback(this);`

22. arduino复制代码    `rabbitTemplate.setMandatory(true);`

23. }

24. @Override

25. public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

26. c复制代码    `log.info("消息主体 message : "` `+ message);`

27. c复制代码    `log.info("消息主体 message : "` `+ replyCode);`

28. c复制代码    `log.info("描述："` `+ replyText);`

29. c复制代码    `log.info("消息使用的交换器 exchange : "` `+ exchange);`

30. c复制代码    `log.info("消息使用的路由键 routing : "` `+ routingKey);`

31. }

32. }

一、简单的开始-简单队列

如下图：

“P”是我们的生产者，“C”是我们的消费者。中间的框是一个队列 - RabbitMQ代表消费者保留的消息缓冲区。

新增SimpleConfig，创建我们要投放的队列：代码如下

1. /**
2. * 队列直接投放
3. * @author lijianqing
4. * @version 1.0
5. * @ClassName SimpleConfig
6. * @date 2019/4/26 15:11
7. */
8. @Configuration
9. public class SimpleConfig {
10. @Bean
11. public Queue simpleQueue() {
12. return new Queue("simple");
13. }
14. }

再分别创建消息发送者与消息接收者：

• 消息发送者

1. import lombok.extern.slf4j.Slf4j;

2. import org.springframework.amqp.rabbit.connection.CorrelationData;

3. import org.springframework.amqp.rabbit.core.RabbitTemplate;

4. import org.springframework.beans.factory.annotation.Autowired;

5. import org.springframework.stereotype.Component;

6. import zero.springboot.study.rabbitmq.model.User;

7. import java.util.UUID;

8. /**

9. * @author lijianqing

10. * @version 1.0

11. * @ClassName HelloSender

12. * @date 2019/4/23 11:22

13. */

14. @Component

15. @Slf4j

16. public class HelloSender {

17. @Autowired

18. private RabbitTemplate rabbitTemplate;

19. public void send() {

20. User user = new User();

21. arduino复制代码    `user.setName("青");`

22. arduino复制代码    `user.setPass("111111");`

23. //发送消息到hello队列

24. c复制代码    `log.info("发送消息：{}", user);`

25. less复制代码    `rabbitTemplate.convertAndSend("hello", user,` `new` `CorrelationData(UUID.randomUUID().toString()));`

26. String msg = "hello qing";

27. c复制代码    `log.info("发送消息：{}", msg);`

28. arduino复制代码    `rabbitTemplate.convertAndSend("simple", msg);`

29. }

30. }

• 消息接收者

1. import com.rabbitmq.client.Channel;

2. import lombok.extern.slf4j.Slf4j;

3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;

4. import org.springframework.amqp.rabbit.annotation.RabbitListener;

5. import org.springframework.amqp.support.AmqpHeaders;

6. import org.springframework.messaging.handler.annotation.Header;

7. import org.springframework.stereotype.Component;

8. import zero.springboot.study.rabbitmq.model.User;

9. import java.io.IOException;

10. /**

11. * 监听hello队列

12. *

13. * @author lijianqing

14. * @version 1.0

15. * @ClassName HelloReceiver

16. * @date 2019/4/23 11:42

17. */

18. @Component

19. @Slf4j

20. @RabbitListener(queues = "simple")

21. public class HelloReceiver {

22. @RabbitHandler

23. public void processUser(User user, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

24. c复制代码    `log.info("收到消息：{}", user);`

25. // 手动ACK

26. try {

27. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

28. go复制代码        `channel.basicAck(tag,` `false);`

29. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

30. // channel.basicNack(deliveryTag, false, false);

31. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

32. // channel.basicReject(deliveryTag,false);

33. } catch (IOException e) {

34. go复制代码        `e.printStackTrace();`

35. }

36. }

37. @RabbitHandler

38. public void processString(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

39. c复制代码    `log.info("收到消息：{}", message);`

40. // 手动ACK

41. try {

42. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

43. go复制代码        `channel.basicAck(tag,` `false);`

44. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

45. // channel.basicNack(deliveryTag, false, false);

46. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

47. // channel.basicReject(deliveryTag,false);

48. } catch (IOException e) {

49. go复制代码        `e.printStackTrace();`

50. }

51. }

52. }

这样就实现了简单的消息发送到指定队列的模式。我们写一个测试类

二、Direct Exchange模式

主要配置我们的Direct Exchange交换机，并且创建队列通过routing key 绑定到交换机上

1. import org.springframework.amqp.core.Binding;
2. import org.springframework.amqp.core.BindingBuilder;
3. import org.springframework.amqp.core.DirectExchange;
4. import org.springframework.amqp.core.Queue;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
7. /**
8. *
9. * @author lijianqing
10. * @version 1.0
11. * @ClassName DirectConfig
12. * @date 2019/4/23 11:15
13. */
14. @Configuration
15. public class DirectConfig {
16. //队列名字
17. public static final String QUEUE_NAME = "direct_name";
18. //交换机名称
19. public static final String EXCHANGE = "zero-exchange";
20. //路由键名称
21. public static final String ROUTING_KEY = "routingKey";
22. @Bean
23. public Queue blueQueue() {
24. return new Queue(QUEUE_NAME, true);
25. }
26. @Bean
27. public DirectExchange defaultExchange() {
28. return new DirectExchange(EXCHANGE);
29. }
30. @Bean
31. public Binding bindingBlue() {
32. return BindingBuilder.bind(blueQueue()).to(defaultExchange()).with(ROUTING_KEY);
33. }
34. }

接下来我们创建生产者与消费者

• 生产者

1. import lombok.extern.slf4j.Slf4j;

2. import org.springframework.amqp.rabbit.connection.CorrelationData;

3. import org.springframework.amqp.rabbit.core.RabbitTemplate;

4. import org.springframework.beans.factory.annotation.Autowired;

5. import org.springframework.stereotype.Component;

6. import zero.springboot.study.rabbitmq.config.DirectConfig;

7. import zero.springboot.study.rabbitmq.model.User;

8. import java.util.UUID;

9. /**

10. * @author lijianqing

11. * @version 1.0

12. * @ClassName HelloSender

13. * @date 2019/4/23 11:22

14. */

15. @Component

16. @Slf4j

17. public class DirectSender {

18. @Autowired

19. private RabbitTemplate rabbitTemplate;

20. public void send() {

21. User user = new User();

22. arduino复制代码    `user.setName("青");`

23. arduino复制代码    `user.setPass("111111");`

24. //发送消息到hello队列

25. c复制代码    `log.info("DirectReceiver发送消息：{}", user);`

26. less复制代码    `rabbitTemplate.convertAndSend(DirectConfig.EXCHANGE,` `DirectConfig.ROUTING_KEY, user,` `new` `CorrelationData(UUID.randomUUID().toString()));`

27. String msg = "hello qing";

28. c复制代码    `log.info("DirectReceiver发送消息：{}", msg);`

29. go复制代码    `rabbitTemplate.convertAndSend(DirectConfig.EXCHANGE,` `DirectConfig.ROUTING_KEY, msg);`

30. }

31. }

• 消费者

1. /**

2. *

3. * @author lijianqing

4. * @version 1.0

5. * @ClassName HelloReceiver

6. * @date 2019/4/23 11:42

7. */

8. @Component

9. @Slf4j

10. @RabbitListener(queues = "direct_name")

11. public class DirectReceiver {

12. @RabbitHandler

13. public void processUser(User user, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

14. c复制代码    `log.info("DirectReceiver收到消息：{}", user);`

15. // 手动ACK

16. try {

17. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

18. go复制代码        `channel.basicAck(tag,` `false);`

19. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

20. // channel.basicNack(deliveryTag, false, false);

21. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

22. // channel.basicReject(deliveryTag,false);

23. } catch (IOException e) {

24. go复制代码        `e.printStackTrace();`

25. }

26. }

27. @RabbitHandler

28. public void processString(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

29. c复制代码    `log.info("收到消息：{}", message);`

30. // 手动ACK

31. try {

32. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

33. go复制代码        `channel.basicAck(tag,` `false);`

34. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

35. // channel.basicNack(deliveryTag, false, false);

36. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

37. // channel.basicReject(deliveryTag,false);

38. } catch (IOException e) {

39. go复制代码        `e.printStackTrace();`

40. }

41. }

42. }

三、Topic Exchange模式

创建队列以及交换机。并通过路由匹配规则将队列与交换机绑定上

1. import org.springframework.amqp.core.Binding;
2. import org.springframework.amqp.core.BindingBuilder;
3. import org.springframework.amqp.core.Queue;
4. import org.springframework.amqp.core.TopicExchange;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
8. /**
9. * queueMessages 匹配topic.#，queueMessage 只匹配 "topic.message"
10. *
11. * @author lijianqing
12. * @version 1.0
13. * @ClassName TopicRabbitConfig
14. * @date 2019/4/23 15:03
15. */
16. @Configuration
17. public class TopicRabbitConfig {
18. final static String message = "topic.message";
19. final static String messages = "topic.messages";
20. @Bean
21. public Queue queueMessage() {
22. return new Queue(TopicRabbitConfig.message);
23. }
24. @Bean
25. public Queue queueMessages() {
26. return new Queue(TopicRabbitConfig.messages);
27. }
28. @Bean
29. TopicExchange exchange() {
30. return new TopicExchange("topicExchange");
31. }
32. @Bean
33. Binding bindingExchangeMessage(@Qualifier("queueMessage") Queue queueMessage, TopicExchange exchange) {
34. return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
35. }
36. @Bean
37. Binding bindingExchangeMessages(@Qualifier("queueMessages") Queue queueMessages, TopicExchange exchange) {
38. return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
39. }
40. }

• 创建生产者

1. import lombok.extern.slf4j.Slf4j;

2. import org.springframework.amqp.rabbit.core.RabbitTemplate;

3. import org.springframework.beans.factory.annotation.Autowired;

4. import org.springframework.stereotype.Component;

5. /**

6. * @author lijianqing

7. * @version 1.0

8. * @ClassName TopicSender

9. * @date 2019/4/23 15:10

10. */

11. @Component

12. @Slf4j

13. public class TopicSender {

14. @Autowired

15. private RabbitTemplate rabbitTemplate;

16. /**

17. * 匹配topic.message，两个队列都会收到

18. */

19. public void send1() {

20. String context = "hi, i am message 1";

21. c复制代码    `log.info("主题发送 : {}"` `, context);`

22. less复制代码    `rabbitTemplate.convertAndSend("topicExchange",` `"topic.message", context);`

23. }

24. /**

25. * 匹配topic.messages

26. */

27. public void send2() {

28. String context = "hi, i am messages 2";

29. c复制代码    `log.info("主题发送 : {}"` `, context);`

30. less复制代码    `rabbitTemplate.convertAndSend("topicExchange",` `"topic.messages", context);`

31. }

32. }

• 创建消费者，这里我们分别创建两个队列的消费者

1. @Component

2. @RabbitListener(queues = "topic.message")

3. @Slf4j

4. public class TopicReceiver {

5. @RabbitHandler

6. public void process(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

7. c复制代码    `log.info("topic.message Receiver1  {}: ", message);`

8. // 手动ACK

9. try {

10. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

11. go复制代码        `channel.basicAck(tag,` `false);`

12. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

13. // channel.basicNack(deliveryTag, false, false);

14. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

15. // channel.basicReject(deliveryTag,false);

16. } catch (IOException e) {

17. go复制代码        `e.printStackTrace();`

18. }

19. }

20. }

第二个消费者

1. @Component

2. @RabbitListener(queues = "topic.messages")

3. @Slf4j

4. public class TopicReceiver2 {

5. @RabbitHandler

6. public void process(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

7. c复制代码    `log.info("topic.messages Receiver2  : {}", message);`

8. // 手动ACK

9. try {

10. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

11. go复制代码        `channel.basicAck(tag,` `false);`

12. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

13. // channel.basicNack(deliveryTag, false, false);

14. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

15. // channel.basicReject(deliveryTag,false);

16. } catch (IOException e) {

17. go复制代码        `e.printStackTrace();`

18. }

19. }

20. }

四、Fanout 模式

也就是发布、订阅。所有绑定在交换机上的队列都会收到消息，发送端指定的routing key的任何字符都会被忽略

配置交换机与队列

1. @Configuration
2. public class FanoutRabbitConfig {
3. @Bean
4. public Queue AMessage() {
5. return new Queue("fanout.A");
6. }
7. @Bean
8. public Queue BMessage() {
9. return new Queue("fanout.B");
10. }
11. @Bean
12. public Queue CMessage() {
13. return new Queue("fanout.C");
14. }
15. @Bean
16. FanoutExchange fanoutExchange() {
17. return new FanoutExchange("fanoutExchange");
18. }
19. @Bean
20. Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
21. return BindingBuilder.bind(AMessage).to(fanoutExchange);
22. }
23. @Bean
24. Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
25. return BindingBuilder.bind(BMessage).to(fanoutExchange);
26. }
27. @Bean
28. Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
29. return BindingBuilder.bind(CMessage).to(fanoutExchange);
30. }
31. }

• 创建发送者

1. @Component

2. @Slf4j

3. public class FanoutSender {

4. @Autowired

5. private RabbitTemplate rabbitTemplate;

6. public void send() {

7. String context = "hi, fanout msg ";

8. csharp复制代码    `rabbitTemplate.convertAndSend("fanoutExchange",` `null, context);`

9. }

10. }

• 创建A、B、C队列消费者

1. @Component

2. @RabbitListener(queues = "fanout.A")

3. @Slf4j

4. public class FanoutReceiverA {

5. @RabbitHandler

6. public void process(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {

7. c复制代码    `log.info("fanout Receiver A  : {}"` `, message);`

8. // 手动ACK

9. try {

10. // //消息确认，代表消费者确认收到当前消息，语义上表示消费者成功处理了当前消息。

11. go复制代码        `channel.basicAck(tag,` `false);`

12. // 代表消费者拒绝一条或者多条消息，第二个参数表示一次是否拒绝多条消息，第三个参数表示是否把当前消息重新入队

13. // channel.basicNack(deliveryTag, false, false);

14. // 代表消费者拒绝当前消息，第二个参数表示是否把当前消息重新入队

15. // channel.basicReject(deliveryTag,false);

16. } catch (IOException e) {

17. go复制代码        `e.printStackTrace();`

18. }

19. }

20. }

剩下的B、C就不重复贴了。

单元测试

1. import org.junit.Test;

2. import org.junit.runner.RunWith;

3. import org.springframework.beans.factory.annotation.Autowired;

4. import org.springframework.boot.test.context.SpringBootTest;

5. import org.springframework.test.context.junit4.SpringRunner;

6. import zero.springboot.study.rabbitmq.direct.DirectSender;

7. import zero.springboot.study.rabbitmq.fanout.FanoutSender;

8. import zero.springboot.study.rabbitmq.simple.HelloSender;

9. import zero.springboot.study.rabbitmq.topic.TopicSender;

10. @RunWith(SpringRunner.class)

11. @SpringBootTest(classes = RabbitmqApplication.class)

12. public class RabbitmqApplicationTests {

13. @Autowired

14. private DirectSender directSender;

15. @Autowired

16. private TopicSender topicSender;

17. @Autowired

18. private FanoutSender fanoutSender;

19. @Autowired

20. private HelloSender helloSender;

21. @Test

22. public void testDirect() {

23. go复制代码    `directSender.send();`

24. }

25. @Test

26. public void topic1() {

27. go复制代码    `topicSender.send1();`

28. }

29. @Test

30. public void topic2() {

31. go复制代码    `topicSender.send2();`

32. }

33. @Test

34. public void testFanout() {

35. go复制代码    `fanoutSender.send();`

36. }

37. @Test

38. public void testSimple() {

39. go复制代码    `helloSender.send();`

40. }

41. }

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有