rocketmq Don't have SubscriptionGroup

1. 问题描述

   rocketmq 生产者发消息正常 mq后台也能够看到发出的消息

   可是消费者一直没消费 好像订阅没成功

 

2. 问题排查

经过上图查看 确实没有检测到订阅者

 

 

3. 问题解决

线上环境是 两台机器 共四个实例

项目中订阅了两个不一样地址不一样topic的mq

而后那个instanceName会有命名冲突

ip@进程id 

从新修改instanceName的value

//设置instanceName
 defaultMQPushConsumer.setInstanceName(System.currentTimeMillis()+ JVMRandom.nextLong(10)+"");

spring-rocketmq.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://www.springframework.org/schema/beans" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">


    <!--好评加分 差评扣分-->
    <bean id="evaluationConsumer" class="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer">
        <property name="consumerGroup" value="${driver_x_group}"/>
        <property name="namesrvAddr" value="${driver_x_url}"/>
    </bean>
    <!--好评消费者 监听器-->
    <bean id="evaluationMessageListener" class="com.x.mq.EvaluationXListener"/>
    <!--消费者启动类-->
    <bean id="evaluationOrderlyConsumer" class="com.x.mq.base.EvaluationXConsumer" init-method="init" destroy-method="destroy" scope="singleton">
        <property name="defaultMQPushConsumer" ref="evaluationConsumer"/>
        <property name="evaluationMessageListener" ref="evaluationMessageListener"/>
        <property name="topic" value="${driver_score_topic}"/>
        <property name="tag" value=""/>
    </bean>
</beans>

View Code

consumer.java

package com.x.mq.base; import com.x.mq.EvaluationMessageListener; import org.apache.commons.lang.math.JVMRandom; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Random; /** * Created by admin on 2018/1/31. */
public class EvaluationConsumer { private static final Logger logger = LoggerFactory.getLogger(EvaluationConsumer.class); private DefaultMQPushConsumer defaultMQPushConsumer; private EvaluationMessageListener evaluationMessageListener; private String tag; private String topic; public EvaluationMessageListener getEvaluationMessageListener() { return evaluationMessageListener; } public void setEvaluationMessageListener(EvaluationMessageListener evaluationMessageListener) { this.evaluationMessageListener = evaluationMessageListener; } public String getTag() { return tag; } public void setTag(String tag) { this.tag = tag; } public String getTopic() { return topic; } public void setTopic(String topic) { this.topic = topic; } public DefaultMQPushConsumer getDefaultMQPushConsumer() { return defaultMQPushConsumer; } public void setDefaultMQPushConsumer(DefaultMQPushConsumer defaultMQPushConsumer) { this.defaultMQPushConsumer = defaultMQPushConsumer; } public void init() throws MQClientException { // 订阅指定Topic下tags
 defaultMQPushConsumer.subscribe(topic, tag); // 设置Consumer第一次启动是从队列头部开始消费仍是队列尾部开始消费<br> // 若是非第一次启动，那么按照上次消费的位置继续消费
 defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); // 设置为集群消费(区别于广播消费)：集群只消费一次，广播会被多个消费者消费
 defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING); // defaultMQPushConsumer.registerMessageListener(messageListenerConcurrently);
 defaultMQPushConsumer.registerMessageListener(evaluationMessageListener); //设置instanceName
        defaultMQPushConsumer.setInstanceName(System.currentTimeMillis()+ JVMRandom.nextLong(10)+""); // 批量消费
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(20); // Consumer对象在使用以前必需要调用start初始化，初始化一次便可<br>
 defaultMQPushConsumer.start(); logger.info("==rocketmq==DefaultMQPushConsumer start success! consumerGroup:{},nameServiceAddr:{},topic:{},tag:{}",defaultMQPushConsumer.getConsumerGroup(),defaultMQPushConsumer.getNamesrvAddr(),topic,tag); } public void destroy(){ defaultMQPushConsumer.shutdown(); } }

View Code