生產(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;
    }
}

消費(fèi)者

public class PulsarConsumerDemo {
    private static String localClusterUrl = "pulsar://localhost:6650";

    public static void main(String[] args) {
        try {
            //將訂閱消費(fèi)者指定的主題和訂閱
            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;
    }
}