Pulsar Consumer 例子

import org.apache.pulsar.client.api.Consumer;import org.apache.pulsar.client.api.Message;import org.apache.pulsar.client.api.PulsarClient;import org.apache.pulsar.client.api.SubscriptionInitialPosition;import org.apache.pulsar.client.api.SubscriptionType;import org.apache.pulsar.client.impl.schema.JSONSchema;public class ReceiveMsgTest {    public static void main(String[] args) {    String url = "http://192.168.1.48:8080";                try{            PulsarClient client =PulsarClient.builder()                    .serviceUrl(url)                                        .build();             Consumer consumer=client.newConsumer(JSONSchema.of(UserModel.class))             .topic("my-tenant/my-namespace/testschema-topic")                        .subscriptionType(SubscriptionType.Exclusive)//订阅模式  Exclusive(独占,默认模式) Failover（灾备）Shared（共享）             .subscriptionName("wbq_1")//订阅者名称             .subscribe();             while (true) {                Message userModelmsg = consumer.receive();                UserModel userModel=userModelmsg.getValue();                System.out.println("receive message: " +userModel.getName()+"="+userModel.getAge());                consumer.acknowledge(userModelmsg.getMessageId());//应答后此订阅者不会在收到此消息             }        }catch(Exception e){            e.printStackTrace();        }    }}

public class UserModel {    private String name;    private int age;    public String getName() {        return name;    }    public void setName(String name) {        this.name = name;    }    public int getAge() {        return age;    }    public void setAge(int age) {        this.age = age;    }}