Send messages - multiple threads

MQ consumer and producer client objects are thread-safe, which can be shared between multiple threads.

You can deploy multiple producer and consumer instances on a server (or multiple servers), or you can use multi-threads within one producer or consumer instance to send or receive messages, so as to increase message sending and receiving TPS. Do not create a client instance for every thread.

The sample code for sharing producer between 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 producer instance configurations
        Properties properties = new Properties();
        //Producer ID, Alibaba Cloud ID verification, which is created on Alibaba Cloud Management Console
        properties.put(PropertyKeyConst.ProducerId, "XXX");
        // AccessKey, Alibaba Cloud ID verification, which is created on Alibaba Cloud Management Console
        properties.put(PropertyKeyConst.AccessKey,"XXX");
        // SecretKey, Alibaba Cloud ID verification, which is created on Alibaba Cloud Management Console
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        //Set transmission timeout (ms) period 
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        // Set a TCP access domain name (the following uses public cloud production environment as an example)
        properties.put(PropertyKeyConst.ONSAddr,
          "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");
        final Producer producer = ONSFactory.createProducer(properties);
        // Before sending messages, the start method must be called once to start the producer.
        producer.start();
        // The producer and consumer objects are thread-safe, which can be shared between multiple threads. Do not create an instance for every thread.
        final Message msg = new Message( //
            // The topic of the message
            "TopicTestMQ",
            //Message tag, which is similar to tag in Gmail, and is used to classify messages.  Consumers can then set filtering conditions for messages to be filtered in MQ broker. 
            "TagA",
            // Message body, which can be any data in binary format.
            // Serialization and deserialization methods need to be negotiated and remain consistent between the producer and the consumer.
            "Hello MQ".getBytes());
        // Share producer object between a thread and another thread, and the two send messages concurrently to the MQ.
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    SendResult sendResult = producer.send(message);
                    // Synchronous message sending will succeed as long as 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) {
                    // If the message sending fails and a retry is needed, the message can be re-sent or persisted for compensated processing.
                    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 {
                    SendResult sendResult = producer.send(message);
                    // Synchronous message sending will succeed as long as 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) {
                    // If the message sending fails and a retry is needed, the message can be re-sent or persisted for compensated processing.
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + MqConfig.TOPIC);
                    e.printStackTrace();
                }
            }
        });
        anotherThread.start();
        // If the producer instance will not be used any more, it can be shut down to release resources.
        // producer.shutdown();
    }
}