Spring Cloud 出站通道適配器

PubSubMessageHandler是GCP發(fā)布/訂閱的出站通道適配器，它在Spring MessageChannel上偵聽新消息。它使用PubSubTemplate將其發(fā)布到GCP發(fā)布/訂閱主題。

為了構(gòu)造消息的Pub / Sub表示，出站通道適配器需要將Spring Message有效載荷轉(zhuǎn)換為Pub / Sub期望的字節(jié)數(shù)組表示。它將這種轉(zhuǎn)換委托給PubSubTemplate。要自定義轉(zhuǎn)換，您可以在PubSubTemplate中指定一個(gè)PubSubMessageConverter，它將Object有效負(fù)載和Spring Message的標(biāo)頭轉(zhuǎn)換為PubsubMessage。

要使用出站通道適配器，必須在用戶應(yīng)用程序側(cè)提供PubSubMessageHandler bean并對(duì)其進(jìn)行配置。

@Bean
@ServiceActivator(inputChannel = "pubsubOutputChannel")
public MessageHandler messageSender(PubSubTemplate pubsubTemplate) {
    return new PubSubMessageHandler(pubsubTemplate, "topicName");
}

提供的PubSubTemplate包含將消息發(fā)布到GCP發(fā)布/訂閱主題的所有必要配置。

PubSubMessageHandler默認(rèn)情況下異步發(fā)布消息。可以將發(fā)布超時(shí)配置為同步發(fā)布。如果未提供任何內(nèi)容，則適配器將無限期等待響應(yīng)。

可以通過setPublishFutureCallback()方法為PubSubMessageHandler中的publish()調(diào)用設(shè)置用戶定義的回調(diào)。如果成功，這些對(duì)于處理消息ID很有用，如果拋出錯(cuò)誤，則對(duì)處理錯(cuò)誤ID是有用的。

要覆蓋默認(rèn)目的地，可以使用GcpPubSubHeaders.DESTINATION標(biāo)頭。

@Autowired
private MessageChannel pubsubOutputChannel;

public void handleMessage(Message<?> msg) throws MessagingException {
    final Message<?> message = MessageBuilder
        .withPayload(msg.getPayload())
        .setHeader(GcpPubSubHeaders.TOPIC, "customTopic").build();
    pubsubOutputChannel.send(message);
}

也可以使用setTopicExpression()或setTopicExpressionString()方法為主題設(shè)置SpEL表達(dá)式。