Los objetos producer y consumer en ApsaraMQ for RocketMQ son seguros para hilos. Comparta una sola instancia entre los hilos en lugar de crear una por cada hilo.
Por qué compartir instancias
Puede desplegar varias instancias de producer y consumer en uno o más brokers. También puede ejecutar varios hilos para enviar o recibir mensajes dentro de una misma instancia de producer o consumer. Crear una instancia por cada hilo desperdicia recursos y puede reducir el rendimiento.
Compartir un solo producer o consumer entre varios hilos ofrece las siguientes ventajas:
Reduce la sobrecarga de recursos tanto en el cliente como en el broker
Mejora las transacciones por segundo (TPS) en el envío y la recepción de mensajes
Notas de uso
No cree una instancia de producer o consumer para cada hilo. Cree una sola instancia y compártala entre los hilos.
Evite enviar mensajes ordenados desde varios hilos. El broker determina el orden de los mensajes según la secuencia en que los recibe de un único producer. Cuando varios hilos envían mensajes de forma concurrente, el orden de llegada al broker puede diferir del orden de envío definido en la lógica de su aplicación.
Producer compartido frente a producer por hilo
Correcto: cree el producer una sola vez y compártalo entre los hilos:
Producer producer = ONSFactory.createProducer(properties);
producer.start();
// Thread 1 and Thread 2 share the same producer
Thread t1 = new Thread(() -> producer.send(msg1));
Thread t2 = new Thread(() -> producer.send(msg2));
t1.start();
t2.start();Incorrecto: crear un producer por cada hilo:
// Each thread creates its own producer -- wastes resources
Thread t1 = new Thread(() -> {
Producer p = ONSFactory.createProducer(properties);
p.start();
p.send(msg);
p.shutdown();
});Código de ejemplo
A continuación se muestra un ejemplo donde se crea un producer compartido y se envían mensajes desde dos hilos de forma concurrente.
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) {
Properties properties = new Properties();
// The ID of the consumer group created in the ApsaraMQ for RocketMQ console.
properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
// Make sure the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
// and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
// Message send timeout in milliseconds.
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
// TCP endpoint. Get this value from the TCP endpoint section on the Instance Details
// page in the ApsaraMQ for RocketMQ console.
properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
final Producer producer = ONSFactory.createProducer(properties);
// Call start() once before sending any messages.
producer.start();
// Both threads share the same producer instance.
Thread thread1 = new Thread(() -> {
try {
Message msg = new Message(
"TopicTestMQ", // Topic for normal messages
"TagA", // Message tag for consumer-side filtering
"Hello MQ".getBytes() // Message body in binary format
);
SendResult result = producer.send(msg);
if (result != null) {
System.out.println("Sent successfully. Message ID: " + result.getMessageId());
}
} catch (Exception e) {
// Handle failures: retry or persist the message for later processing.
e.printStackTrace();
}
});
Thread thread2 = new Thread(() -> {
try {
Message msg = new Message("TopicTestMQ", "TagA", "Hello MQ".getBytes());
SendResult result = producer.send(msg);
if (result != null) {
System.out.println("Sent successfully. Message ID: " + result.getMessageId());
}
} catch (Exception e) {
e.printStackTrace();
}
});
thread1.start();
thread2.start();
// Call shutdown() when the producer is no longer needed to release resources.
// producer.shutdown();
}
}Reemplace los siguientes marcadores de posición con sus valores reales:
|
Marcador de posición |
Descripción |
Dónde encontrarlo |
|
|
ID del grupo de consumidores |
Consola de ApsaraMQ for RocketMQ |
|
|
Endpoint TCP |
Sección de endpoint TCP en la página Instance Details |
Puntos clave:
El método
send()utiliza transmisión síncrona. Si no se lanza ninguna excepción, el mensaje se envía correctamente.Una etiqueta de mensaje es un identificador (similar a las etiquetas de Gmail) que los consumidores utilizan para filtrar mensajes en el broker.
El cuerpo del mensaje consiste en datos binarios. ApsaraMQ for RocketMQ no procesa los cuerpos de los mensajes, por lo que el producer y el consumer deben acordar los métodos de serialización y deserialización.
Un topic utilizado para mensajes normales no puede usarse para otros tipos de mensajes.
Escalamiento más allá de una sola instancia
Para obtener un mayor rendimiento, despliegue varias instancias de producer o consumer en uno o más brokers. Cada instancia se comparte entre un pool de hilos; no cree una instancia por hilo, sin importar la escala.
|
Escala |
Enfoque |
|
Pocos hilos |
Cree hilos dedicados que compartan un solo producer (como se mostró anteriormente) |
|
Muchos emisores concurrentes |
Utilice un pool de hilos que comparta un solo producer |
|
Rendimiento muy alto |
Despliegue varias instancias de producer, cada una compartida por un pool de hilos |