生產(chǎn)者
public class PulsarProducer {
private static String localClusterUrl = "pulsar://localhost:6650";
public static void main(String[] args) {
try {
Producer<byte[]> producer = getProducer();
String msg = "test send";
Long start = System.currentTimeMillis();
MessageId msgId = producer.send(msg.getBytes());
System.out.println("spend=" + (System.currentTimeMillis() - start) + ";send a message msgId = " + msgId.toString());
} catch (Exception e) {
System.err.println(e);
}
}
public static Producer<byte[]> getProducer() throws Exception {
PulsarClient client;
client = PulsarClient.builder().serviceUrl(localClusterUrl).build();
Producer<byte[]> producer = client.newProducer()
.topic("persistent://my-tenant/my-namespace/my-topic")
.producerName("test-producer")
.create();
return producer;
}
}
消費者
public class PulsarConsumerDemo {
private static String localClusterUrl = "pulsar://localhost:6650";
public static void main(String[] args) {
try {
//將訂閱消費者指定的主題和訂閱
Consumer<byte[]> consumer = getClient().newConsumer()
.topic("persistent://my-tenant/my-namespace/my-topic")
.subscriptionName("my-subscription")
.subscriptionType(SubscriptionType.Failover)
.subscribe();
while (true) {
Message msg = consumer.receive();
System.out.printf("consumer-Message received: %s. \n", new String(msg.getData()));
// 確認(rèn)消息代乃，以便broker刪除消息
consumer.acknowledge(msg);
}
} catch (Exception e) {
System.out.println(e);
}
}
public static PulsarClient getClient() throws Exception {
PulsarClient client;
client = PulsarClient.builder()
.serviceUrl(localClusterUrl).build();
return client;
}
}
