安裝

去官網下載rocketmq：https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.4.0/rocketmq-all-4.4.0-bin-release.zip，上傳到linux中

解壓

[root@suzhe opt]# unzip rocketmq-all-4.4.0-bin-release.zip

調整runserver.sh jvm引數，這裡我根據自己的機器情況調整成如下的大小

[root@suzhe bin]# vim runserver.sh 

-Xms1g 最小堆的大小， 也就是當你的虛擬機器啟動後， 就會分配這麼大的堆記憶體給你 

-Xmx1g 為jvm執行過程中分配的最大記憶體

-Xmn512m 設定新生代的初始值及最大值。

也需要調整runbroker.sh的jvm引數

[root@suzhe bin]# vim runbroker.sh 

啟動nameServer

 nohup sh bin/mqnamesrv &

修改配置 新增namesrvAddr和brokerIP1（外網可以訪問）

[root@suzhe rocketmq-all-4.4.0-bin-release]# vim conf/broker.conf 

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
namesrvAddr=118.89.196.99:9876
brokerIP1=118.89.196.99

啟動broker：將{ip}替換成你的ip

sh bin/mqbroker -n {ip}:9876 autoCreateTopicEnable=true -c /opt/rocketmq-all-4.4.0-bin-release/conf/broker.conf &

java程式

消費者：

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");

        // Specify name server addresses.
        consumer.setNamesrvAddr("{ip}:9876");

        // Subscribe one more more topics to consume.
        consumer.subscribe("TopicTest", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
               msgs.forEach(item ->{
                   System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(item.getBody()));
               });
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //Launch the consumer instance.
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

生產者同步傳送

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("118.89.196.99:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 5; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}