2017年11月聚合文章--RocketMQ快速入门

前面几篇文章介绍了为什么选择RocketMQ，以及与kafka的一些对比：阿里 RocketMQ 优势对比，方便大家对于RocketMQ有一个简单的整体了解，之后介绍了：MQ 应用场景，让我们知道MQ在什么时候可以使用，可以解决什么问题，之后介绍了：RocketMQ集群部署配置；本篇文章接着上篇内容之后，来给大家介绍下RocketMQ快速入门。

如何使用

1、引入 rocketmq-client

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

2、编写Producer

 DefaultMQProducer producer = new DefaultMQProducer("producer_demo");       //指定NameServer地址        producer.setNamesrvAddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的        /**         * Producer对象在使用之前必须要调用start初始化，初始化一次即可         * 注意：切记不可以在每次发送消息时，都调用start方法         */        producer.start();        for (int i = 0; i < 997892; i++) {            try {                //构建消息                Message msg = new Message("TopicTest" /* Topic */,                    "TagA" /* Tag */,                    ("测试RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)                );                //发送同步消息                SendResult sendResult = producer.send(msg);                System.out.printf("%s%n", sendResult);            } catch (Exception e) {                e.printStackTrace();                Thread.sleep(1000);            }        }producer.shutdown();

3、编写Consumer

/** * Consumer Group,非常重要的概念，后续会慢慢补充 */DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_demo");//指定NameServer地址，多个地址以 ; 隔开consumer.setNamesrvAddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的/** * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 * 如果非第一次启动，那么按照上次消费的位置继续消费 */consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicTest", "*");consumer.registerMessageListener(new MessageListenerConcurrently() {    @Override    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,    ConsumeConcurrentlyContext context) {    try {        for(MessageExt msg:msgs){        String msgbody = new String(msg.getBody(), "utf-8");        System.out.println("  MessageBody: "+ msgbody);//输出消息内容        }    } catch (Exception e) {        e.printStackTrace();        return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试    }    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功    }});consumer.start();System.out.printf("Consumer Started.%n");

4、说明

各位根据自己的环境，修改NamesrvAddr的值，我的集群请参考：RocketMQ集群部署配置。稍后通过RocketMQ管控台就可以看到之前搭建的多Master多Slave模式，异步复制集群模式。

5、通过RocketMQ管控台

rocketmq-console-ng获取方式为：rocketmq-console-ng，之后通过mavne进行编译获取jar，命令如下：

mvn clean package -Dmaven.test.skip=truejava -jar target/rocketmq-console-ng-1.0.0.jar

得到rocketmq-console-ng-1.0.0.jar之后，找到rocketmq-console-ng-1.0.0.jar\BOOT-INF\classes\application.properties文件，根据自己的NamesrvAddr进行修改rocketmq.config.namesrvAddr的值。

直接启动：