rocketmq

1.nameserver是无状态的，控制着broker，producer，consumer集群的同步

2.broker是处理消息中转，负责储存和过滤消息

3.rocketmq有两种消息消费模型，广播模式，一个消费者组中的每一个消费者都消费；集群模式，一个消费者组中的每一个消费者平均消费消息

4.顺序消费，生产者向一个topic的一个队列，顺序发送消息，切生产者只能单线程发送，消费者在消费时使用顺序消费模式


rocketmq是什么

1.一种队列模型的消息中间件，高性能，高可靠，高实时，分布式

2.producer，consumer，broker都可以分布式

3.默认情况下，prducer会向一个topic下的一些队列轮流发送消息


api

生产者

//顺序发送
DefaultMQProducer producer = new DefaultMQProducer(PRODUCT_GROUP);
        producer.setNamesrvAddr(ROCKET_SERVER);
        Message message = new Message("test", "tags", "1", info.getBytes());
        producer.start();
        for (int i = 0; i < 10; i++) {
            SendResult result = producer.send(message,new    SelectMessageQueueByHash(),i);
        }
        producer.shutdown();

 

//随机发送
DefaultMQProducer producer = new DefaultMQProducer(PRODUCT_GROUP);
        producer.setNamesrvAddr(ROCKET_SERVER);
        Message message = new Message("test", "tags", "1", info.getBytes());
        producer.start();
        for (int i = 0; i < 10; i++) {
            SendResult result = producer.send(message);
        }
        producer.shutdown();

消费者

//普通消费
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setNamesrvAddr(ROCKET_SERVER);
        consumer.subscribe("test", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                if (CollectionUtils.isNotEmpty(list)) {
                    Iterator it = list.iterator();
                    while (it.hasNext()) {
                        MessageExt msg = (MessageExt) it.next();
                        System.out.println(msg.toString());
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();

 

//顺序消费
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setNamesrvAddr(ROCKET_SERVER);
        consumer.subscribe("test", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                if (CollectionUtils.isNotEmpty(list)) {
                    Iterator it = list.iterator();
                    while (it.hasNext()) {
                        MessageExt msg = (MessageExt) it.next();
                        System.out.println(msg.toString());
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();

启动广播模式

consumer.setMessageModel(MessageModel.BROADCASTING);