Envío de mensajes normales en modo síncrono, asíncrono o unidireccional - ApsaraMQ for RocketMQ

ApsaraMQ for RocketMQ admite tres modos de transmisión para mensajes normales. Cada modo ofrece un equilibrio diferente entre la confiabilidad de entrega y el throughput.

Modo	Respuesta del broker	Confiabilidad	Throughput	Recomendado para
Síncrono	Bloquea hasta recibir confirmación	Sin pérdida de mensajes	Alto	Notificaciones por correo electrónico, confirmaciones de registro, mensajes promocionales
Asíncrono	Se entrega mediante callback	Sin pérdida de mensajes	Alto	Flujos de trabajo sensibles al tiempo de respuesta, pipelines de transcodificación de video
Unidireccional	Ninguna	Posible pérdida de mensajes	Máximo	Recolección de registros

Los tres modos comparten la misma configuración del productor. La única diferencia radica en la llamada de envío:

Síncrono: producer.send(msg) -- bloquea la ejecución y devuelve SendResult
Asíncrono: producer.send(msg, callback) -- devuelve el control de inmediato; el resultado se entrega mediante SendCallback
Unidireccional: producer.sendOneway(msg) -- envía sin esperar respuesta (fire-and-forget), sin valor de retorno

Requisitos previos

Community Edition del SDK para Java 4.5.2 o posterior
Environment prepared
AccessKey pair created para su cuenta de Alibaba Cloud

Configuración común del productor

La siguiente inicialización del productor aplica a los tres modos. Reemplace los marcadores de posición con sus valores:

Marcador de posición	Descripción	Ejemplo
`YOUR GROUP ID`	Group ID creado en la consola de ApsaraMQ for RocketMQ	GID_example
`YOUR ACCESS POINT`	Endpoint obtenido de la consola de ApsaraMQ for RocketMQ	`http://MQ_INST_XXXX.aliyuncs.com:80`
`YOUR TOPIC`	Topic creado en la consola de ApsaraMQ for RocketMQ	topic_example
`YOUR MESSAGE TAG`	Etiqueta para el filtrado de mensajes	tag_example

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

// Retrieve credentials from environment variables
private static RPCHook getAclRPCHook() {
    return new AclClientRPCHook(new SessionCredentials(
        System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
        System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
}

// Create a producer with message trace enabled.
// To disable message trace, use: new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);

// Set the access channel to CLOUD for message trace on Alibaba Cloud.
// Skip this line if message trace is not needed.
producer.setAccessChannel(AccessChannel.CLOUD);

// Set the endpoint from the ApsaraMQ for RocketMQ console.
producer.setNamesrvAddr("YOUR ACCESS POINT");

producer.start();

Transmisión síncrona

El productor envía un mensaje y bloquea la ejecución hasta que el broker devuelve una respuesta. Esto garantiza la confirmación de entrega antes de enviar el siguiente mensaje.

Synchronous transmission

Equilibrio: Es confiable, ya que no se pierden mensajes. Sin embargo, el throughput es menor que en el modo asíncrono porque el productor espera cada respuesta antes de enviar el siguiente mensaje.

Casos de uso: Notificaciones por correo electrónico, confirmaciones de registro y mensajes promocionales; es decir, escenarios donde se requiere confirmación de entrega antes de continuar.

Código de ejemplo

import java.util.Date;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class RocketMQProducer {
    /**
    * Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
    * and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
    */
    private static RPCHook getAclRPCHook() {
	  return new AclClientRPCHook(new SessionCredentials(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a producer with message trace enabled.
        // To disable message trace: new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
        DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);

        // Set to CLOUD for message trace on Alibaba Cloud. Skip if not needed.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Endpoint from the ApsaraMQ for RocketMQ console (format: http://MQ_INST_XXXX.aliyuncs.com:80).
        producer.setNamesrvAddr("YOUR ACCESS POINT");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("YOUR TOPIC",
                    "YOUR MESSAGE TAG",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                // Resend or persist the message if delivery fails.
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting the application. (Optional)
        producer.shutdown();
    }
}

producer.send(msg) bloquea la ejecución y devuelve un objeto SendResult que contiene el ID del mensaje y el estado del envío.

Transmisión asíncrona

El productor envía un mensaje y continúa inmediatamente sin esperar la respuesta del broker. Implemente un SendCallback para gestionar la respuesta del broker de forma asíncrona.

Asynchronous transmission

Equilibrio: Ofrece la misma confiabilidad que el modo síncrono (sin pérdida de mensajes), pero el productor no se bloquea entre envíos. La complejidad adicional reside en implementar SendCallback para manejar los casos de éxito y fallo.

Casos de uso: Flujos de trabajo sensibles al tiempo de respuesta con procesos posteriores de larga duración. Por ejemplo, después de subir un video, un callback activa la transcodificación y otro callback envía el resultado.

Código de ejemplo

import java.util.Date;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class RocketMQAsyncProducer {
    /**
    * Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
    * and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
    */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a producer with message trace enabled.
        // To disable message trace: new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
        DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);

        // Set to CLOUD for message trace on Alibaba Cloud. Skip if not needed.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Endpoint from the ApsaraMQ for RocketMQ console (format: http://MQ_INST_XXXX.aliyuncs.com:80).
        producer.setNamesrvAddr("YOUR ACCESS POINT");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("YOUR TOPIC",
                    "YOUR MESSAGE TAG",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override public void onSuccess(SendResult result) {
                        // Message delivered successfully.
                        System.out.println("send message success. msgId= " + result.getMsgId());
                    }

                    @Override public void onException(Throwable throwable) {
                        // Resend or persist the message if delivery fails.
                        System.out.println("send message failed.");
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                // Resend or persist the message if delivery fails.
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting the application. (Optional)
        producer.shutdown();
    }
}

producer.send(msg, new SendCallback() {...}) no bloquea la ejecución. El resultado se entrega mediante onSuccess o onException.

Transmisión unidireccional

El productor envía un mensaje y retorna de inmediato. No se recibe respuesta del broker ni se activa ningún callback. Un mensaje puede enviarse en microsegundos.

单向发送

Equilibrio: Ofrece el mayor throughput de los tres modos, pero los mensajes pueden perderse porque el productor no confirma la entrega.

Casos de uso: Flujos de datos de alto volumen y baja criticidad donde la pérdida ocasional de mensajes es aceptable, como la recolección de registros.

Código de ejemplo

import java.util.Date;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class RocketMQOnewayProducer {
    /**
    * Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
    * and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
    */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a producer with message trace enabled.
        // To disable message trace: new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
        DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);

        // Set to CLOUD for message trace on Alibaba Cloud. Skip if not needed.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Endpoint from the ApsaraMQ for RocketMQ console (format: http://MQ_INST_XXXX.aliyuncs.com:80).
        producer.setNamesrvAddr("YOUR ACCESS POINT");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message("YOUR TOPIC",
                    "YOUR MESSAGE TAG",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.sendOneway(msg);
            } catch (Exception e) {
                // Resend or persist the message if delivery fails.
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting the application. (Optional)
        producer.shutdown();
    }
}

producer.sendOneway(msg) no devuelve ningún SendResult ni invoca callbacks; el mensaje se envía con base en el mejor esfuerzo posible.

Suscripción a mensajes normales

Independientemente del modo de transmisión usado para enviar mensajes, consúmalos con DefaultMQPushConsumer. El broker envía los mensajes al consumidor, que los procesa mediante un listener MessageListenerConcurrently.

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

public class RocketMQPushConsumer {
    /**
    * Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
    * and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
    */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }
  public static void main(String[] args) throws MQClientException {
        // Create a consumer with message trace enabled.
        // To disable message trace:
        // DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely());
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);

        // Endpoint of the ApsaraMQ for RocketMQ instance.
    consumer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");

        // Set to CLOUD for message trace on Alibaba Cloud. Skip if not needed.
    consumer.setAccessChannel(AccessChannel.CLOUD);

        // Topic created in the ApsaraMQ for RocketMQ console.
    consumer.subscribe("YOUR TOPIC", "*");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
        ConsumeConcurrentlyContext context) {
        System.out.printf("Receive New Messages: %s %n", msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    consumer.start();
  }
}