提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

前言

提示：以下是本篇文章正文内容，下面案例可供参考

一、发送普通消息

阿里云消息队列RocketMQ版提供三种方式来发送普通消息：同步发送、异步发送和单向（Oneway）发送。本文仅介绍同步发送，此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。。

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 {
    
        /** * 替换为您阿里云账号的AccessKey ID和AccessKey Secret。 */ 
    private static RPCHook getAclRPCHook() {
    
        return new AclClientRPCHook(new SessionCredentials("YOUR ACCESS KEY", "YOUR SECRET KEY"));
    }

    public static void main(String[] args) throws MQClientException {
    
        /** *创建Producer，并开启消息轨迹。设置为您在阿里云消息队列RocketMQ版控制台创建的Group ID。 *如果不想开启消息轨迹，可以按照如下方式创建： *DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); */
        DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);
        /** *设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。 */
        producer.setAccessChannel(AccessChannel.CLOUD);
        /** *设置为您从阿里云消息队列RocketMQ版控制台获取的接入点信息，类似“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) {
    
                //消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        //在应用退出前，销毁Producer对象。
        //注意：如果不销毁也没有问题。
        producer.shutdown();
    }
}

二、订阅普通消息

订阅普通消息的方式仅此一种

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 {
    
       /** * 替换为您阿里云账号的AccessKey ID和AccessKey Secret。 */
    private static RPCHook getAclRPCHook() {
    
        return new AclClientRPCHook(new SessionCredentials("YOUR ACCESS KEY", "YOUR SECRET KEY"));
    }
    public static void main(String[] args) throws MQClientException {
    
        /** * 创建Consumer，并开启消息轨迹。设置为您在阿里云消息队列RocketMQ版控制台创建的Group ID。 * 如果不想开启消息轨迹，可以按照如下方式创建： * DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely()); */ 
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
        //设置为阿里云消息队列RocketMQ版实例的接入点。
        consumer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");
        //阿里云上消息轨迹需要设置为CLOUD方式，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
        consumer.setAccessChannel(AccessChannel.CLOUD);
        // 设置为您在阿里云消息队列RocketMQ版控制台上创建的Topic。
        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();
    }
}