rocketmq——HelloWorld

正常情况先启动Consumer再启动Producer（先订阅再发送）,否则消费消息的顺序会错乱。

依赖包：

已搭建好了broker和nameserver服务器

Producer类：

public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("quickstart_producer");
        producer.setNamesrvAddr("192.168.0.2:9876;192.168.0.3:9876");
        //消息发送失败重试10次
        producer.setRetryTimesWhenSendFailed(10);
        producer.start();
        for(int i=0; i<20; i++) {
            Message msg = new Message("TopicQuickStart","TagA",("Hello " + i).getBytes());
            try {
                SendResult sendResult = producer.send(msg,2000);//2秒钟内未发送成功自动重试
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        producer.shutdown();
    }

Consumer类：

public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("quickstart_consumer");
        //设置consumer第一次启动是从队列头部开始还是尾部开始消费，若非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicQuickStart", "*");
        //批量消费，一次消费多少条消息，默认为1条,最大情况能拿多少条不代表每次能拿这么多条
        //consumer.setConsumeMessageBatchMaxSize(3);
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //System.out.println(Thread.currentThread().getName() + "Receive: " + msgs);
                //获取一次性消费多少条消息
                //System.out.println("消息条数 ： " + msgs.size());
                MessageExt msg1 = null;
                try {
                    for(MessageExt msg : msgs) {
                        msg1 = msg;
                        String topic = msg.getTopic();
                        String msgbody = new String(msg.getBody(),"utf-8");
                        String tag = msg.getTags();
                        System.out.println("收到消息： " + "topic:" + topic + " tags:" + tag + " msg:" + msgbody);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    //若已经重试了5次则不再重试
                    if(msg1.getReconsumeTimes() == 5) {
                        //此处记录日志操作。。。
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.setNamesrvAddr("192.168.0.2:9876;192.168.0.3:9876");
        consumer.start();
        System.out.println("Consumer started...");
    }

producer coonsole:

SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000C84, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=0], queueOffset=6]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000D09, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=1], queueOffset=6]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000D8E, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=2], queueOffset=6]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000E13, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=3], queueOffset=6]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000858, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=0], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F00000000000008DD, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=1], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000962, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=2], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F00000000000009E7, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000E98, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000F1D, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000000FA2, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=2], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000001028, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=3], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000A6C, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=0], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000AF2, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=1], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000B78, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8000300002A9F0000000000000BFE, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-b, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F00000000000010AE, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000001134, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F00000000000011BA, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=2], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8000200002A9F0000000000001240, messageQueue=MessageQueue [topic=TopicQuickStart, brokerName=broker-a, queueId=3], queueOffset=8]

consumer console:

Consumer started...
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 0
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 1
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 2
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 3
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 4
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 5
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 6
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 7
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 8
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 9
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 10
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 11
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 12
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 13
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 14
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 15
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 16
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 17
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 18
收到消息： topic:TopicQuickStart tags:TagA msg:Hello 19

访问控制台：

broker集群和consumer端集群自动支持负载均衡，即同一个group的broker和订阅同一个主题的consumer。

Consumer端重试机制

1.timeout:
MQ发送到Consumer失败，没有返回值，重试次数是无限制的，会给Consumer集群轮着重新发送。

2.exception:
MQ发送到Consumer报异常，返回值为RECONSUME_LATER，
MQ重试机制为每次发送时间间隔依次递增，如：1s,2s,5s,10s….,2h，直到重试一定次数或发送成功。

需要在消费端考虑除去重复的消息，可以给每个消息唯一的key标识，在消费时判断key是否已插入数据库。