tencent cloud

// 实例化消息生产者Producer
DefaultMQProducer producer = new DefaultMQProducer(
     namespace, 
     groupName,
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) 
     // ACL权限
);
// 设置NameServer的地址
producer.setNamesrvAddr(nameserver);
// 启动Producer实例
producer.start();

for (int i = 0; i < 10; i++) {
     // 创建消息实例，设置topic和消息内容
     Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
     // 发送消息
     SendResult sendResult = producer.send(msg);
     System.out.printf("%s%n", sendResult);
}

// 设置发送失败后不重试
producer.setRetryTimesWhenSendAsyncFailed(0);
// 设置发送消息的数量
int messageCount = 10;
final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
for (int i = 0; i < messageCount; i++) {
     try {
             final int index = i;
             // 创建消息实体，设置topic和消息内容
             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) {
                             // 消息发送成功逻辑
                             countDownLatch.countDown();
                             System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                     }

                     @Override
                     public void onException(Throwable e) {
                             // 消息发送失败逻辑
                             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++) {
     // 创建消息实例，设置topic和消息内容
     Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
     // 发送单向消息
     producer.sendOneway(msg);
}

// 实例化消费者
DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
     namespace,                                                  
     groupName,                                              
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL权限
// 设置NameServer的地址
pushConsumer.setNamesrvAddr(nameserver);

// 实例化消费者
DefaultLitePullConsumer pullConsumer = new DefaultLitePullConsumer(
     namespace,                                               
     groupName,                                             
     new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
// 设置NameServer的地址
pullConsumer.setNamesrvAddr(nameserver);
// 设置从第一个偏移量开始消费
pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

// 订阅topic
pushConsumer.subscribe(topic_name, "*");
// 注册回调实现类来处理从broker拉取回来的消息
pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
     // 消息处理逻辑
     System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
     // 标记该消息已经被成功消费, 根据消费情况，返回处理状态
     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 启动消费者实例
pushConsumer.start();

// 订阅topic
pullConsumer.subscribe(topic_name, "*");
// 启动消费者实例
pullConsumer.start();
try {
     System.out.printf("Consumer Started.%n");
     while (true) {
             // 拉取消息
             List<MessageExt> messageExts = pullConsumer.poll();
             System.out.printf("%s%n", messageExts);
     }
} finally {
     pullConsumer.shutdown();
}