阿里RocketMQ安装服务器及客户端

# 1.上传服务器
rocketmq-all-4.4.0-bin-release.zip
# 2.解压
unzip rocketmq-all-4.4.0-bin-release.zip
# 3.移动到/usr/local
mv rocketmq-all-4.4.0-bin-release /usr/local/rocketmq

编辑bin/runbroker.sh 和 bin/runserver.sh文件,修改里面的 
# JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m" 

# 进入bin目录下
# 1.启动nameserver
nohup ./bin/mqnamesrv & 
# 2.启动broker
nohup ./mqbroker -n localhost:9876 &

<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.2</version>
</dependency>

public class RocketMQSendTest {
    public static void main(String[] args) throws Exception {
        //1. 创建消息生产者, 指定生产者所属的组名
        DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
        //2. 指定Nameserver地址
        producer.setNamesrvAddr("192.168.109.131:9876");
        //3. 启动生产者
        producer.start();
        //4. 创建消息对象，指定主题、标签和消息体
        Message msg = new Message("myTopic", "myTag",
        ("RocketMQ Message").getBytes());
        //5. 发送消息
        SendResult sendResult = producer.send(msg,10000); System.out.println(sendResult);
        //6. 关闭生产者
        producer.shutdown();
  }
}

//接收消息
public class RocketMQReceiveTest {
    public static void main(String[] args) throws MQClientException {
        //1. 创建消息消费者, 指定消费者所属的组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer- group");
        //2. 指定Nameserver地址
        consumer.setNamesrvAddr("192.168.109.131:9876");
        //3. 指定消费者订阅的主题和标签
        consumer.subscribe("myTopic", "*");
        //4. 设置回调函数，编写处理消息的方法
        consumer.registerMessageListener(new MessageListenerConcurrently() { @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
        msgs,
        ConsumeConcurrentlyContext
        context) {
        System.out.println("Receive New Messages: " + msgs);
        //返回消费状态
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        });
        //5. 启动消息消费者
        consumer.start(); System.out.println("Consumer Started.");
  }
}