Spring Cloud Stream GCP如何重新排队失败的消息

Question

根据文档here，当前支持的绑定器（Rabbit 和 Kafka）依赖于 RetryTemplate。对于 GCP？

我的项目详情

Spring Boot 版本 2.1.3.RELEASE

依赖pom.xml

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
    <version>1.1.0.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-gcp-pubsub-stream-binder</artifactId>
    <version>1.1.0.RELEASE</version>
</dependency>

application.properties

spring.cloud.stream.bindings.input.destination=inputtopic
spring.cloud.stream.bindings.output.destination=outputtopic

spring.cloud.gcp.project-id=testinggcp
spring.cloud.gcp.credentials.location=file:C:/Users/my_gcp_credentials.json

休息控制器

@EnableBinding({Source.class,Sink.class})
@RestController
public class SourceExample {

    @Autowired
    private Source source;


    @GetMapping("/newMessage")
    public UserMessage sendMessage(@RequestParam("messageBody") String messageBody,
                                   @RequestParam("username") String username) {
        UserMessage userMessage = new UserMessage(messageBody, username, LocalDateTime.now());

        this.source.output().send(new GenericMessage<>(userMessage));
        return userMessage;
    }


    @StreamListener(target = Sink.INPUT)
    public void handle(UserMessage userMessage) throws IOException {
        System.out.println(userMessage);
    }


}

Answer 1

不，Spring Cloud GCP Pub/Sub Binder 不提供任何重试挂钩。

在@ServiceActivator 上使用RequestHandlerRetryAdvice 代替@StreamListener 很容易。因此，您在 POJO 方法中的所有失败都将根据您的配置进行重试。 RequestHandlerRetryAdvice 有一个 RecoveryCallback 选项，这可能只是简单的 ErrorMessageSendingRecoverer，您可以在其中配置一些错误处理并将错误消息发送到 GCP Pub/Sub 上的某些死信主题。

在参考手册中查看更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/#retry-advice

还有一点关于Advising Endpoints Using Annotations.