分布式消息队列RocketMq入门安装和使用教程

下载与安装：

提示：下面以Mac为例，windows用户可参考。

1. 进入官网 点击Getting Started

2. 找到下载入口

3. 下载rocketmq

这里推荐下载binary包。不推荐下载source包，还得需要自己再编译一遍。

4. 下载rocketmq dashboard

在rocket mq 的github readme里有dashboard的下载链接，是source包，
下载后，设置一下rocketmq.config.namesrvAddr值，然后再通过mvn打成jar包。

启动RocketMq和RockerMq Dashboard： 1.启动nameServer和Broker

Mac上启动nameServer的命令如下

nohup  ./bin/mqnamesrv &

Mac上启动Broker的命令如下

nohup ./bin/mqbroker -n localhost:9876 &

看nameServer和Broker是否启动成功, 输入Jps命令查看进程NameSrvStartUp和BrokerStartup是否存在即可。

2.启动rocketmq dashboard

启动dashboard的命令如下

java -jar xx-dashborad.jar

生产者消费者Demo：

在搭建好nameServer和Broker后,即搭建好了消息存储的平台，所以现在需要生产者生产消息，消费者消费消息。下面以IDEA展示。

1. 新建Maven工程，引入依赖

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

2. 生产者生产消息

import org.apache.rocketmq.client.exception.MQBrokerException;
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.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

public class Producer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("Producer_ycx");
        // 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 10; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagC" /* Tag */,
                    ("Hello ycx " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
        }
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }
}

3. 消费者消费消息

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.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Consumer_ycx");

        // 设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TopicTest", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                for (int i = 0; i <msgs.size() ; i++) {
                    System.out.println("Receive New Message: "+new String(msgs.get(i).getBody()));
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

更多的demo可参考官方文档