tencent cloud

Feedback

Sending and Receiving General Messages

Last updated: 2023-10-30 10:38:25

    Overview

    This document describes how to use open-source SDK to send and receive messages by using the SDK for Java as an example and helps you better understand the message sending and receiving processes.

    Prerequisites

    You have created or prepared the required resources as instructed in Resource Creation and Preparation.
    You have installed JDK 1.8 or later.
    You have installed Maven 2.5 or later.
    You have downloaded the demo or obtained the demo in TencentCloud/rocketmq-demo in GitHub.

    Directions

    Step 1. Install the Java dependent library

    Introduce dependencies in a Java project and add the following dependencies to the pom.xml file. This document uses a Maven project as an example.
    Note
    The dependency version must be v4.9.3 or later, preferably v4.9.4.
    <!-- in your <dependencies> block -->
    <dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.4</version>
    </dependency>
    <dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-acl</artifactId>
    <version>4.9.4</version>
    </dependency>

    Step 2. Produce messages

    Creating a message producer

    // Instantiate the message producer
    DefaultMQProducer producer = new DefaultMQProducer(
    groupName,
    new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL permission
    );
    // Set the Nameserver address
    producer.setNamesrvAddr(nameserver);
    // Start the producer instance
    producer.start();
    
    Parameter
    Description
    groupName
    Producer group name. We recommend that you use the corresponding topic name as the producer name.
    accessKey
    Role token, which can be copied in the Token column on the Role Management page.
    
    secretKey
    Role name, which can be copied on the Role Management page.
    nameserver
    Cluster access address, which can be obtained from Access Address in the Operation column on the Cluster page in the console. The namespace access address can be obtained under the Namespace tab on the Cluster page.

    Sending messages

    Messages can be sent in the sync, async, or one-way mode.
    Sync sending
    for (int i = 0; i < 10; i++) {
    // Create a message instance and set the topic and message content
    Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    // Send the message
    SendResult sendResult = producer.send(msg);
    System.out.printf("%s%n", sendResult);
    }
    Parameter
    Description
    topic_name
    Topic name, which can be copied under the Topic tab on the Cluster page in the console.
    TAG
    A parameter used to set the message tag.
    Async sending
    // Disable retry upon sending failures
    producer.setRetryTimesWhenSendAsyncFailed(0);
    // Set the number of messages to be sent
    int messageCount = 10;
    final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
    for (int i = 0; i < messageCount; i++) {
    try {
    final int index = i;
    // Create a message instance and set the topic and message content
    Message msg = new Message(topic_name, "TAG", ("Hello rocketMq " + index).getBytes(RemotingHelper.DEFAULT_CHARSET));
    producer.send(msg, new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {
    // Logic for message sending successes
    countDownLatch.countDown();
    System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
    }
    @Override
    public void onException(Throwable e) {
    // Logic for message sending failures
    countDownLatch.countDown();
    System.out.printf("%-10d Exception %s %n", index, e);
    e.printStackTrace();
    }
    });
    } catch (Exception e){
    e.printStackTrace();
    }
    }
    countDownLatch.await(5, TimeUnit.SECONDS);
    Parameter
    Description
    topic_name
    Topic name, which can be copied under the Topic tab on the Cluster page in the console.
    TAG
    A parameter used to set the message tag.
    One-way sending
    for (int i = 0; i < 10; i++) {
    // Create a message instance and set the topic and message content
    Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    // Send one-way messages
    producer.sendOneway(msg);
    }
    Parameter
    Description
    topic_name
    Topic name, which can be copied under the Topic tab on the Cluster page in the console.
    TAG
    A parameter used to set the message tag.
    Note
    For batch sending and other cases, see TencentCloud/rocketmq-demo in GitHub or the Apache RocketMQ documentation.

    Step 3. Consume messages

    Creating a consumer

    TDMQ for RocketMQ supports two consumption modes: push and pull. The push mode is recommended.
    // Instantiate the consumer
    DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
    groupName,
    new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); // ACL permission
    // Set the Nameserver address
    pushConsumer.setNamesrvAddr(nameserver);
    Parameter
    Description
    groupName
    Consumer group name, which can be copied under the Group tab on the Cluster page in the console.
    nameserver
    Cluster access address, which can be obtained from Access Address in the Operation column on the Cluster page in the console. The namespace access address can be obtained under the Namespace tab on the Cluster page.
    secretKey
    Role name, which can be copied on the Role Management page.
    accessKey
    Role token, which can be copied in the Token column on the Role Management page.
    

    Subscribing to messages

    The subscription modes vary by consumption mode.
    // Subscribe to a topic
    pushConsumer.subscribe(topic_name, "*");
    // Register a callback implementation class to process messages pulled from the broker
    pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    // Message processing logic
    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
    // Mark the message as being successfully consumed and return the consumption status
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    // Start the consumer instance
    pushConsumer.start();
    Parameter
    Description
    topic_name
    Topic name, which can be copied under the Topic tab on the Cluster page in the console.
    "*"
    If the subscription expression is left empty or specified as asterisk (*), all messages are subscribed to. tag1 || tag2 || tag3 means subscribing to multiple types of tags.

    Step 4. View consumption details

    Log in to the TDMQ console, go to the Cluster > Group page, and view the list of clients connected to the consumer group. Click Consumer Details in the Operation column to view consumer details.
    
    Note
    Above is a brief introduction to message publishing and subscription. For more information, see TencentCloud/rocketmq-demo or the Apache RocketMQ documentation.
    
    Contact Us

    Contact our sales team or business advisors to help your business.

    Technical Support

    Open a ticket if you're looking for further assistance. Our Ticket is 7x24 avaliable.

    7x24 Phone Support