十分钟入门RcoketMQ http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/
基本上是有个初步概念，具体细节啥也没有，可以看看了解一下模式。
阿里云java操作 https://www.aliyun.com/jiaocheng/807813.html?spm=5176.100033.2.21.HIouWL
apache官方rocketmq http://rocketmq.apache.org/
简书分析 https://www.jianshu.com/p/fe8c89a781a3
rocketmq操作 https://www.cnblogs.com/gmq-sh/p/6232633.html
用户指南 https://wenku.baidu.com/view/bbae7400580216fc700afd6f.html
博客 http://valleylord.github.io/post/201607-mq-rocketmq/

启动与停止

1、rocketmq的启动

进入rocketMQ解压目录下的bin文件夹
启动namesrv服务：nohup sh bin/mqnamesrv &
日志目录：{rocketMQ解压目录}/logs/rocketmqlogs/namesrv.log

启动broker服务：nohup sh bin/mqbroker &
日志目录：{rocketMQ解压目录}/logs/rocketmqlogs/broker.log

以上的启动日志可以在启动目录下的nohub.out中看到

2、rocketmq服务关闭

关闭namesrv服务：sh bin/mqshutdown namesrv
关闭broker服务 ：sh bin/mqshutdown broker

三个producerdemo

public class MainProducerTest {
        public static void main(String[] args) throws Exception {
//            syncProducer();
//            asyncProducer();
            alibabaProducer();


        }
    private static void alibabaProducer() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
            final DefaultMQProducer producer = new DefaultMQProducer("myproducer");
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.setInstanceName("Producer");
            //初始化
            producer.start();
            for(int i = 0;i<10;i++) {
                try {
                    Message message = new Message("TopicTest1", "TagA", "helloMQ".getBytes());
                    SendResult sendResult = producer.send(message);
                    System.out.println(sendResult);
                } catch (MQClientException e) {
                    e.printStackTrace();
                } catch (RemotingException e) {
                    e.printStackTrace();
                } catch (MQBrokerException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                @Override
                public void run() {
                    producer.shutdown();
                }
            }));
            System.exit(0);
    }


    private static void syncProducer() throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        producer.setNamesrvAddr("127.0.0.1:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
    private static void asyncProducer() throws UnsupportedEncodingException, RemotingException, MQClientException, InterruptedException {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
        //Launch the instance.
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        for (int i = 0; i < 100; i++) {
            final int index = i;
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                }
                @Override
                public void onException(Throwable e) {
                    System.out.printf("%-10d Exception %s %n", index, e);
                    e.printStackTrace();
                }
            });
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

ConsumerDemo

public class MainConsumerTest {
    public static void main(String[] args) throws MQClientException {
        alibabaConsumer();
    }
    private static void alibabaConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumeer");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setInstanceName("Consumer");
        consumer.subscribe("TopicTest1","TagA");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            System.out.print("start to consum");
            System.out.println(Thread.currentThread().getName()
                    +" Receive New Messages: " + msgs.size());
            MessageExt msg = msgs.get(0);
            if (msg.getTopic().equals("TopicTest1")) {
                if (msg.getTags() != null && msg.getTags().equals("TagA")) {
                    System.out.println(msg.getTopic()+":"+msg.getTags()+":"+new String(msg.getBody()));
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("ConsumerStarted");
    }
}

涉及到的maven包

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

生产者发布消息经典报错

解决：No route info of this topic, TopicTest

需要在启动brokersrv的时候更改启动命令为

nohup sh bin/mqbroker -n 127.0.0.1:9876 autoCreateTopicEnable=true &

指定自动创建新的Topic

SpringBoot起步依赖

官方没有提供起步依赖,不过同性交友网站上有很多封装的。
https://github.com/maihaoche/rocketmq-spring-boot-starter
这个封装的还是挺方便的
application.yml

server:
  port: 9091
spring:
    rocketmq:
      name-server-address: localhost:9876
      # 可选, 如果无需发送消息则忽略该配置
      producer-group: local_pufang_producer
      # 发送超时配置毫秒数, 可选, 默认3000
      send-msg-timeout: 3000
      # 追溯消息具体消费情况的开关，默认打开
      trace-enabled: true
      # 是否启用VIP通道，默认打开
      vip-channel-enabled: false

消费端和生产端都要配置
启动类 配置config注解

@SpringBootApplication
@EnableMQConfiguration
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

生产者bean

@MQProducer
public class StarterProducer extends AbstractMQProducer {
}

mvc触发调用

@RestController
public class RocketMQController {
    @Autowired
    private StarterProducer starterProducer;

    @GetMapping("/test")
    public String producer(String params){
        Message message = MessageBuilder
                .of(params)
                .topic("some_msg_topic")
                .tag("TagA")
                .build();
        starterProducer.syncSend(message);
        return message.toString();
    }
}

消费端过滤监听

@MQConsumer(topic = "some_msg_topic",consumerGroup = "local_sucloger_dev",tag = {"TagA"})
public class StarterConsumer extends AbstractMQPushConsumer{
    @Override
    public boolean process(Object o, Map map) {
        System.out.println(o);
        return true;
    }
}