tencent cloud

// Instantiate the message producers
DefaultMQProducer producer = new DefaultMQProducer(
     namespace, 
     groupName,
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) 
     // ACL permission
);
// Set the NameServer address
producer.setNamesrvAddr(nameserver);
// Start the producer instances
producer.start();

for (int i = 0; i < 10; i++) {
     // Create a message instance and set the topic and message content
     Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
     // Send the message
     SendResult sendResult = producer.send(msg);
     System.out.printf("%s%n", sendResult);
}

// Disable retry upon sending failures
producer.setRetryTimesWhenSendAsyncFailed(0);
// Set the number of messages to be sent
int messageCount = 10;
final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
for (int i = 0; i < messageCount; i++) {
     try {
             final int index = i;
             // Create a message instance and set the topic and message content
             Message msg = new Message(topic_name, "TAG", ("Hello rocketMq " + index).getBytes(RemotingHelper.DEFAULT_CHARSET));
             producer.send(msg, new SendCallback() {
                     @Override
                     public void onSuccess(SendResult sendResult) {
                             // Logic for message sending successes
                             countDownLatch.countDown();
                             System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                     }

                     @Override
                     public void onException(Throwable e) {
                             // Logic for message sending failures
                             countDownLatch.countDown();
                             System.out.printf("%-10d Exception %s %n", index, e);
                             e.printStackTrace();
                     }
             });
     } catch (Exception e){
             e.printStackTrace();
     }
}
countDownLatch.await(5, TimeUnit.SECONDS);

for (int i = 0; i < 10; i++) {
     // Create a message instance and set the topic and message content
     Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
     Send one-way messages
     producer.sendOneway(msg);
}

// Instantiate the consumer
DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
     namespace,                                                  
     groupName,                                              
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL permission
// Set the NameServer address
pushConsumer.setNamesrvAddr(nameserver);

// Instantiate the consumer
DefaultLitePullConsumer pullConsumer = new DefaultLitePullConsumer(
     namespace,                                               
     groupName,                                             
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
// Set the NameServer address
pullConsumer.setNamesrvAddr(nameserver);
// Specify the first offset as the start offset for consumption
pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

// Subscribe to a topic
pushConsumer.subscribe(topic_name, "*");
// Register a callback implementation class to process messages pulled from the broker
pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
     // Message processing logic
     System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
     // Mark the message as being successfully consumed and return the consumption status
     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// Start the consumer instance
pushConsumer.start();

// Subscribe to a topic
pullConsumer.subscribe(topic_name, "*");
// Start the consumer instance
pullConsumer.start();
try {
     System.out.printf("Consumer Started.%n");
     while (true) {
             // Pull the message
             List<MessageExt> messageExts = pullConsumer.poll();
             System.out.printf("%s%n", messageExts);
     }
} finally {
     pullConsumer.shutdown();
}