Linux安装RocketMQ及springboot2.x简单示例

安装篇

下载地址: https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.2.0/rocketmq-all-4.2.0-source-release.zip
解压、编译

unzip rocketmq-all-4.2.0-source-release.zip
cd rocketmq-all-4.2.0/
mvn -Prelease-all -DskipTests clean install -U

/data/rocketmq-all-4.2.0/distribution/target/apache-rocketmq

启动 NameServer并查看日志

nohup sh bin/mqnamesrv -n 172.16.4.26:9876 &
tail -f ~/logs/rocketmqlogs/namesrv.log

启动 Broker并查看日志

nohup sh bin/mqbroker -n 172.16.4.26:9876 -c conf/broker.conf autoCreateTopicEnable=true &
tail -f ~/logs/rocketmqlogs/broker.log

运维相关命令

sh bin/mqshutdown broker    // 停止 broker
sh bin/mqshutdown namesrv   // 停止 nameserver
sh bin/mqbroker -m  // 查看broker参数

查看集群情况 ./mqadmin clusterList -n 127.0.0.1:9876
查看 broker 状态 ./mqadmin brokerStatus -n 127.0.0.1:9876 -b 172.20.1.138:10911 (注意换成你的 broker 地址)
查看 topic 列表 ./mqadmin topicList -n 127.0.0.1:9876
查看 topic 状态 ./mqadmin topicStatus -n 127.0.0.1:9876 -t MyTopic (换成你想查询的 topic)
查看 topic 路由 ./mqadmin topicRoute -n 127.0.0.1:9876 -t MyTopic

还需要注意一点的是要开启9876,10911这两个端口

客户端示例篇(以springboot 2.x整合rocketMQ为示例)

添加maven依赖

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

生产者示例

@Slf4j
@RestController
public class ScheduledMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @RequestMapping(value = "send")
    public String send() {
        String msg = "呵呵呵我是消息体";
        String status = "";

        try{
            //普通消息发送
            rocketMQTemplate.convertAndSend("delay-message-test", msg);
            Message<String> message = new GenericMessage<>(msg);
            //延时消息发送，level对应的默认延时时间依次为1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
            rocketMQTemplate.syncSend("delay-message-test", message, 10000, 3);
            log.info("消息已发送：{}", msg);
            status="消息发送成功";
        }
        catch (Exception ex){
            log.error("发送失败:" + ex.getMessage(), ex);
            status="消息发送失败";
        }
        return status;
    }
}

消费者示例

@Slf4j
@Component
@RocketMQMessageListener(topic = "delay-message-test", consumerGroup = "xht-group")
public class ScheduledMessageConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        log.info("received message: {}", message);
    }
}

问题解决

遇到There is insufficient memory for the Java Runtime Environment to continue问题时的解决办法：
修改/distribution/target/apache-rocketmq/bin 下的 3 个配置文件： runserver.sh、runbroker.sh 、tools.sh
调整jvm为合适大小，因为默认初始值比较大

最后编辑于：2019.08.15 11:48:26