Simple Message Queue (formerly MNS):Distribución de mensajes de un productor a varios consumidores

<dependencies>
    <dependency>
        <groupId>com.aliyun.mns</groupId>
        <artifactId>aliyun-sdk-mns</artifactId>
        <version>1.1.9.2</version>
    </dependency>
</dependencies>

// Full form: create queues automatically with a queue metadata template
public CloudPullTopic createPullTopic(
    TopicMeta topicMeta,
    Vector<String> queueNameList,
    boolean needCreateQueue,
    QueueMeta queueMetaTemplate
)

// Short form: use existing queues (no automatic creation)
public CloudPullTopic createPullTopic(
    TopicMeta topicMeta,
    Vector<String> queueNameList
)

package doc;

import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.client.CloudPullTopic;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.model.QueueMeta;
import com.aliyun.mns.model.TopicMeta;
import com.aliyun.mns.model.TopicMessage;
import com.aliyun.mns.model.RawTopicMessage;
import com.aliyun.mns.model.Message;
import com.aliyun.mns.common.ServiceException;
import com.aliyun.mns.common.ClientException;
import java.util.Vector;

public class DemoTopicMessageBroadcast {
    public static void main(String[] args) {

        // Initialize the client with your AccessKey ID, AccessKey Secret, and endpoint.
        CloudAccount account = new CloudAccount(
            ServiceSettings.getMNSAccessKeyId(),
            ServiceSettings.getMNSAccessKeySecret(),
            ServiceSettings.getMNSAccountEndpoint());
        MNSClient client = account.getMNSClient();

        // Define three consumer queues.
        Vector<String> consumerNameList = new Vector<String>();
        String consumerName1 = "consumer001";
        String consumerName2 = "consumer002";
        String consumerName3 = "consumer003";
        consumerNameList.add(consumerName1);
        consumerNameList.add(consumerName2);
        consumerNameList.add(consumerName3);

        // Set long polling to 30 seconds for the auto-created queues.
        QueueMeta queueMetaTemplate = new QueueMeta();
        queueMetaTemplate.setPollingWaitSeconds(30);

        try {
            // --- Producer: create a topic and publish a message ---

            String topicName = "demo-topic-for-pull";
            TopicMeta topicMeta = new TopicMeta();
            topicMeta.setTopicName(topicName);

            // createPullTopic creates the topic, creates the three consumer queues,
            // and subscribes each queue to the topic — all in one call.
            CloudPullTopic pullTopic = client.createPullTopic(
                topicMeta, consumerNameList, true, queueMetaTemplate);

            // Publish a raw message to the topic. SMQ copies it into all subscribed queues.
            String messageBody = "broadcast message to all the consumers:hello the world.";
            TopicMessage tMessage = new RawTopicMessage();
            tMessage.setBaseMessageBody(messageBody);
            pullTopic.publishMessage(tMessage);

            // --- Consumers: pull the message from each queue ---

            CloudQueue queueForConsumer1 = client.getQueueRef(consumerName1);
            CloudQueue queueForConsumer2 = client.getQueueRef(consumerName2);
            CloudQueue queueForConsumer3 = client.getQueueRef(consumerName3);

            // popMessage(30) waits up to 30 seconds for a message (long polling).
            Message consumer1Msg = queueForConsumer1.popMessage(30);
            if (consumer1Msg != null) {
                System.out.println("consumer1 received: "
                    + consumer1Msg.getMessageBodyAsRawString());
            } else {
                System.out.println("consumer1 queue is empty");
            }

            Message consumer2Msg = queueForConsumer2.popMessage(30);
            if (consumer2Msg != null) {
                System.out.println("consumer2 received: "
                    + consumer2Msg.getMessageBodyAsRawString());
            } else {
                System.out.println("consumer2 queue is empty");
            }

            Message consumer3Msg = queueForConsumer3.popMessage(30);
            if (consumer3Msg != null) {
                System.out.println("consumer3 received: "
                    + consumer3Msg.getMessageBodyAsRawString());
            } else {
                System.out.println("consumer3 queue is empty");
            }

            // Clean up: delete the topic, its subscriptions, and all associated queues.
            pullTopic.delete();

        } catch (ClientException ce) {
            System.out.println("Network error between the client and SMQ. "
                + "Check your network and DNS settings.");
            ce.printStackTrace();
        } catch (ServiceException se) {
            se.printStackTrace();
        }

        client.close();
    }
}

Comprobación del resultado

Después de ejecutar el ejemplo, confirme que cada consumidor recibió el mensaje. Salida esperada:

consumer1 received: broadcast message to all the consumers:hello the world.
consumer2 received: broadcast message to all the consumers:hello the world.
consumer3 received: broadcast message to all the consumers:hello the world.

Si una cola muestra "queue is empty", es posible que el mensaje aún no se haya entregado. Aumente el tiempo de espera de popMessage o compruebe que el tema y las suscripciones se crearon correctamente.

Explicación del código

El ejemplo publica un mensaje sin procesar ( RawTopicMessage ). Analice sintácticamente los mensajes sin procesar con getMessageBodyAsRawString() . Si usa Base64TopicMessage en su lugar, analícelos con getMessageBodyAsString() .