rocketMq 消费者疑问

Question

@Override
    public void receiveResult(String id, long timeout) {
        //计数器
        LOGGER.info(">>>开始接收订单结果:{}", id);
        CountDownLatch countDownLatch = new CountDownLatch(1);
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
        //从消息队列头开始消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    String result =new String(msgs.get(0).getBody(), "UTF-8");
                    LOGGER.info(">>>接收到结果:{}",result);
                    countDownLatch.countDown();
                } catch (Exception e) {
                    LOGGER.error(">>>消息处理异常", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            //设置过滤的top 和tag
            defaultMQPushConsumer.subscribe(topic, id);
            // 启动消费者  等待结果
            defaultMQPushConsumer.start();
            if (countDownLatch.await(timeout, TimeUnit.MILLISECONDS)) {
                LOGGER.info(">>>接收成功");
                return;
            }
            LOGGER.info(">>>接收超时");
        } catch (Exception e) {
            LOGGER.error(">>>等待消息异常", e);
        } finally {
            shutdown(defaultMQPushConsumer);
        }
        
    }

    private void shutdown(DefaultMQPushConsumer defaultMQPushConsumer) {
        try {
            defaultMQPushConsumer.shutdown();
        } catch (Exception e) {
            LOGGER.info(">>>资源释放异常", e);
        }
    }

 

有个业务场景： 在一个同步线程A过程中 执行到某个点的时候暂停 需要等待另外一个线程B通知 才继续往下面执行。 单机的话 可以直接使用countDownLatch 实现。但是在分布式环境中 B线程可能在另外一台机器上。所以我引入了rocketMq. 每次根据订单号传给tag 进行过滤。但是这样的话 就需要每次创建和销毁consumer。总是感觉这样做 不好。但是又没有其他方式实现。请rocketMq大神帮忙看看。

Answer 1

这个跟顺序消息 没什么关系吧

Answer 2

rocketMQ不是好像有个顺序消息么?