具有 CustomAttributes 的 PubSub - 未调用接收器函数，但出现 json 解析异常

Posted 2023-02-16

【中文标题】具有 CustomAttributes 的 PubSub - 未调用接收器函数，但出现 json 解析异常

【英文标题】：PubSub with CustomAttributes - Receiver function is not invoked but a json parse exception is coming up

【发布时间】：2021-01-03 04:36:55

【问题描述】：

Google Cloud 和 PubSub 服务。

我在下面的示例代码行中实现了一个简单的 CustomAttributeReceiver

https://github.com/googleapis/java-pubsub/blob/master/samples/snippets/src/main/java/pubsub/PublishWithCustomAttributesExample.java

我怎么会得到以下异常

com.google.gson.JsonSyntaxException: java.lang.IllegalStateException: 预期为字符串，但在路径 $ 处为 com.google.gson.Gson.fromJson(Gson.java:944) 处的 BEGIN_OBJECT com.google.gson.Gson.fromJson(Gson.java:1003) 在 com.google.cloud.functions.invoker.NewBackgroundFunctionExecutor$TypedFunctionExecutor.serviceLegacyEvent(NewBackgroundFunctionExecutor.java:257) 在 com.google.cloud.functions.invoker.NewBackgroundFunctionExecutor.serviceLegacyEvent(NewBackgroundFunctionExecutor.java:343) 在

有人可以说明我在这里缺少什么吗？

出版商方面

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import PubSubMessage;


            TopicName topicName = TopicName.of(projectId, topicId);
        System.out.println("informListenersAboutSucessfulRegisteration=" + topicName);
        Publisher publisher = null;
        try 
            publisher = Publisher.newBuilder(topicName).build();
            PubSubMessage newUserRegisterMsg = new PubSubMessage();
            Map<String, String> attributes = new HashMap<String, String>();
            attributes.put(PubSubMessage.newUserLanguage, newUserLanguage);
            newUserRegisterMsg.setAttributes(attributes);
            ByteString data = ByteString.copyFromUtf8("NewUserRegisteration");
            PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).putAllAttributes(attributes).build();
            ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
            String messageId = messageIdFuture.get();
            System.out.println("Published message ID: " + messageId);
         catch (Exception e) 
            Logger.getLogger(PubSubMessenger.name).log(Level.SEVERE, "Exception while publishing message", e);
         finally 
            if (publisher != null) 
                publisher.shutdown();
                publisher.awaitTermination(1, TimeUnit.MINUTES);
            
        

接收方

    import java.nio.charset.StandardCharsets;
    import java.util.Base64;
    
    import com.google.cloud.functions.BackgroundFunction;
    import com.google.cloud.functions.Context;
    import PubSubMessage;

    
public class SendEmailFromSendGrid implements BackgroundFunction<PubSubMessage> 
    public SendEmailFromSendGrid() 
    

    public void accept(PubSubMessage message, Context context) throws Exception 
        System.out.println("invoked accept");
        String name = "World";
        if (message != null && message.getData() != null) 
            name = new String(Base64.getDecoder().decode(message.getData().getBytes(StandardCharsets.UTF_8)),
                    StandardCharsets.UTF_8);
        
        System.out.println(String.format("Hello %s!", name));
        return;
    

PubSubMessage 定义

    import java.util.Map;

    public class PubSubMessage 

    public static final String newUserLanguage = "userLanguage";

    private String data;
    private Map<String, String> attributes;
    private String messageId;
    private String publishTime;

    public String getData() 
        return data;
    

    public void setData(String data) 
        this.data = data;
    

    public Map<String, String> getAttributes() 
        return attributes;
    

    public void setAttributes(Map<String, String> attributes) 
        this.attributes = attributes;
    

    public String getMessageId() 
        return messageId;
    

    public void setMessageId(String messageId) 
        this.messageId = messageId;
    

    public String getPublishTime() 
        return publishTime;
    

    public void setPublishTime(String publishTime) 
        this.publishTime = publishTime;
    

谢谢

【问题讨论】：

你能分享你的代码来重现我们这边的最小示例吗？

@guillaumeblaquiere pl 现在在问题中找到相同的内容。提前致谢。

您是否打印了message.getData() 值？你确定它仍然是b64格式吗？大多数客户端库会自动将内容转换为字符串。

嗨@guillaumeblaquiere 第一个 System.out.println("invoked accept");本身没有被调用。在调用此接受函数之前不确定其 PubSub 内部转换，并且我在发送 PubSub 消息时缺少某些格式

无法重现...您也可以分享 Java 导入吗？ （文件顶部）

【参考方案1】：

此答案由@user1241724 在评论部分提供：

重新完成了整个练习，现在它正在工作。唯一的区别是 在 PubSubMessage 中添加了默认构造函数。