spring-cloud-stream-redis

Read in other languages: 简体中文

Introduction

Sending and Receiving Redis Messages Based on Spring Cloud Stream Specification，Official Version Aligned with Spring Cloud Stream.

Documentation

https://guoshiqiufeng.github.io/spring-cloud-stream-redis/en/

Development Framework

• Spring Cloud Stream 4.1.3
• Spring Boot

Features

• PUBLISH SUBSCRIBE message
• QUEUE message（BLPOP BRPOP LPUSH RPUSH）

Tips 1: The two function modes cannot be mixed, i.e., if you send a message in PUBLISH SUBSCRIBE mode, you cannot receive a message in QUEUE mode, and vice versa.

Tips 2： PUBLISH SUBSCRIBE mode messages will be lost if not received, QUEUE mode will not.

Use

Introduces a uniform version dependency, so you don't have to specify a version number when you use it.

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.github.guoshiqiufeng.cloud</groupId>
            <artifactId>spring-cloud-stream-dependencies</artifactId>
            <version>0.3.0</version>
            <type>import</type>
        </dependency>
    </dependencies>
</dependencyManagement>

Introducing starter dependencies

<dependency>
    <groupId>io.github.guoshiqiufeng.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-redis</artifactId>
</dependency>

yml configuration

spring:
  cloud:
    stream:
      default-binder: redis
      binders:
        redis:
          type: redis
      redis:
        binder:
          configuration:
            host: 127.0.0.1
            port: 6379
            password: 123456
            database: 7
          support-type: queue_channel
      #        bindings:
      #          send-in-0:
      #            consumer:
      #              destination-is-pattern: true
      bindings:
        out-0:
          destination: test-topic
          content-type: text/plain
          group: push-producer-group
        send-in-0:
          destination: test-topic
          content-type: text/plain
          group: test-send-group

Messaging

@Autowired
private StreamBridge streamBridge;

@GetMapping("/send")
public String send() {
    MessageVO messageVO = new MessageVO();
    messageVO.setKey(UUID.randomUUID().toString());
    messageVO.setMsg("hello ");
    messageVO.setIds(Set.of("1", "2"));
    messageVO.setCreateTime(LocalDateTime.now());
    streamBridge.send("out-0", JSON.toJSONString(messageVO, JSONWriter.Feature.WriteClassName));
    return "success";
}

Message reception

@Slf4j
@Component("send")
public class MessageHandler implements Consumer<Message<String>> {

    /**
     * Performs this operation on the given argument.
     *
     * @param messageVOMessage the input argument
     */
    @Override
    public void accept(Message<String> messageVOMessage) {
        log.info("send Receive New Messages: {}", messageVOMessage.getPayload());
    }
}

For more usage references check the documentation