Tips ： When the article is finished , Directories can be generated automatically , How to generate it, please refer to the help document on the right

Preface

Tips ： The following is the main body of this article , The following cases can be used for reference

One 、 Send a normal message

Alicloud message queue RocketMQ Version provides three ways to send ordinary messages ： The synchronous 、 Asynchronous transmission and one-way transmission （Oneway） send out . This article only introduces synchronous transmission , This method has a wide range of application scenarios , For example, important notice email 、 Sign up for SMS notification 、 Marketing SMS system, etc ..

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 {
    
        /** *  Replace with your alicloud account AccessKey ID and 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 {
    
        /** * establish Producer, And turn on the message track . It is set to you in Alibaba cloud message queue RocketMQ Created by version console Group ID. * If you don't want to turn on the message track , You can create ： *DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); */
        DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null);
        /** * Set the access method to Alibaba cloud , When using message tracks on the cloud , You need to set this , If you do not turn on the message trace function , This item is not set when running . */
        producer.setAccessChannel(AccessChannel.CLOUD);
        /** * It is set to you from Alibaba cloud message queue RocketMQ The access point information obtained from the console , similar “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) {
    
                // Message delivery failed , Need to retry processing , You can resend the message or persist the data for compensation processing .
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Before the app exits , The destruction Producer object .
        // Be careful ： If you don't destroy it, there's no problem .
        producer.shutdown();
    }
}

Two 、 Subscribe to general news

There is only one way to subscribe to ordinary messages

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 {
    
       /** *  Replace with your alicloud account AccessKey ID and 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 {
    
        /** *  establish Consumer, And turn on the message track . It is set to you in Alibaba cloud message queue RocketMQ Created by version console Group ID. *  If you don't want to turn on the message track , You can create ： * DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely()); */ 
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
        // Set as alicloud message queue RocketMQ Access point of version instance .
        consumer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");
        // The message track on Alibaba cloud needs to be set to CLOUD The way , When using message tracks on the cloud , You need to set this , If you do not turn on the message trace function , This item is not set when running .
        consumer.setAccessChannel(AccessChannel.CLOUD);
        //  It is set to you in Alibaba cloud message queue RocketMQ Created on the console 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();
    }
}