Send messages (using multiple threads)

The consumer and producer client objects of RocketMQ are thread-secure and can be shared among multiple threads.

You can deploy multiple producer and consumer instances on one or more servers. A producer or consumer instance can also run multiple threads to send or receive messages, improving the message transmitting or receiving TPS. Do not create a client instance for every thread.

The sample code for sharing a producer instance among multiple threads is as follows:

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import java.util.Properties;
public class SharedProducer {
    public static void main(String[] args) {
        // Initialize the producer configuration.
        Properties properties = new Properties();
        // The group ID you created in the console.
        properties.put(PropertyKeyConst.GROUP_ID, "XXX");
        // The AccessKeyId you created in the Alibaba Cloud console for identity authentication.
        properties.put(PropertyKeyConst.AccessKey,"XXX");
        // The AccessKeySecret you created in the Alibaba Cloud console for identity authentication.
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // Set the message transmission timeout period (in milliseconds). 
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        // Set the TCP endpoint: Go to the **Instances** page in the RocketMQ console, and view the endpoint in the **Endpoint Information** area.
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
          "XXX ");
        final Producer producer = ONSFactory.createProducer(properties);
        //Before sending a message, call the start() method once to start the producer.
        producer.start();
        // The created producer and consumer objects are thread-secure and can be shared among multiple threads. Do not create a client instance for every thread.
        // Two threads share the producer object and concurrently send messages to RocketMQ.
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Message msg = new Message( //
                    // The topic of the message.
                    "TopicTestMQ",
                    // The message tag, which is similar to a Gmail tag. It is used to sort messages, enabling the consumer to filter messages on the RocketMQ broker based on the specified criteria.
                    "TagA",
                    // The message body in any binary format. RocketMQ does not process the message body.
                    // The producer and consumer must negotiate the consistent serialization and deserialization methods.
                    "Hello MQ".getBytes());
                    SendResult sendResult = producer.send(msg);
                    // Send the message in synchronous mode. The message is sent if no exception is thrown.
                    if (sendResult ! = null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + MqConfig.TOPIC + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    // The message failed to be sent and requires a retry. The system can resend the message or store message data persistently.
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + MqConfig.TOPIC);
                    e.printStackTrace();
                }
            }
        });
        thread.start();
        Thread anotherThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Message msg = new Message("TopicTestMQ", "TagA", "Hello MQ".getBytes());
                    SendResult sendResult = producer.send(msg);
                    // Send the message in synchronous mode. The message is sent if no exception is thrown.
                    if (sendResult ! = null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + MqConfig.TOPIC + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    // The message failed to be sent and requires a retry. The system can resend the message or store message data persistently.
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + MqConfig.TOPIC);
                    e.printStackTrace();
                }
            }
        });
        anotherThread.start();
        // If the producer instance is no longer used, disable it to release resources.
        // producer.shutdown();
    }
}