十分鐘入門RcoketMQ http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/
基本上是有個(gè)初步概念备蚓，具體細(xì)節(jié)啥也沒有，可以看看了解一下模式鸳君。
阿里云java操作 https://www.aliyun.com/jiaocheng/807813.html?spm=5176.100033.2.21.HIouWL
apache官方rocketmq http://rocketmq.apache.org/
簡書分析 http://www.reibang.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的啟動

進(jìn)入rocketMQ解壓目錄下的bin文件夾
啟動namesrv服務(wù)：nohup sh bin/mqnamesrv &
日志目錄：{rocketMQ解壓目錄}/logs/rocketmqlogs/namesrv.log

啟動broker服務(wù)：nohup sh bin/mqbroker &
日志目錄：{rocketMQ解壓目錄}/logs/rocketmqlogs/broker.log

以上的啟動日志可以在啟動目錄下的nohub.out中看到

2吃引、rocketmq服務(wù)關(guān)閉

關(guān)閉namesrv服務(wù)：sh bin/mqshutdown namesrv
關(guān)閉broker服務(wù) ：sh bin/mqshutdown broker

三個(gè)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>

生產(chǎn)者發(fā)布消息經(jīng)典報(bào)錯

解決：No route info of this topic, TopicTest

需要在啟動brokersrv的時(shí)候更改啟動命令為

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

指定自動創(chuàng)建新的Topic

SpringBoot起步依賴

官方?jīng)]有提供起步依賴,不過同性交友網(wǎng)站上有很多封裝的。
https://github.com/maihaoche/rocketmq-spring-boot-starter
這個(gè)封裝的還是挺方便的
application.yml

server:
  port: 9091
spring:
    rocketmq:
      name-server-address: localhost:9876
      # 可選, 如果無需發(fā)送消息則忽略該配置
      producer-group: local_pufang_producer
      # 發(fā)送超時(shí)配置毫秒數(shù), 可選, 默認(rèn)3000
      send-msg-timeout: 3000
      # 追溯消息具體消費(fèi)情況的開關(guān)，默認(rèn)打開
      trace-enabled: true
      # 是否啟用VIP通道期升，默認(rèn)打開
      vip-channel-enabled: false

消費(fèi)端和生產(chǎn)端都要配置
啟動類 配置config注解

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

生產(chǎn)者bean

@MQProducer
public class StarterProducer extends AbstractMQProducer {
}

mvc觸發(fā)調(diào)用

@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();
    }
}

消費(fèi)端過濾監(jiān)聽

@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;
    }
}