RocketMQ使用

不念不忘少年蓝@ 2023-01-23 08:54 146阅读 0赞

# RocketMQ在SpringBoot中的使用 #

### 导入依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
        <version>2.3.0.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.0.4</version>
    </dependency>

###  application.properties配置 ###

server.port=9096
    spring.application.name=demo
    rocketmq.name-server=192.168.44.162:9876
    rocketmq.producer.group=my_test_producer_group
    rocketmq.producer.send-message-timeout=3000

### Producer端使用 ###

@Component
    public class MessageSender {
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        public void syncSend(){
            /**
             * 发送可靠同步消息 ,可以拿到SendResult 返回数据
             * 同步发送是指消息发送出去后，会在收到mq发出响应之后才会发送下一个数据包的通讯方式。
             * 这种方式应用场景非常广泛，例如重要的右键通知、报名短信通知、营销短信等。
             *
             * 参数1： topic:tag
             * 参数2:  消息体 可以为一个对象
             * 参数3： 超时时间 毫秒
             */
            SendResult result= rocketMQTemplate.syncSend("springboot-topic:tag","这是一条同步消息",10000);
        }
    
        /**
         * 发送 可靠异步消息
         * 发送消息后，不等mq响应，接着发送下一个数据包。发送方通过设置回调接口接收服务器的响应，并可对响应结果进行处理。
         * 异步发送一般用于链路耗时较长，对于RT响应较为敏感的业务场景，例如用户上传视频后通过启动转码服务，转码完成后通推送转码结果。
         *
         * 参数1： topic:tag
         * 参数2:  消息体 可以为一个对象
         * 参数3： 回调对象
         */
        public void asyncSend() throws Exception{
            rocketMQTemplate.asyncSend("springboot-topic:tag1", "这是一条异步消息", new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {}
    
                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                }
            });
            TimeUnit.SECONDS.sleep(100000);
        }
    
        /**
         * 发送单向消息
         * 参数1： topic:tag
         * 参数2:  消息体 可以为一个对象
         */
        public void sendOneWay(){
            rocketMQTemplate.sendOneWay("springboot-topic:tag1", "这是一条单向消息");
        }
    
        /**
         * 发送单向的顺序消息
         */
        public void sendOneWayOrderly(){
            for(int i=0;i<10;i++){
                rocketMQTemplate.sendOneWayOrderly("springboot-topic:tag1", "这是一条顺序消息"+i,"2673");
            }
        }
    }

### Consumer端使用 ###

/**
     * MessageModel：集群模式；广播模式
     * ConsumeMode：顺序消费；无序消费
     */
    @Component
    @RocketMQMessageListener(topic = "springboot-topic",consumerGroup = "qs-consumer-group",
            //selectorExpression = "tag1",selectorType = SelectorType.TAG,
            messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.CONCURRENTLY)
    public class MessageConsumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            try {
                System.out.println("----------接收到rocketmq消息:" + message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

# Java API基本使用 #

### 导入rocketMQ依赖 ###

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-broker</artifactId>
        <version>4.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-common</artifactId>
        <version>4.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-tools</artifactId>
        <version>4.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-filter</artifactId>
        <version>4.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-srvutil</artifactId>
        <version>4.7.1</version>
    </dependency>

### Producer端使用 ###

public class Producer {
        public static void main(String[] args) throws MQClientException, InterruptedException {
            DefaultMQProducer producer = new DefaultMQProducer("my_test_producer_group");
            producer.setNamesrvAddr("192.168.44.163:9876;192.168.44.164:9876");
            producer.start();
    
            for (int i = 0; i < 6; i++){
                try {
                    // tags 用于过滤消息 keys 索引键,多个用空格隔开,RocketMQ可以根据这些key快速检索到消息
                    Message msg = new Message("q-2-1", "TagA", "2673",
                            ("RocketMQ "+String.format("%05d", i)).getBytes());
                    SendResult sendResult = producer.send(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
    
            producer.shutdown();
        }
    }

producer发送消息时可自定义发送规则，发送给哪个message queue

// 自定义发送规则，重写MessageQueueSelector实现
    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
            Integer id = (Integer) arg;
            int index = id % mqs.size();
            return mqs.get(index);
        }
    }, i);

发送消息到一个broker

//Call send message to deliver message to one of brokers.
    producer.sendOneway(msg);

###  Consumer端使用 ###

public class Consumer {
        private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();
    
        public static void main(String[] args) throws MQClientException {
            // 创建消费者
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("my_test_consumer_group");
            consumer.setNamesrvAddr("192.168.44.162:9876");
            consumer.start();
            // 从指定topic中拉取所有消息队列
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("test_topic_1");
            for (MessageQueue mq : mqs) {
                System.out.println("Consume from the queue: " + mq);
                SINGLE_MQ: while (true) {
                    try {
                        // 获取消息的offset，指定从store中获取
                        long offset = consumer.fetchConsumeOffset(mq, true);
                        PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
    
                        if (null != pullResult.getMsgFoundList()) {
                            for (MessageExt messageExt : pullResult.getMsgFoundList()) {
                                System.out.print(new String(messageExt.getBody()));
                            }
                        }
    
                        putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                        switch (pullResult.getPullStatus()) {
                            case FOUND:
                                break;
                            case NO_MATCHED_MSG:
                                break;
                            case NO_NEW_MSG:
                                break SINGLE_MQ;
                            case OFFSET_ILLEGAL:
                                break;
                            default:
                                break;
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
            consumer.shutdown();
        }
    
        private static void putMessageQueueOffset(MessageQueue mq, long offset) {
            offseTable.put(mq, offset);
        }
    
        // 需要消费者自己记录消费的偏移量  因为消息在broker上是持久的 (
        private static long getMessageQueueOffset(MessageQueue mq) {
            Long offset = offseTable.get(mq);
            if (offset != null){
                return offset;
            }
            return 0;
        }
    }