如何将Redis配置为Spring Cloud数据流源？

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-data-redis</artifactId>
</dependency>

spring:
  redis:
    host: <Redis主机地址>
    port: <Redis端口号>
    password: <Redis密码>

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.integration.redis.outbound.RedisMessageHandler;
import org.springframework.integration.redis.serializer.RedisSerializer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Configuration
@ConditionalOnProperty("spring.redis.host") // 只有当spring.redis.host配置存在时才生效
public class RedisSourceConfiguration {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        template.setDefaultSerializer(RedisSerializer.string());
        return template;
    }

    @Bean
    public MessageChannel redisInputChannel() {
        return new DirectChannel();
    }

    @Bean
    @InboundChannelAdapter(channel = "redisInputChannel", poller = @Poller(fixedDelay = "1000"))
    public MessageSource<Object> redisMessageSource(RedisTemplate<String, Object> redisTemplate) {
        RedisInboundChannelAdapter adapter = new RedisInboundChannelAdapter(redisTemplate);
        adapter.setTopics("<Redis订阅的主题>");
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "redisInputChannel")
    public MessageHandler redisMessageHandler() {
        return new RedisMessageHandler(redisTemplate(redisConnectionFactory));
    }
}

@SpringBootApplication(scanBasePackages = "<RedisSourceConfiguration类所在的包路径>")
public class DataStreamApplication {

    public static void main(String[] args) {
        SpringApplication.run(DataStreamApplication.class, args);
    }
}

stream create myStream --definition "redis-source --spring.redis.host=<Redis主机地址> --spring.redis.port=<Redis端口号> --spring.redis.password=<Redis密码> | <后续处理步骤>" --deploy