通过控制台可以查看生产者、消费者、Broker集群等信息，非常直观。

功能很多，这里就不一一介绍了。

测试

环境搭好之后，就可以进行测试了。

引入依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>

生产者发送消息

    @Test
    public void sendTest() throws Exception{
        //创建一个生产者，指定生产者组为ldProducer
        DefaultMQProducer producer = new DefaultMQProducer("ldProducer");

        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.3.158:9876");
        // 第一次发送可能会超时，我设置的比较大
        producer.setSendMsgTimeout(60000);

        // 启动生产者
        producer.start();

        // 创建一条消息
        // topic为 ldTopic
        // 消息内容为 java学习日记
        // tags 为 TagA
        Message msg = new Message("ldTopic", "TagA", "java学习日记 ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 发送消息并得到消息的发送结果，然后打印
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        // 关闭生产者
        producer.shutdown();
    }

• 构建一个消息生产者DefaultMQProducer实例，然后指定生产者组为ldProducer；
• 指定NameServer的地址：服务器的ip:9876，因为需要从NameServer拉取Broker的信息
• producer.start() 启动生产者
• 构建一个内容为三友的java日记的消息，然后指定这个消息往ldTopic这个topic发送
• producer.send(msg)：发送消息，打印结果
• 关闭生产者

消费者消费消息

public class ConsumerMsg {
    public static void main(String[] args) throws Exception {
        // 通过push模式消费消息，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ldConsumer");
        consumer.setNamesrvAddr("192.168.3.158:9876");
        // 订阅这个topic下的所有的消息
        consumer.subscribe("ldTopic", "*");
        // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

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

• 创建一个消费者实例对象，指定消费者组为ldConsumer
• 指定NameServer的地址：服务器的ip:9876
• 订阅 ldTopic 这个topic的所有信息
• consumer.registerMessageListener ，这个很重要，是注册一个监听器，这个监听器是当有消息的时候就会回调这个监听器，处理消息，所以需要用户实现这个接口，然后处理消息。
• 启动消费者

启动之后，消费者就会消费刚才生产者发送的消息，于是控制台就打印出如下信息

再去看控制台，已消费

SpringBoot环境下集成RocketMQ

集成

在实际项目中肯定不会像上面测试那样用，都是集成SpringBoot的。

1、引入依赖

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

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <version>2.1.1.RELEASE</version>
</dependency>

2、yml配置

rocketmq:
  producer:
    group: ldProducer
  name-server: 192.168.3.158:9876

3、创建消费者

SpringBoot底下只需要实现RocketMQListener接口，然后加上@RocketMQMessageListener注解即可

@Component
@RocketMQMessageListener(consumerGroup = "ldConsumer", topic = "ldDelayTaskTopic")
@Slf4j
public class LdRocketMQListener implements RocketMQListener<String> {
    @Override
    public void onMessage(String msg) {
        log.info("获取到延迟任务消息:{}",msg);
    }
}

@RocketMQMessageListener需要指定消费者属于哪个消费者组，消费哪个topic，NameServer的地址已经通过yml配置文件配置类

4、测试

@RestController
@Slf4j
public class RocketMQDelayTaskController {

    @Resource
    private DefaultMQProducer producer;

    @GetMapping("/rocketmq/add")
    public void addTask(@RequestParam("task") String task) throws Exception {
        Message msg = new Message("ldDelayTaskTopic", "TagA", task.getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setDelayTimeLevel(2);
        // 发送消息并得到消息的发送结果，然后打印
        log.info("提交延迟任务");
        producer.send(msg);
    }

}

可能遇到的问题

搭完mq单主单从集群之后，美滋滋想发一下message, 没想到碰到一个坑爹的问题：

Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 14  DESC: service not available now, maybe disk full, CL:  0.90 CQ:  0.90 INDEX:  0.90, maybe your broker machine memory too small.

org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [549]ms, Topic: ldTopicA, BrokersSent: [broker-a, broker-a, broker-a]
See http://rocketmq.apache.org/docs/faq/ for further details.

    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:665)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1343)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1289)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:325)
    at com.example.delay.MQTest.sendTest(MQTest.java:46)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    ...
Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 14  DESC: service not available now, maybe disk full, CL:  0.90 CQ:  0.90 INDEX:  0.90, maybe your broker machine memory too small.
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.processSendResponse(MQClientAPIImpl.java:665)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageSync(MQClientAPIImpl.java:505)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:487)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:431)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendKernelImpl(DefaultMQProducerImpl.java:854)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:584)

看报错应该是磁盘空间不足的问题，看到一个帖子https://bbs.csdn.net/topics/392568834，还挺符合的，虽然给出的解决方案说的没那么详细，但是值得一试。

查看磁盘空间

已用91%，查阅百度之后发现rocketmq源码的DefaultMessageStore类里，默认会把剩余磁盘的比率不足75%(rocketmq版本不同这个比率好像不一样)当做磁盘空间不足处理，看来磁盘是有点不够了。

先cd到rocketmq配置文件的路径，我这里配置的是双主双从同步的模式，所以cd到配置文件（根据配置的不同文件夹的路径不一样，但都在/conf下）。

1. cd rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/
2. vim broker-a.properties
3. 在最后加一行diskMaxUsedSpaceRatio=99(所有节点的配置文件都加一下)，表示剩余磁盘比例不足99才报错

4. :wq 保存退出

5. 重启mq

 重新发送消息Ok了