rocketmq-reactive-client-spring's Introduction

RocketMQ-Reactive-Consumer-Spring

This project aims to help developers quickly integrate RocketMQ-Spring with Reactive.

Features

Consuming messages from RocketMQ with Reactive
Using Spring content ObjectMapper deserialization

Usage

使用和配置方式类似于 Spring Cloud Stream RocketMQ Binder，但解决一个Stream不能处理流中出现异常无法报告消息失败的问题。

application.yml

rocketmq:
  name-server: 192.168.1.21:9876
  producer:
    group: group-test-producer
  bindings:
    testFunction1-in-0:
      consumer:
        topic: test-topic
        group: group-test-consumer
        retry:
          maxAttempts: 3
          backoff:
            delay: 10000
            multiplier: 0.4

consumer

    @Bean
    fun contractCreateChannel(): (Message<String>) -> Mono<Void> = { str ->
        Mono.just("${str.payload}").then()
    }


    @Bean
    fun testChannel(): (Message<String>) -> Mono<ConsumerResult> = { str ->
        Mono.just("${str.payload}").doOnNext(::println).thenReturn(ConsumerResult.SUCCESS)
    }

    @Bean
    fun testReplayChannel(): (Message<String>) -> Mono<ReplayData> = { str ->
        Mono.just("${str.payload}").doOnNext(::println).thenReturn(ReplayData())
    }


    @Bean
    fun testConsumer1(): Consumer<Message<String>> = object: TypedConsumer<Message<String>>(Consumer { msg ->
            println("Receive New Create Messages: ${msg.payload}")
        }){}


    @Bean
    fun testFunction1() = object: TypedReplayConsumer<String>(Function { msg ->
            Mono.just("${msg}").thenReturn("replay")
        }) {}