ApsaraMQ for RocketMQ:Envie e receba mensagens com o SDK para Java

Consumidor push

Crie um arquivo PushConsumerExample.java e execute-o:

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // Uncomment for serverless instances accessed over the Internet:
                // .setNamespace("<your-instance-id>")
                // Uncomment for Internet access or serverless VPC without authentication-free:
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // Subscribe to all messages in the topic (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // Keep the consumer running
        Thread.sleep(Long.MAX_VALUE);

        // To shut down the consumer gracefully, call:
        // pushConsumer.close();
    }
}

Para lógicas de negócio de longa duração, use um consumidor simples (modo pull). O consumidor simples permite controlar a quantidade exata de mensagens puxadas por vez. Além disso, puxe o próximo lote apenas após processar e confirmar o lote atual. Defina InvisibleDuration (o período durante o qual uma mensagem recebida permanece invisível para outros consumidores, de 10 segundos a 12 horas) para controlar o tempo limite de confirmação e estendê-lo dinamicamente, se necessário.

Consumidor simples

Crie um arquivo SimpleConsumerExample.java e execute-o:

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // Uncomment for serverless instances accessed over the Internet:
                // .setNamespace("<your-instance-id>")
                // Uncomment for Internet access or serverless VPC without authentication-free:
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // Long polling timeout
        Duration awaitDuration = Duration.ofSeconds(10);
        // Subscribe to all messages in the topic (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();

        // Maximum messages per pull
        int maxMessageNum = 16;
        // Messages stay invisible to other consumers for this duration after being received
        Duration invisibleDuration = Duration.ofSeconds(10);

        // Poll for messages in a loop. For real-time consumption, use multiple threads.
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                System.out.println("Received message: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    // ACK each message to commit the consumption result to the broker
                    consumer.ack(message);
                    System.out.println("Message is acknowledged successfully, messageId= " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        // To shut down the consumer gracefully, call:
        // consumer.close();
    }
}

SDK para Java 5.x (rocketmq-client-java)

Versão mínima: 5.0.6

Defina o namespace em ClientConfiguration:

ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    .setEndpoints(endpoints)
    .setNamespace("InstanceId")
    .setCredentialProvider(sessionCredentialsProvider)
    .build();

SDK para Java 5.x (rocketmq-client)

Versão mínima: 5.2.0

Defina o namespace no produtor e no consumidor, separadamente:

// Producer
producer.setNamespaceV2("InstanceId");

// Consumer
consumer.setNamespaceV2("InstanceId");

SDK de cliente TCP para Java 1.x

Versão mínima: 1.9.0.Final

Defina o namespace por meio das propriedades:

properties.setProperty(PropertyKeyConst.Namespace, "InstanceId");