Send messages (multi-threaded)

You can deploy multiple producer and consumer instances on a server (or multiple servers), or use multiple threads to send or receive messages in the same producer or consumer instance. This improves the TPS of message sending or receiving. Avoid creating a client instance for each thread.

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

import java.util.Properties;
import com.alipay.sofa.sofamq.client.PropertyKeyConst;
import io.openmessaging.api.Message;
import io.openmessaging.api.MessagingAccessPoint;
import io.openmessaging.api.OMS;
import io.openmessaging.api.OMSBuiltinKeys;
import io.openmessaging.api.Producer;
import io.openmessaging.api.SendResult;

public class Main {
    public static void main(String... args) {
        Properties credentials = new Properties();
        // The AccessKey pair of an Alibaba Cloud account has permissions to access all API operations. This is a high risk. We strongly recommend that you create and use a RAM user for API access or routine O&M. Log on to the RAM console to create a RAM user. 
        // Save the AccessKey pair and AccessKeySecret in environment variables. 
        // We strongly recommend that you do not save the AccessKey and AccessKeySecret in the code. This may cause key leakage.
        credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "SOFA_AK_ENV");        
        credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "SOFA_SK_ENV");
        // Set the TCP endpoint and go to the Overview page in the console to view the endpoint configuration.
        MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
                .withCredentials(credentials).build();
        Properties properties = new Properties();
        // Specify a user instance and go to the Overview page in the console to view the endpoint configuration.
        properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");

        // The group ID that you created in the console.
        properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
        finalProducer producer = accessPoint.createProducer(properties);
        producer.start();

        // The created producer and consumer objects are thread safety and can be shared among multiple threads to prevent each thread from creating an instance. 

        // Share the producer object in the thread and anotherThread, and concurrently send messages to the MSMQ. 
        Thread thread = new Thread(newRunnable() {
            @Override
            public void run() {
                try {
                    Message msg = new Message(//
                            // The topic to which the message belongs.
                            "TopicTestMQ",
                            // Message tags can be understood as Gmail tags. Message tags are used to categorize messages so that the consumer can specify filter conditions to filter messages on the MSMQ server.
                            "TagA",
                            // The message body can be any binary data. The MSMQ does not perform any intervention. 
                            // The producer and consumer must agree on the serialization and deserialization methods.
                            "Hello MQ".getBytes());
                    SendResult sendResult = producer.send(msg);
                    // Send the message in synchronous mode. If no error occurs, the message is sent.
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + MqConfig.TOPIC + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    // Specify the logic to resend or persist the message if the message fails to be sent.
                    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. If no error occurs, the message is sent.
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + MqConfig.TOPIC + " msgId is: " + sendResult.getMessageId());
                    }
                } catch (Exception e) {
                    // Specify the logic to resend or persist the message if the message fails to be sent.
                    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, shut it down to release resources.
        // producer.shutdown();
    }
}