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. To obtain the endpoint, log in to the MQ console, go to the Producers page, and click Endpoint from the Actions column.  (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.
        // 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 {
                    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());
                    SendResult sendResult = producer.send(msg);
                    // 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 {
                    Message msg = new Message("TopicTestMQ", "TagA", "Hello MQ".getBytes());
                    SendResult sendResult = producer.send(msg);
                    // 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();
    }
}