<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

@Data
public class OrderForm {
  private String productName;
}

@Configuration
@Slf4j
public class MessageQueueConfig {

  @Bean
  public Function<OrderForm, OrderForm> inventory() {
    return orderForm -> {
      log.info("Inventory Received Message: " + orderForm);
      return orderForm;
    };
  }

  @Bean
  public Consumer<OrderForm> report() {
    return orderForm -> {
      log.info("Report Received Message: " + orderForm);
    };
  }
}

@Slf4j
@RestController
public class OrderController {

  @Autowired
  private BeanFactoryChannelResolver resolver;

  @PostMapping("order")
  public String order(@RequestBody OrderForm orderForm) {
    log.info("Received Request " + orderForm);
    resolver.resolveDestination("inventory-in-0").send(new GenericMessage<>(orderForm));
    return "success";
  }
}

spring:
 cloud:
  stream:
#   配置消息中間件信息
   binders:
    myLocalRabbit:
     type: rabbit
     environment:
      spring:
       rabbitmq:
        host: localhost
        port: 31003
        username: guest
        password: guest
        virtual-host: /
#   重點(diǎn)，如何綁定通道，這里有個(gè)約定，開(kāi)頭是函數(shù)名，in表示消費(fèi)消息，out表示生產(chǎn)消息，最后的數(shù)字是函數(shù)接受的參數(shù)的位置，destination后面為訂閱的主題
#   比如Function<Tuple2<Flux<String>, Flux<Integer>>, Flux<String>> gather()
#   gather函數(shù)接受的第一個(gè)String參數(shù)對(duì)應(yīng) gather-in-0，第二個(gè)Integer參數(shù)對(duì)應(yīng) gather-in-1，輸出對(duì)應(yīng) gather-out-0
   bindings:
    inventory-in-0:
     destination: inventory
    inventory-out-0:
     destination: report
    report-in-0:
     destination: report
#   注冊(cè)聲明的三個(gè)函數(shù)
   function:
    definition: inventory;report

POST http://localhost:8080/order
Content-Type: application/json

{
 "productName": "999"
}

POST http://localhost:8080/order

HTTP/1.1 200 
Content-Type: text/plain;charset=UTF-8
Content-Length: 7
Date: Sat, 30 May 2020 15:27:56 GMT
Keep-Alive: timeout=60
Connection: keep-alive

success

Response code: 200; Time: 56ms; Content length: 7 bytes

2020-05-30 23:27:56.956 INFO 8760 --- [nio-8080-exec-1] c.e.springcloudstream.OrderController  : Received Request OrderForm(productName=999)
2020-05-30 23:27:56.956 INFO 8760 --- [nio-8080-exec-1] o.s.i.h.s.MessagingMethodInvokerHelper  : Overriding default instance of MessageHandlerMethodFactory with provided one.
2020-05-30 23:27:56.957 INFO 8760 --- [nio-8080-exec-1] c.e.s.MessageQueueConfig         : Inventory Received Message: OrderForm(productName=999)
2020-05-30 23:27:56.958 INFO 8760 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory    : Attempting to connect to: [localhost:31003]
2020-05-30 23:27:56.964 INFO 8760 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory    : Created new connection: rabbitConnectionFactory.publisher#6131841e:0/SimpleConnection@192fe472 [delegate=amqp://guest@127.0.0.1:31003/, localPort= 2672]
2020-05-30 23:27:56.965 INFO 8760 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin     : Auto-declaring a non-durable, auto-delete, or exclusive Queue (inventory.anonymous.wtaFwHlNRkql5IUh2JCNAA) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2020-05-30 23:27:56.965 INFO 8760 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin     : Auto-declaring a non-durable, auto-delete, or exclusive Queue (report.anonymous.SJgpJKiJQf2tudszgf623w) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2020-05-30 23:27:56.979 INFO 8760 --- [f2tudszgf623w-1] o.s.i.h.s.MessagingMethodInvokerHelper  : Overriding default instance of MessageHandlerMethodFactory with provided one.
2020-05-30 23:27:56.980 INFO 8760 --- [f2tudszgf623w-1] c.e.s.MessageQueueConfig         : Report Received Message: OrderForm(productName=999)