快速入门RocketMQ

红太狼 2020-11-01 08:09 998阅读 0赞

以下是rocketMQ的简单小demo：

Producer：

package com.olayc.edi.rocketmq;
    
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    
    
    /**
     * @description: 生产者模型
     * @author: Mr.Wang
     * @create: 2019-05-05 15:39
     **/
    public class Producer {
        public static void main(String[] args) throws Exception {
            // new一个rocketMQ中producer默认的名字 -> group
            DefaultMQProducer producer = new DefaultMQProducer("group");
            // 给一个namesrv的地址
            producer.setNamesrvAddr("192.168.56.200:9876");
            // 启动producer  和远程的namesrv进行连接
            producer.start();
            // 写数据
            for (int i = 0; i < 100; i++) {
                // topic(主题的名字) -> orders  所有属于一个topic的消息都放在一起  相当于对消息进行分类
                //  ("order" + i).getBytes() -> 消息体   (字节数组)
                Message msg = new Message("orders", ("order" + i).getBytes());
                SendResult result = producer.send(msg);
                System.out.println(result);
                System.out.println(msg + "send out");
                Thread.sleep(500L);
            }
            producer.shutdown();
    
        }
    
    }

PushConsumer：

package com.olayc.edi.rocketmq;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    /**
     * @description: 监听器模式的consumer -- 观察者模式
     * @author: Mr.Wang
     * @create: 2019-05-05 16:02
     **/
    public class PushConsumer {
        public static void main(String[] args) {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumers");
            consumer.setNamesrvAddr("192.168.56.200:9876");
    
            try {
                // 订阅消息  topic -> orders   第二个参数是tag 没有设置为null
                consumer.subscribe("orders", null);
                // 程序第一次启动从消息队列头取数据
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                consumer.registerMessageListener(new MessageListenerConcurrently() {
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        MessageExt msg = list.get(0);
                        System.out.println(new String(msg.getBody()));
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                });
                consumer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }
    
    }