0

我对 AvroSchema 有一个疑问，我正在创建一个具有 AvroSchema 的 Java 程序是一种模式。在这一行中 Producer producer = pulsarClient.newProducer(AvroSchema.of(Foo.class)).topic("test_topic")。创造（）; newProducer 方法的第一个参数是 schema，这里我们传递了一个 Foo 类的 AvroSchema，但是它给出的错误 AvroSchema 无法解析。

import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;

public class Test {

public static class Foo {
   private int field1 = 1;
   private String field2;
   private long field3;
public void setField1(int i) {
    // TODO Auto-generated method stub
    field1 = i;
}
public void setField2(String string) {
    // TODO Auto-generated method stub
    field2 = string;
}
public void setField3(long currentTimeMillis) {
    // TODO Auto-generated method stub
    field3 = currentTimeMillis;
}
}

public static void main(String[] args) throws Exception {
   PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
   Producer<Foo> producer = pulsarClient.newProducer(AvroSchema.of(Foo.class)).topic("test_topic").create();

   for (int i = 0; i < 1000; i++) {
       Foo foo = new Foo();
       foo.setField1(i);
       foo.setField2("foo" + i);
       foo.setField3(System.currentTimeMillis());
       producer.newMessage().value(foo).send();
   }
   producer.close();
   pulsarClient.close();
}
}

我正在关注https://pulsar.apache.org/docs/en/sql-getting-started/ 这个 url 以实现 pulsar sql 连接。这个问题的原因是什么？提前致谢 ！！

4

1 回答 1

0

我在您的代码中没有看到import org.apache.pulsar.client.impl.schema.AvroSchema;，我认为这是问题所在。如果没有，你能发布整个例外吗？

于 2019-04-26T13:34:24.610 回答