原理：

交换机种类

RabbitMQ常用的交换器类型有fanout、direct、topic、headers这四种类型
（1）fanout：它会把所有的交换器上的消息路由到所有与该交换器邦定的队列中，不需要BindingKey生效
（2）direct：它会把消息路由到BindingKey与RoutingKey完全匹配的队列中。比如在发送消息的时候，设置Label中RoutingKey为warning，则消息会路由到Queue1与Queue2上（请看下图）。
demo-rabbitmq-developer - 图1
（3）topic：是direct上的扩展，同样是利用RoutingKey与BindingKey相匹配，但是匹配规则不一样，支持模糊匹配。有如下的规则

  1.   - RoutingKey为一个点号“.”分隔的字符串，每个被隔开的独立字符串即为一个单词，是匹配的单位；
  2.   - BindingKey和RoutingKey一样，也是"."分割的字符串；
  3.   - 但不同的是BindingKey，可以用“#”，“*”进行类似于占位符的模糊匹配，“#”表示一个单词，"*"表示多个单词（也可以是零个）

比如:
RoutingKey为com.hidden.client的消息只会到队列Queue2中：因为只有Queue2的BindingKey=..client匹配com.hidden.client
RoutingKey为com.rabbitmq.client的消息会到队列Queue1-2中: 因为Queue1的BindingKey=.rabbitmq..匹配com.rabbitmq.client，Queue2的BindingKey=..client匹配com.rabbitmq.client
RoutingKey为java.rabbitmq.demo的消息只会到队列Queue1中：因为只有Queue1的BindingKey=.rabbitmq..匹配java.rabbitmq.demo
RoutingKey为java.util.concurrent的消息会被丢弃或者返回给生产者（需要设置）
demo-rabbitmq-developer - 图2

（4）headers：依赖发送消息内容中的hearders属性进行匹配，在绑定队列和交换器时指定一组键值对，这里的也就是headers，当发送消息到交换器时，RabbitMQ会获取到该消息的headers，通过比较会路由到相关队列中，这种交换器性能会很差，一般不会使用。

应答模式

  • AcknowledgeMode.NONE：不确认
  • AcknowledgeMode.AUTO：自动确认
  • AcknowledgeMode.MANUAL：手动确认

    全局配置

    1. spring:
    2. rabbitmq:
    3.   listener:
    4.     simple:
    5.       acknowledge-mode: manual

    在特定接受者上配置

    ```java @RabbitListener(queues =”topic.message”) public void process1(@Payload String message,@Header(AmqpHeaders.DELIVERY_TAG)long deliveryTag, Channel channel)throws Exception {
    1.  //定义队列的消费者
    2.  Consumer consumer = new DefaultConsumer(channel);
    3.  //监听队列，设置手动返回完成，第二个参数 false表示需要手动确认返回
    4.  channel.basicConsume("topic.message",false,consumer);
    5.  //手动确认返回-不成功就用basicNack
    6.  channel.basicAck(deliveryTag,false);
    7.  System.out.print("这里是手动返回确认信息 ");
    }
  2. <a name="qN1Mp"></a>
  3. # pom配置
  4. ```xml
  5. <?xml version="1.0" encoding="UTF-8"?>
  6. <project xmlns="http://maven.apache.org/POM/4.0.0"
  7.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  8.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  9.     <parent>
  10.         <artifactId>springcloud-demo</artifactId>
  11.         <groupId>demo</groupId>
  12.         <version>1.0-SNAPSHOT</version>
  13.     </parent>
  14.     <modelVersion>4.0.0</modelVersion>
  16.     <artifactId>demo-stream-rabbit-provider</artifactId>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-web</artifactId>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.springframework.cloud</groupId>
  25.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  26.         </dependency>
  27.         <!--sentinel-->
  28.         <dependency>
  29.             <groupId>com.alibaba.cloud</groupId>
  30.             <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
  31.         </dependency>
  33.         <!--健康监控-->
  34.         <dependency>
  35.             <groupId>org.springframework.boot</groupId>
  36.             <artifactId>spring-boot-starter-actuator</artifactId>
  37.         </dependency>
  39.         <dependency>
  40.             <groupId>com.alibaba.cloud</groupId>
  41.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  42.         </dependency>
  43.         <dependency>
  44.             <groupId>com.alibaba.cloud</groupId>
  45.             <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
  46.         </dependency>
  47.         <!-- mybatis 和SpringBoot 整合-->
  48.         <dependency>
  49.             <groupId>org.mybatis.spring.boot</groupId>
  50.             <artifactId>mybatis-spring-boot-starter</artifactId>
  51.         </dependency>
  53.         <!-- MySQL 驱动 -->
  54.         <dependency>
  55.             <groupId>mysql</groupId>
  56.             <artifactId>mysql-connector-java</artifactId>
  57.         </dependency>
  58.         <dependency>
  59.             <groupId>com.alibaba</groupId>
  60.             <artifactId>druid</artifactId>
  61.         </dependency>
  62.         <!-- jdbc -->
  63.         <dependency>
  64.             <groupId>org.springframework.boot</groupId>
  65.             <artifactId>spring-boot-starter-jdbc</artifactId>
  66.         </dependency>
  68.         <dependency>
  69.             <groupId>org.springframework.boot</groupId>
  70.             <artifactId>spring-boot-devtools</artifactId>
  71.             <scope>runtime</scope>
  72.             <optional>true</optional>
  73.         </dependency>
  75.         <dependency>
  76.             <groupId>org.projectlombok</groupId>
  77.             <artifactId>lombok</artifactId>
  78.         </dependency>
  80.         <dependency>
  81.             <groupId>org.springframework.boot</groupId>
  82.             <artifactId>spring-boot-starter-test</artifactId>
  83.             <scope>test</scope>
  84.         </dependency>
  85.         <dependency>
  86.             <groupId>junit</groupId>
  87.             <artifactId>junit</artifactId>
  88.         </dependency>
  89.     </dependencies>
  91. </project>

yaml配置

  1. server:
  2.   port: 8003
  4. spring:
  5.   application:
  6.     name: demo-stream-rabbit-provider
  7.   datasource:
  8.     type: com.alibaba.druid.pool.DruidDataSource
  9.     driver-class-name: com.mysql.jdbc.Driver
  10.     url: jdbc:mysql://192.168.2.20:30569/test?useUnicode=true&characterEncoding=utf-8&useSSL=false
  11.     username: root
  12.     password: root
  13.   cloud:
  14.     nacos:
  15.       server-addr: 192.168.2.20:30010
  16.       username: nacos
  17.       password: nacos
  18.     sentinel:
  19.       transport:
  20.         dashboard: 192.168.2.20:31596    # 控制台的安装位置
  21.     #        port: 8719    # 与sentinel单独连接的端口
  22.     #        client-ip: 192.168.2.6    # 本机的ip，如果sentinel装在虚拟机，必须配这个
  23.     #        port: 30195    # 与sentinel单独连接的端口
  24.     #        client-ip: 192.168.2.20    # 本机的ip，如果sentinel装在虚拟机，必须配这个
  25.     stream:
  26.       binders: # 在此处配置要绑定的rabbitmq的服务信息；
  27.         defaultRabbit: # 表示定义的名称，用于于binding整合
  28.           type: rabbit # 消息组件类型
  29. #          environment: # 设置rabbitmq的相关的环境配置
  30. #            spring:
  31. #              rabbitmq:
  32. #                host: 192.168.2.20
  33. #                port: 31672
  34. #                username: user
  35. #                password: ze2tgb2WGC
  37.       bindings: # 服务的整合处理
  38.         output: # 这个名字是一个通道的名称
  39.           destination: studyExchange # 表示要使用的Exchange名称定义
  40.           content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
  41.           binder: defaultRabbit  # 设置要绑定的消息服务的具体设置
  42.           group: default-group
  43.         input: # 这个名字是一个通道的名称
  44.           destination: studyExchange # 表示要使用的Exchange名称定义
  45.           binder: defaultRabbit  # 设置要绑定的消息服务的具体设置
  46.           group: default-group
  48.   rabbitmq:
  49.     host: 192.168.2.20
  50.     port: 31672
  51.     username: user
  52.     password: ze2tgb2WGC
  55. # 暴露应用信息
  56. management:
  57.   endpoints:
  58.     web:
  59.       exposure:
  60.         include: '*'

示例：controller

  1. package com.test.rabbit.consumer.controller;
  3. import com.test.rabbit.consumer.service.MessageProviderService;
  4. import lombok.RequiredArgsConstructor;
  5. import org.springframework.web.bind.annotation.GetMapping;
  6. import org.springframework.web.bind.annotation.RequestMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  9. /**
  10.  * @author jia
  11.  * @since 2022-04-05 11:39
  12.  */
  13. @RestController
  14. @RequestMapping("/rabbit")
  15. @RequiredArgsConstructor
  16. public class SendMessageController {
  17.     protected final MessageProviderService messageProviderService;
  19.     @GetMapping("/sendMessage")
  20.     public String sendMessage() {
  21.         return messageProviderService.send();
  22.     }
  23. }

示例：service

  1. package com.test.rabbit.consumer.serviceimpl;
  3. import com.test.rabbit.consumer.service.MessageProviderService;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.cloud.stream.annotation.EnableBinding;
  6. import org.springframework.cloud.stream.messaging.Source;
  7. import org.springframework.messaging.MessageChannel;
  8. import org.springframework.messaging.support.MessageBuilder;
  10. import javax.annotation.Resource;
  11. import java.util.UUID;
  13. /**
  14.  * @author jia
  15.  * @since 2022-04-05 11:42
  16.  */
  17. @EnableBinding(Source.class)
  18. @Slf4j
  19. public class MessageProviderServiceImpl implements MessageProviderService {
  20.     @Resource
  21.     MessageChannel output;
  23.     @Override
  24.     public String send() {
  25.         String serial = UUID.randomUUID().toString();
  26.         output.send(MessageBuilder.withPayload(serial).build());
  27.         log.info("***** serial: {}", serial);
  28.         return null;
  29.     }
  30. }

参考

https://github.com/spring-cloud/spring-cloud-stream-samples/
https://www.jianshu.com/p/33cbe3132e63