spring-boot - 带有 spring : know the message is publish or not? 的 PubSub

Question

我的发布者代码如下所示:

public abstract class PubSubPublisher {

    private static final Logger LOGGER = LoggerFactory.getLogger(PubSubPublisher.class);

    private final PubSubTemplate pubSubTemplate;

    protected PubSubPublisher(PubSubTemplate pubSubTemplate) {
        this.pubSubTemplate = pubSubTemplate;
    }

    protected abstract String topic(String topicName);

    public void publish(String topicName, String message) throws StatusRuntimeException {
        LOGGER.info("Publishing to topic [{}]. Message: [{}]", topicName, message);
        pubSubTemplate.publish(topicName, message);
    }

}

我的组件

@Component
public class HelloPubSubPublisher extends PubSubPublisher {

    @Autowired
    public HelloPubSubPublisher(PubSubTemplate pubSubTemplate) throws StatusRuntimeException{
        super(pubSubTemplate);
    }

    @Override
    protected String topic(String topicName) {
        return topicName;
    }

}

现在在我的服务层上，我如何获得我是否成功将消息发布到主题的天气，请注意我正在使用的所有 google api 都是异步的。

try {
    publisher.publish(topicName, payload);
}catch (Exception e) {
    LOGGER.error("ioException occured: "+e);
    throw new TopicNotFoundException();
}

不幸的是，我无法捕获任何错误，程序光标没有进入 catch block 。

最终，我想知道代码是否将消息推送到主题中，如果没有，那么我必须记录它并将该错误抛给客户端，而我当前的代码通过适当的异常处理不会发生这种情况。

感谢任何帮助或指导，谢谢。

Answer 1

使用函数 publish()，您应该能够捕获 future，您可以在其中检查消息是否已发布。

您在 Google's PubSub documentation 上有一个例子:

// Once published, returns a server-assigned message id (unique within the topic)
ApiFuture<String> future = publisher.publish(pubsubMessage);

// Add an asynchronous callback to handle success / failure
ApiFutures.addCallback(
    future,
    new ApiFutureCallback<String>() {

      @Override
      public void onFailure(Throwable throwable) {
        if (throwable instanceof ApiException) {
          ApiException apiException = ((ApiException) throwable);
          // details on the API exception
          System.out.println(apiException.getStatusCode().getCode());
          System.out.println(apiException.isRetryable());
        }
        System.out.println("Error publishing message : " + message);
      }

      @Override
      public void onSuccess(String messageId) {
        // Once published, returns server-assigned message ids (unique within the topic)
        System.out.println(messageId);
      }
    },
    MoreExecutors.directExecutor());