SpringCloud Stream集成rabbitMQ（函数式集成，全网仅此一篇）

我就是我 2023-09-29 09:15 42阅读 0赞

#### 1.引入依赖 ####

implementation 'org.springframework.cloud:spring-cloud-stream'
    implementation 'org.springframework.cloud:spring-cloud-starter-stream-rabbit'

#### **2.创建yaml配置文件** ####

spring:
      application:
        name: rabbitmq-dome
      cloud:
        function:
          definition: source;sink; #函数名称，对应服务中的注入的Bean
        stream:
          binders: #需要绑定的rabbitmq的服务信息
            default-binder: #定义的名称，用于bidding整合
              type: rabbit #消息组件类型
              environment: #配置rabbimq连接环境
                spring:
                  rabbitmq:
                    addresses: 10.0.1.141:5672 #服务器的地址和端口
                    username: xt3dev #用户名
                    password: 4V9prcFbRoYxLCMd #密码
          bindings:
            source-out-0: #自定义消息通道的名称
              destination: QUEUE_DOME #exchange名称,交换模式默认是topic，创建时同时会创建QUEUE_DOME.${spring.application.name}队列
              content-type: application/json #设置消息的类型为json
              group: ${spring.application.name} #分组
              binder: default-binder #绑定的binder名称
            sink-in-0:
              destination: QUEUE_DOME
              content-type: application/json
              group: ${spring.application.name}
              binder: default-binder

#### 3.创建生产者类 ####

*  生产者接口

public interface ProducerService {
    
        /**
         * 发送消息
         * @param message
         */
        void sendMessage(String message);
    }

*  生产者实现类

注入source bean对象，然后通过创建Sinks对象，使用sinks来发送消息

@Service
    @Slf4j
    public class ProducerServiceImpl implements ProducerService {
    
        private final Sinks.Many<Message<String>> sinks =
                Sinks.many().multicast().onBackpressureBuffer();
    
        @Bean
        public Supplier<Flux<Message<String>>> source(){
            return sinks::asFlux;
        }
    
        @Override
        public void sendMessage(String message) {
            log.info("生产者发送消息:{}", message);
            Message<String> msg = MessageBuilder.withPayload(message).build();
            while (sinks.tryEmitNext(msg).isFailure()) {
                LockSupport.parkNanos(10);
            }
        }
    }

#### 4.创建消费者类 ####

*  消费者接口

public interface ConsumerService {
    }

*  消费者实现类

注入sink bean消费者对象来监听消息

@Service
    @Slf4j
    public class ConsumerServiceImpl implements ConsumerService {
    
        @Bean
        Consumer<String> sink() {
            return message-> {
                log.info("消费者监听到消息:{}", message);
            };
        }
    }

#### 5.创建controller类 ####

@RestController
    @RequiredArgsConstructor
    public class MessageController {
    
        private final ProducerService producerService;
    
        @GetMapping("/sendMessage")
        public void sendMessage(@RequestParam(value = "message", defaultValue = "hello world") String message) {
            producerService.sendMessage(message);
        }
    }

7.启动项目测试

在浏览器中访问[http://localhost:8080/sendMessage?message=hello rabbitMQ][http_localhost_8080_sendMessage_message_hello rabbitMQ]

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5pCs56CW6ICB5Yac5bel_size_20_color_FFFFFF_t_70_g_se_x_16][]

发现生产者发送的消息，消费者收到了。

[下节我们将讲述延时队列的实现][Link 1]

[http_localhost_8080_sendMessage_message_hello rabbitMQ]: http://localhost:8080/sendMessage?message=hello%20rabbitMQ
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5pCs56CW6ICB5Yac5bel_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/d95235bee00c41faafa6cddc091d4c16.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5pCs56CW6ICB5Yac5bel,size_20,color_FFFFFF,t_70,g_se,x_16
[Link 1]: https://mp.csdn.net/mp_blog/creation/editor/124041328