介绍

消息队列通常有三个概念

  • 发送消息（生产者）
  • 队列
  • 接收消息（消费者）。

RabbitMQ在这个基本概念之上，多做了一层抽象，在发送消息和队列之间，加入了交换机。这样发送消息和队列就没有直接关系，而是通过交换机来做转发，交换机会根据分发策略把消息转给队列。

RabbitMQ比较重要的几个概念：

  • 虚拟主机：RabbitMQ支持权限控制，但是最小控制粒度为虚拟主机。一个虚拟主机可以包含多个交换机、队列、绑定。
  • 交换机：RabbitMQ分发器，根据不同的策略将消息分发到相关的队列。
  • 队列：缓存消息的容器。
  • 绑定：设置交换机与队列的关系。

黄色的圈圈就是我们的消息推送服务，将消息推送到 中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。

image.png

队列

位于org.springframework.amqp.core这个包
当声明一个队列的时候，它会自动绑定到默认交换机上，并以队列名字作为路由键。
队列构造器参数

  • name 队列名字
  • durable=”true” ，持久化 rabbitmq重启的时候不需要创建新的队列，默认是true
  • exclusive 表示该消息队列是否只在当前connection生效,默认是false
  • autoDelete 表示消息队列没有在使用时将被自动删除 默认是false

交换机

服务器必须实现Direct类型交换机，包含一个空白字符串名称的默认交换器。
交换机构造器参数

  • durable=”true” ，rabbitmq重启的时候不需要创建新的交换机
  • auto-delete 表示交换机没有在使用时将被自动删除 默认是false

交换机类型

  • DirectExchange交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
  • topic交换器，采用模糊匹配路由键的原则进行转发消息到队列中
  • fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。

    DirectExchange 直连型交换机

    直连型交换机，根据消息携带的路由键将消息投递给对应队列。

大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。

  • 配置交换机、队列 ```java import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;

@Configuration public class DirectExchangeConfig {

  1. @Bean(name = "queueDirect")
  2. public Queue directQueue(){
  3.     return new Queue("queue-direct");
  4. }
  6. @Bean(name = "directExchange")
  7. public DirectExchange directExchange() {
  8.     return new DirectExchange("directExchange", true, false);
  9. }
  11. @Bean
  12. public Binding bindingDirect(Queue queueDirect, DirectExchange directExchange) {
  13.     return BindingBuilder.bind(queueDirect).to(directExchange).with("routingKey-direct");
  14. }

}

  2. - 生产者
  3. ```java
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  10. @Component
  11. public class Producter {
  12.     private Logger logger = LoggerFactory.getLogger(Producter.class);
  13.     @Autowired
  14.     private RabbitTemplate rabbitTemplate;
  15.     public void direct(String message){
  16.         logger.info("product direct:" + message);
  17.         rabbitTemplate.convertAndSend("directExchange","routingKey-direct",message);
  18.     }
  19. }
  • 消费者 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Consumer { private Logger logger = LoggerFactory.getLogger(Consumer.class);

  1. @RabbitHandler
  2. @RabbitListener(queues = "queue-direct")
  3. public void direct(String message){
  4.     logger.info("consume direct:"+message);
  5. }

}

  2. - 测试
  3. ```java
  4. @RestController
  5. @RequestMapping(value = "mq-test")
  6. public class MQTest {
  7.     private Logger logger = LoggerFactory.getLogger(MQTest.class);
  9.     @Autowired
  10.     private Producter producter;
  12.     @GetMapping
  13.     public void test(){
  14.         producter.direct("direct message");
  15.     }
  16. }

image.png

TopicExchange 主题交换机

TopicExchange 主题交换机，采用模糊匹配路由键的原则进行转发消息到队列中，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
简单地介绍下规则：

  • * (星号) 用来表示一个单词 (必须出现的)
  • # (井号) 用来表示任意数量（零个或多个）单词

通配符的绑定键是跟队列进行绑定的，举个小例子
队列Q1 绑定键为 .TT. 队列Q2绑定键为 TT.#
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；

主题交换机是非常强大的，理由有二：

  • 当一个队列的绑定键为 “#”（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
  • 当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。

所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。

  • 配置交换机、队列 ```java 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;

// topic交换器 采用模糊匹配路由键的原则进行转发消息到队列中 @Configuration public class TopicExchangeConfig { //topic模式 public static final String TOPIC_QUEUE_FIRST = “topicQueueFirst”; public static final String TOPIC_QUEUE_SECOND = “topicQueueSecond”; //路由key public static final String ROUTING_KEY_TOPIC_FIRST = “routingKey-topic-first.*”; public static final String ROUTING_KEY_TOPIC_SECOND = “routingKey-topic-second.#”; // topic交换机 public static final String TOPIC_EXCHANGE = “topicExchange”; @Bean(name = “firstTopicQueue”) public Queue firstTopicQueue() { return new Queue(TOPIC_QUEUE_FIRST,true,false,false); }

  1. @Bean(name = "secondTopicQueue")
  2. public Queue secondTopicQueue() {
  3.     return new Queue(TOPIC_QUEUE_SECOND,true,false,false);
  4. }
  6. @Bean(value = "topicExchange")
  7. public TopicExchange topicExchange() {
  8.     return new TopicExchange(TOPIC_EXCHANGE);
  9. }
  11. @Bean
  12. public Binding bindingFirstTopicExchange(Queue firstTopicQueue, TopicExchange topicExchange) {
  13.     return BindingBuilder.bind(firstTopicQueue).to(topicExchange).with(ROUTING_KEY_TOPIC_FIRST);
  14. }
  16. @Bean
  17. public Binding bindingSecondTopicExchange(Queue secondTopicQueue, TopicExchange topicExchange) {
  18.     return BindingBuilder.bind(secondTopicQueue).to(topicExchange).with(ROUTING_KEY_TOPIC_SECOND);
  19. }

}

  2. - 生产者
  3. ```java
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  10. @Component
  11. public class Producter {
  12.     private Logger logger = LoggerFactory.getLogger(Producter.class);
  14.     @Autowired
  15.     private RabbitTemplate rabbitTemplate;
  17.     public void topicFirstOne(String message){
  18.         logger.info("product topic first:" + message);
  19.         rabbitTemplate.convertAndSend("topicExchange","routingKey-topic-first.one",message);
  20.     }
  21.     // 并未发送到topicQueueSecond队列里
  22.     public void topicFirstTwo(String message){
  23.         logger.info("product topic first:" + message);
  24.         rabbitTemplate.convertAndSend("topicExchange","routingKey-topic-first.one.two",message);
  25.     }
  27.     public void topicSecondOne(String message){
  28.         logger.info("product topic second:" + message);
  29.         rabbitTemplate.convertAndSend("topicExchange","routingKey-topic-second.one",message);
  30.     }
  32.     public void topicSecondTwo(String message){
  33.         logger.info("product topic second:" + message);
  34.         rabbitTemplate.convertAndSend("topicExchange","routingKey-topic-second.one.two",message);
  35.     }
  36. }
  • 消费者 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Consumer { private Logger logger = LoggerFactory.getLogger(Consumer.class); @Autowired private RabbitTemplate rabbitTemplate;

  1. @RabbitListener(queues = "topicQueueFirst")
  2. @RabbitHandler
  3. public void topicFirst(String message){
  4.     logger.info("consume topic first:"+message);
  5. }
  7. @RabbitHandler
  8. @RabbitListener(queues = "topicQueueSecond")
  9. public void topicSecond(String message){
  10.     logger.info("consume topic second:"+message);
  11. }

}

  2. - 测试
  3. ```java
  4. @RestController
  5. @RequestMapping(value = "mq-test")
  6. public class MQTest {
  7.     private Logger logger = LoggerFactory.getLogger(MQTest.class);
  9.     @Autowired
  10.     private Producter producter;
  12.     @GetMapping
  13.     public void test(){
  14.         producter.topicFirstOne("First One");
  15.         producter.topicFirstTwo("First Two");
  16.         producter.topicSecondOne("Second One");
  17.         producter.topicSecondTwo("Second Two");
  18.     }
  19. }

image.png

FanoutExchang 扇型交换机

扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

  • 配置交换机、队列 ```java import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;

@Configuration public class FanoutExchangeConfig {

  1. @Bean(name = "fanoutQueueFirst")
  2. public Queue firstFanoutQueue() {
  3.     return new Queue("queue-fanout-first");
  4. }
  6. @Bean(name = "fanoutQueueSecond")
  7. public Queue secondFanoutQueue() {
  8.     return new Queue("queue-fanout-second");
  9. }
  11. @Bean(name = "fanoutExchange")
  12. public FanoutExchange fanoutExchange() {
  13.     return new FanoutExchange("fanoutExchange");
  14. }
  16. @Bean
  17. public Binding bindingFirstFanoutExchange(Queue fanoutQueueFirst, FanoutExchange fanoutExchange) {
  18.     return BindingBuilder.bind(fanoutQueueFirst).to(fanoutExchange);
  19. }
  21. @Bean
  22. public Binding bindingSecondFanoutExchange(Queue fanoutQueueSecond, FanoutExchange fanoutExchange) {
  23.     return BindingBuilder.bind(fanoutQueueSecond).to(fanoutExchange);
  24. }

}

  2. - 生产者
  3. ```java
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  10. @Component
  11. public class Producter {
  12.     private Logger logger = LoggerFactory.getLogger(Producter.class);
  13.     @Autowired
  14.     private RabbitTemplate rabbitTemplate;
  15.     public void fanout(String message){
  16.         logger.info("product fanout: "+message);
  17.         rabbitTemplate.convertAndSend("fanoutExchange", null, message);
  18.     }
  19. }
  • 消费者 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Consumer { private Logger logger = LoggerFactory.getLogger(Consumer.class); @RabbitHandler @RabbitListener(queues = “queue-fanout-first”) public void fanoutFirst(String message){ logger.info(“consume fanout first:”+message); } @RabbitHandler @RabbitListener(queues = “queue-fanout-second”) public void fanoutSecond(String message){ logger.info(“consume fanout second:”+message); } }

  2. - 测试
  3. ```java
  4. @RestController
  5. @RequestMapping(value = "mq-test")
  6. public class MQTest {
  7.     private Logger logger = LoggerFactory.getLogger(MQTest.class);
  9.     @Autowired
  10.     private Producter producter;
  12.     @GetMapping
  13.     public void test(){
  14.         producter.fanout("fanout message");
  15.     }
  16. }

image.png

HeadersExchange交换机

  • 配置交换机、队列 ```java import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.HeadersExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;

import java.util.HashMap; import java.util.Map;

@Configuration public class HeadersExchangeConfig { @Bean(name = “headersQueueFirst”) public Queue firstHeadersQueue() { return new Queue(“queue-headers-first”); }

  1. @Bean(name = "headersQueueSecond")
  2. public Queue secondHeadersQueue() {
  3.     return new Queue("queue-headers-second");
  4. }
  6. @Bean(name = "headersExchangeFirst")
  7. public HeadersExchange firstHeadersExchange() {
  8.     return new HeadersExchange("headersExchange-first");
  9. }
  11. @Bean(name = "headersExchangeSecond")
  12. public HeadersExchange secondHeadersExchange() {
  13.     return new HeadersExchange("headersExchange-second");
  14. }
  16. @Bean
  17. public Binding bindingFirstHeadersExchange(Queue headersQueueFirst, HeadersExchange headersExchangeFirst) {
  18.     Map<String,Object> headerValues = new HashMap<>();
  19.     headerValues.put("type", "message");
  20.     headerValues.put("name", "headers");
  21.     return BindingBuilder.bind(headersQueueFirst).to(headersExchangeFirst).whereAll(headerValues).match();
  22. }
  24. @Bean
  25. public Binding bindingSecondHeadersExchange(Queue headersQueueSecond, HeadersExchange headersExchangeSecond) {
  26.     Map<String,Object> headerValues = new HashMap<>();
  27.     headerValues.put("type", "message");
  28.     headerValues.put("name", "headers");
  29.     return BindingBuilder.bind(headersQueueSecond).to(headersExchangeSecond).whereAny(headerValues).match();
  30. }

}

  2. - 生产者
  3. ```java
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.amqp.core.Message;
  7. import org.springframework.amqp.core.MessageDeliveryMode;
  8. import org.springframework.amqp.core.MessageProperties;
  9. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.stereotype.Component;
  13. import java.util.Map;
  15. @Component
  16. public class Producter {
  17.     private Logger logger = LoggerFactory.getLogger(Producter.class);
  18.     @Autowired
  19.     private RabbitTemplate rabbitTemplate;
  20.     public void headersFirst(Map<String, Object> headers,String message){
  21.         logger.info("product headers first: "+ message);
  22.         rabbitTemplate.convertAndSend("headersExchange-first",null, getMessage(headers, message));
  23.     }
  24.     public void headersSecond(Map<String, Object> headers,String message){
  25.         logger.info("product headers second: "+ message);
  26.         rabbitTemplate.convertAndSend("headersExchange-second", null, getMessage(headers, message));
  27.     }
  28.     private Object getMessage(Map<String, Object> head, String msg) {
  29.         // 声明消息 (消息体, 消息属性)
  30.         MessageProperties messageProperties = new MessageProperties();
  31.         // 设置消息是否持久化。Persistent表示持久化，Non-persistent表示不持久化
  32.         messageProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
  33.         messageProperties.setContentType("UTF-8");
  34.         messageProperties.getHeaders().putAll(head);
  36.         return new Message(msg.getBytes(), messageProperties);
  38.     }
  39. }
  • 消费者 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Consumer { private Logger logger = LoggerFactory.getLogger(Consumer.class); @RabbitHandler @RabbitListener(queues = “queue-headers-first”) public void headersFirst(String message){ logger.info(“consume headers first:”+message); } @RabbitHandler @RabbitListener(queues = “queue-headers-second”) public void headersSecond(String message){ logger.info(“consume headers second:”+message); } }

  2. - 测试
  3. ```java
  4. @RestController
  5. @RequestMapping(value = "mq-test")
  6. public class MQTest {
  7.     private Logger logger = LoggerFactory.getLogger(MQTest.class);
  9.     @Autowired
  10.     private Producter producter;
  12.     @GetMapping
  13.     public void test(){
  14.         Map<String,Object> headers = new HashMap<>();
  15.         headers.put("type", "message");
  16.         Map<String,Object> headersAll = new HashMap<>();
  17.         headersAll.put("type", "message");
  18.         headersAll.put("name", "headers");
  19.         producter.headersFirst(headers,"headers message first");
  20.         producter.headersFirst(headersAll,"headers message first all");
  22.         producter.headersSecond(headers,"headers message second");
  23.         producter.headersSecond(headersAll,"headers message second all");
  24.     }
  25. }

image.png

拓展

发送对象

注意：传递的对象必须支持序列化（实现了Serializable接口）

  • 对象类 ```java import java.io.Serializable;

public class Message implements Serializable { private Integer status; private String content;

  1. public Message() {
  2. }
  4. public Message(Integer status, String content) {
  5.     this.status = status;
  6.     this.content = content;
  7. }
  9. @Override
  10. public String toString() {
  11.     return "Message{" +
  12.             "status=" + status +
  13.             ", content='" + content + '\'' +
  14.             '}';
  15. }
  17. public Integer getStatus() {
  18.     return status;
  19. }
  21. public void setStatus(Integer status) {
  22.     this.status = status;
  23. }
  25. public String getContent() {
  26.     return content;
  27. }
  29. public void setContent(String content) {
  30.     this.content = content;
  31. }

}

  2. - 配置交换机、队列
  3. ```java
  4. import org.springframework.amqp.core.*;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. @Configuration
  9. public class DirectConfig {
  10.     @Bean(name = "classQueue")
  11.     public Queue classQueue(){
  12.         return new Queue("queue-class");
  13.     }
  14.     @Bean
  15.     public DirectExchange directExchange() {
  16.         return new DirectExchange("directExchange-class", true, false);
  17.     }
  18.     @Bean
  19.     public Binding bindingDirect(Queue classQueue, DirectExchange directExchange) {
  20.         return BindingBuilder.bind(classQueue).to(directExchange).with("routingKey-class");
  21.     }
  22. }
  • 生产者 ```java import com.example.demo.rabbit.entity.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Producter { private Logger logger = LoggerFactory.getLogger(Producter.class);

  1. @Autowired
  2. private RabbitTemplate rabbitTemplate;
  4. public void classTest(Message message){
  5.     logger.info("product class:" + message.toString());
  6.     String routingKey = "routingKey-class";
  7.     rabbitTemplate.convertAndSend("directExchange-class",routingKey,message);
  8. }

}

  2. - 消费者
  3. ```java
  4. import com.example.demo.rabbit.entity.Message;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  9. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.stereotype.Component;
  13. @Component
  14. public class Consumer {
  15.     private Logger logger = LoggerFactory.getLogger(Consumer.class);
  17.     @RabbitHandler
  18.     @RabbitListener(queues = "queue-class")
  19.     public void classTest(Message message){
  20.         logger.info("consume class:"+message);
  21.     }
  22. }
  • 测试 ```java

@RestController @RequestMapping(value = “mq-test”) public class MQTest { private Logger logger = LoggerFactory.getLogger(MQTest.class);

  1. @Autowired
  2. private Producter producter;
  4. @GetMapping
  5. public void test(){
  6.     Message message = new Message(1,"Message");
  7.     producter.classTest(message);
  8. }

}

  2. - 结果
  4. ![image.png](https://cdn.nlark.com/yuque/0/2021/png/1416299/1630305134801-14c015cc-3f2a-40b7-852c-ab4af64c036b.png#clientId=uc8cf89ec-e11c-4&from=paste&height=38&id=u2f9d40e9&margin=%5Bobject%20Object%5D&name=image.png&originHeight=50&originWidth=864&originalType=binary&ratio=1&size=39169&status=done&style=none&taskId=u42f207d7-fbf5-42f4-91aa-1d29a1c3045&width=648)
  6. <a name="N9A4U"></a>
  7. ### 工作模式 - 多个消费者
  8. 采用上面DirectExchange交换机，需启动多个消费者,并不会重复消费
  10. - 消费者
  11. ```java
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  15. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  16. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  17. import org.springframework.beans.factory.annotation.Autowired;
  18. import org.springframework.stereotype.Component;
  21. @Component
  22. public class Consumer {
  23.     private Logger logger = LoggerFactory.getLogger(Consumer.class);
  25.     @RabbitHandler
  26.     @RabbitListener(queues = "queue-direct")
  27.     public void direct(String message){
  28.         logger.info("consume direct:"+message);
  29.     }
  30.     @RabbitHandler
  31.     @RabbitListener(queues = "queue-direct")
  32.     public void direct1(String message){
  33.         logger.info("consume direct 1:"+message);
  34.     }
  35. }

  • 测试

    1. @RestController
    2. @RequestMapping(value = "mq-test")
    3. public class MQTest {
    4.   private Logger logger = LoggerFactory.getLogger(MQTest.class);
    6.   @Autowired
    7.   private Producter producter;
    9.   @GetMapping
    10.   public void test(){
    11.       for (int i=0;i<5;i++){
    12.           producter.direct("direct message "+i);
    13.       }
    14.   }
    15. }

    image.png

RPC

RabbitMQ支持RPC远程调用，同步返回结果。返回的结果可以是一个对象
image.png

  • 配置交换机、队列，采用上面DirectExchange 交换机配置
  • 生产者 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

@Component public class Producter { private Logger logger = LoggerFactory.getLogger(Producter.class); @Autowired private RabbitTemplate rabbitTemplate; public void sender(String content){ logger.info(“product message: “+content); String message= (String) rabbitTemplate.convertSendAndReceive(“directExchange”,”routingKey-direct”,content); logger.info(“return message:”+message); } }

  2. - 消费者
  3. ```java
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Component;
  13. @Component
  14. public class Consumer {
  15.     private Logger logger = LoggerFactory.getLogger(Consumer.class);
  17.     @RabbitHandler
  18.     @RabbitListener(queues = "queue-direct")
  19.     public String receive(String content){
  20.         logger.info("consume message: "+content);
  21.         return "OK";
  22.     }
  23. }

  • 测试

    1. @RestController
    2. @RequestMapping(value = "mq-test")
    3. public class MQTest {
    4.   private Logger logger = LoggerFactory.getLogger(MQTest.class);
    6.   @Autowired
    7.   private Producter producter;
    9.   @GetMapping
    10.   public void test(){
    11.       producter.sender("RPC");
    12.   }
    13. }

    image.png