This topic provides sample code to show how to publish and subscribe to delayed messages by using TCP client SDK for Java.

Prerequisites

The following operations are complete:

  • The SDK for Java is downloaded. For information about the release notes of the SDK for Java, see Release notes.
  • An environment is set up. For more information, see Prepare the environment.
  • Optional:Logging settings are configured. For more information, see Logging settings.

Background information

Delayed messages are delivered to a client after a delay. For example, if the delay period is 3 seconds, messages are forwarded from the Message Queue for Apache RocketMQ broker to the client for consumption after a 3 second delay. Delayed messages can be used when a time window is required between the production and consumption of the messages, or when tasks need to be triggered after a delay. Delayed messages are similar to delayed queues.

For information about the concepts and precautions of delayed messages, see Scheduled messages and delayed messages.

Note The first time you use Message Queue for Apache RocketMQ, we recommend that you refer to Demo project to create a Message Queue for Apache RocketMQ project before you publish and subscribe to messages.

Publish delayed messages

To obtain the sample code, go to the Message Queue for Apache RocketMQ code library.

The following sample code shows how to publish delayed messages:

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // The AccessKey ID that you created in the Resource Access Management (RAM) console. The AccessKey ID is used for identity authentication. 
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // The AccessKey secret that you created in the RAM console. The AccessKey secret is used for identity authentication. 
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // The TCP endpoint. To obtain the TCP endpoint, log on to the Message Queue for Apache RocketMQ console. In the left-side navigation pane, click Instances. On the Instances page, click the name of your instance. On the Instance Details page, scroll to the Basic Information section and view the TCP endpoint on the Endpoints tab. 
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
          "XXX");

        Producer producer = ONSFactory.createProducer(properties);
        // Before you publish a message, call the start() method to start the producer. You can call this method only once. 
        producer.start();
        Message msg = new Message( 
                // The topic that you created in the Message Queue for Apache RocketMQ console. 
                "Topic",
                // The message tag. The message tag is similar to a Gmail tag, and is used to sort messages and filter messages for the consumer on the Message Queue for Apache RocketMQ broker based on specified conditions. 
                "tag",
                // The message body in a binary format. Message Queue for Apache RocketMQ does not process the message body. The producer and the consumer must agree on the serialization and deserialization methods. 
                "Hello MQ".getBytes());
        // The key of the message. The key is the business-specific attribute of the message and must be globally unique. 
        // A unique key can be used to query and republish a message in the Message Queue for Apache RocketMQ console if the message fails to be received. 
        // Note: You can publish and subscribe to messages without the need to specify message keys. 
        msg.setKey("ORDERID_100");
        try {
            // The delay time (after the current time) in milliseconds (ms) after which the message is forwarded. You can set the delay time to up to 40 days. 
            // In the following example, messages are forwarded after a delay of 3 seconds. 
            long delayTime = System.currentTimeMillis() + 3000;

            // The point in time when the Message Queue for Apache RocketMQ broker starts to forward the message. 
            msg.setStartDeliverTime(delayTime);

            SendResult sendResult = producer.send(msg);
            // Publish the message in synchronous mode. If no error occurs, the message is published. 
            if (sendResult != null) {
            System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
            }
            } catch (Exception e) {
            // Specify the logic that you want to use to republish or persist the message if the message fails to be published and needs to be published again. 
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
        }
        // Before you exit the application, shut down the producer. 
        // Note: This step is optional. 
        producer.shutdown();
    }
}           

Subscribe to delayed messages

The mode that is used to subscribe to delayed messages is the same as the mode that is used to subscribe to normal messages. For more information, see Subscribe to messages.