public class RocketMQProducer {
    public static void main(String[] args) throws MQClientException {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("group");
        // 指定NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动生产者
        producer.start();

        try {
            // 创建消息实例
            Message message = new Message("topic", "tag", "Hello RocketMQ".getBytes());
            // 发送消息
            SendResult result = producer.send(message);
            System.out.println(result);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.shutdown();
        }
    }
}

public class RocketMQConsumer {
    public static void main(String[] args) throws MQClientException {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
        // 指定NameServer地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 订阅Topic和Tag
        consumer.subscribe("topic", "tag");
        // 注册消息监听器
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
            // 处理消息
            for (MessageExt msg : msgs) {
                System.out.println(new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动消费者
        consumer.start();
    }
}

// 对象实例化
DefaultMQProducer producer = new DefaultMQProducer("Group Name");
producer.setNamesrvAddr("IP Address:Port");
producer.start();
// 创建消息实例
Message message = new Message("Topic Name","Tag Name","Message Body".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 发送消息
SendResult result = producer.send(message);
// 关闭生产端实例
producer.shutdown();

// 创建消费者实例
DefaultMQConsumer consumer = new DefaultMQConsumer("Group Name");
consumer.setNamesrvAddr("IP Address:Port");
consumer.subscribe("Topic Name","Tag Name");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for(MessageExt message:msgs){
        System.out.println(new String(message.getBody()));
    }
});
// 启动消费者实例
consumer.start();