tencent cloud

package com.tencent.cloud.tdmq.pulsar.simple;

import com.tencent.cloud.tdmq.pulsar.Constant;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

import java.nio.charset.StandardCharsets;

/**
 * Synchronously send messages
 */
public class SimpleProducer {

    public static void main(String[] args) throws PulsarClientException, InterruptedException {

        // Initialize Pulsar client
        PulsarClient pulsarClient = Constant.initPulsarClient();
        // Construct a producer
        Producer<byte[]> producer = pulsarClient.newProducer()
                // Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`
                .topic("persistent://pulsar-xxx/sdk_java/topic1").create();
        System.out.println(">> pulsar producer created.");
        for (int i = 0; i < 10; i++) {
            String value = "my-sync-message-" + i;
            // Send the message
            MessageId msgId = producer.newMessage().key("key" + i).value(value.getBytes(StandardCharsets.UTF_8)).send();
            System.out.println("deliver msg " + msgId + ",value:" + value);

            Thread.sleep(500);
        }
        // Disable the producer
        producer.close();
        // Close the client
        pulsarClient.close();
    }
}



package com.tencent.cloud.tdmq.pulsar.simple;

import com.tencent.cloud.tdmq.pulsar.Constant;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * Consumer
 */
public class SimpleConsumer {


    public static void main(String[] args) throws PulsarClientException {
        // Initialize Pulsar client
        PulsarClient pulsarClient = Constant.initPulsarClient();
        // Construct a consumer
        Consumer<byte[]> consumer = pulsarClient.newConsumer()
                // Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`, copied from [Topic Management]
                .topic("persistent://pulsar-xxx/sdk_java/topic1")
                // You need to create a subscription on the topic details page in the console and enter the subscription name here
                .subscriptionName("sub1_topic1")
                // Declare the exclusive mode as the consumption mode
                .subscriptionType(SubscriptionType.Exclusive)
                // Configure consumption starting at the earliest offset; otherwise, historical messages may not be consumed
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
        System.out.println(">> pulsar consumer created.");
        for (int i = 0; i < 10; i++) {
            // Receive a message corresponding to the current offset
            Message<byte[]> msg = consumer.receive();
            MessageId msgId = msg.getMessageId();
            String value = new String(msg.getValue());
            System.out.println("receive msg " + msgId + ",value:" + value);
            // Messages must be acknowledged after being received, otherwise the offset will stay at the current message, causing message backlog
            consumer.acknowledge(msg);
        }
        // Close the consumer
        consumer.close();
        // Close the client
        pulsarClient.close();
    }
}