Apache Pulsar para Iniciantes: Configuração Básica e Exemplos em Java

package com.gasparbarancelli;

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

public class ConsumerDemo extends Thread {

    private final PulsarClient client;
    private final Consumer<byte[]> consumer;
    private volatile boolean running = true;

    public ConsumerDemo() throws Exception {
        // 1) Criamos um client do Pulsar apontando para pulsar://localhost:6650
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        // 2) Definimos o tópico, o tipo de assinatura (Exclusive) e a política de DLQ
        consumer = client.newConsumer()
                .topic("gasparbarancelli-demo-pulsar")
                .subscriptionName("demo-subscription")
                .subscriptionType(SubscriptionType.Exclusive)
                .deadLetterPolicy(
                    DeadLetterPolicy.builder()
                            .maxRedeliverCount(3) // após 3 tentativas, mensagem vai para DLQ
                            .deadLetterTopic("dlq-gasparbarancelli-demo-pulsar")
                            .build()
                )
                .subscribe();
    }

    public static void main(String[] args) throws Exception {
        // 3) Criamos e iniciamos a thread do consumidor
        ConsumerDemo consumerThread = new ConsumerDemo();
        consumerThread.start();

        // 4) Mantemos a execução por 10 segundos (para demonstração)
        Thread.sleep(10000);

        // 5) Encerramos a thread de consumo
        consumerThread.shutdown();
    }

    @Override
    public void run() {
        while (running) {
            try {
                // 6) Recebe a mensagem do tópico configurado
                Message<byte[]> msg = consumer.receive();
                String content = new String(msg.getData());

                // 7) Exibe o conteúdo recebido no console
                System.out.println("Recebido: " + content);

                // 8) Aqui, poderíamos testar alguma condição de erro; se falhar:
                // consumer.negativeAcknowledge(msg);

                // 9) Se processar com sucesso, confirmamos a mensagem
                consumer.acknowledge(msg);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void shutdown() {
        running = false;
        try {
            // 10) Fecha o consumer e o cliente
            consumer.close();
            client.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

package com.gasparbarancelli;

import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;

public class ProducerDemo extends Thread {

    private final PulsarClient client;
    private final Producer<byte[]> producer;
    private volatile boolean running = true;

    public ProducerDemo() throws Exception {
        // 1) Criamos um client do Pulsar apontando para pulsar://localhost:6650
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        // 2) Criamos o producer apontando para o mesmo tópico do consumidor
        producer = client.newProducer()
                .topic("gasparbarancelli-demo-pulsar")
                .create();
    }

    public static void main(String[] args) throws Exception {
        // 3) Criamos e iniciamos a thread do produtor
        ProducerDemo producerThread = new ProducerDemo();
        producerThread.start();

        // 4) Mantemos a execução por 10 segundos (para demonstração)
        Thread.sleep(10000);

        // 5) Encerramos a thread de produção
        producerThread.shutdown();
    }

    @Override
    public void run() {
        int count = 0;
        while (running) {
            try {
                // 6) Montamos a mensagem com um contador
                String message = "Mensagem " + count + " com delay de 1s";

                // 7) Enviamos a mensagem ao Pulsar
                MessageId msgId = producer.send(message.getBytes());
                System.out.println("Enviado: " + message + " | ID: " + msgId);

                // 8) Incrementamos o contador e aguardamos 1 segundo
                count++;
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void shutdown() {
        running = false;
        try {
            // 9) Fecha o producer e o client
            producer.close();
            client.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}