springboot集成rocketmq失败记录

一般集成一个中间件,比如redis,elasticjob等,分为以下几个步骤

1.导入包含springboot对应的start

2.写配置文件,相应的配置信息

3.定义配置文件对应的属性类,加载配置信息(可不用,用注解读取配置文件属性)

4.写配置类,初始化相应的bean

比如elasticjob需要zk,所以要配置类配置zk信息,今天说的roketmq需要nameserver服务作为类似注册中心,所以需要配置nameserver信息。但是既然是springboot start最大的好处就是简化配置,理想情况下,只需要前两步,引入依赖,添加配置信息就行了。至于加载配置信息,创建bean应该是springboot自己做好了。

来,

1.

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.2.1</version>
        </dependency>

2. 

springboot集成rocketmq失败记录

 怎么发送一个同步消息

@Slf4j
public class SyncProducer {

    public static void main(String[] args) throws Exception {
	DefaultMQProducer producer = new DefaultMQProducer("group_suo");
	producer.setNamesrvAddr("47.103.143.191:8201;47.103.143.191:8202");
	producer.start();
	Message msg = new Message("Topic3", "Tag", UUID.randomUUID().toString(),
			("Hello RocketMQ , I`m sk4").getBytes(RemotingHelper.DEFAULT_CHARSET));
	SendResult sendResult = producer.send(msg);
	System.out.printf("%s%n", sendResult);
	log.info("sendResult: ", JSON.toJSONString(sendResult));

	producer.shutdown();
    }

}

创建一个生产者,很多教程这样写,配置nameServer,启动,发送,关闭

有以下疑问

1,生产者应该封装成工具类

2,每次发消息,都要自己启动和关闭吗,mq没有自己做

先不这样做,我们的目的就是拿到生产者,然后发消息,工具类如下

@Component
public class RocketMQProducerUtil {
    private static final Logger log = LoggerFactory.getLogger(RocketMQProducerUtil.class);

    @Autowired(required = false)
    private DefaultMQProducer defaultMQProducer;

    public RocketMQProducerUtil() {
    }

    public SendResult sendMessage(String topic, String tags, String keys, String contentText) {
	this.checkParam(topic, tags, keys, contentText);
	Message message = new Message(topic, tags, keys, contentText.getBytes());
	this.generateTraceId(message);

	try {
	    SendResult sendResult = this.defaultMQProducer.send(message);
	    return sendResult;
	} catch (Exception var7) {
	    log.error("rocketMq message send error :{}", var7.getMessage());
	    var7.printStackTrace();
	    throw new RRException("rocketMq message send error", var7);
	}
    }
针对,DefaultMQProducer,那么我们不自己创建bean,springboot是怎么做的,怎么加载到nameserver的?看下这个配置类,RocketMQAutoConfiguration,(比较奇怪在spring-factories没有找到这个配置类,也没啥影响,在start中,单例可以理解),这个配置类在rocketmq-start中。看下这个配置类信息,做的什么

1.配置信息

rocketmq.name-server 是必须的,要这样命名

2.创建了生产者,消费者bean

3.把生产者,消费者bean,加入

RocketMQTemplate 中,所以我们可以用这个类,获取生产者消费者,再去调用他们的方法

springboot集成rocketmq失败记录

springboot集成rocketmq失败记录 

但是,创建生产者时,即使按照条件配置了,断点还是无法走到这一步,无法创建生产者bean,不知道为什么?

@ConditionalOnProperty(
    prefix = "rocketmq",
    value = {"name-server", "producer.group"}
)

如果不使用,springboot自己创建的bean,我们只有自己创建

@Component
public class MQProducerConfigure {

    private static final Logger log = LoggerFactory.getLogger(MQProducerConfigure.class);
    @Value("${rocketmq.producer.groupName:#{null}}")
    private String producerGroupName;
    @Value("${rocketmq.name-server:#{null}}")
    private String nameserAddr;
    @Value("${rocketmq.producer.maxMessageSize:131072}")
    private int maxMessageSize;
    @Value("${rocketmq.producer.sendMsgTimeout:10000}")
    private int sendMsgTimeout;
    private DefaultMQProducer producer;

    public MQProducerConfigure() {
    }

    @Bean(name = { "defaultMQProducer" })
    @Primary
    public DefaultMQProducer getRocketMQProducer() {
	if (!Objects.isNull(this.producerGroupName) && this.producerGroupName.trim().length() != 0) {
	    if (Objects.isNull(this.nameserAddr)) {
		throw new RRException(5000001, "rocketmq.namesrvAddr is not setted");
	    } else {
		this.producer = new DefaultMQProducer(this.producerGroupName);
		this.producer.setNamesrvAddr(this.nameserAddr);
		this.producer.setMaxMessageSize(this.maxMessageSize);
		this.producer.setSendMsgTimeout(this.sendMsgTimeout);
		this.producer.setVipChannelEnabled(false);

		try {
		    this.producer.start();
		    log.info("rocketMQ is start !!groupName : {},nameserAddr:{}", this.producerGroupName,
				    this.nameserAddr);
		} catch (MQClientException var5) {
		    log.error(String.format("rocketMQ start error,{}", var5.getMessage()));
		    var5.printStackTrace();
		}
		return this.producer;
	    }
	} else {
	    log.info("RocketMQProducerConfig-getRocketMQProducer():{}",
			    "can not find producer group name [rocketmq.producer.groupName], will not create RocketMq producer");
	    return null;
	}
    }
}

通过断点可追踪到,先创建了生产者bean, 然后创建

RocketMQTemplate 时,报错

springboot集成rocketmq失败记录

 springboot集成rocketmq失败记录

Caused by: org.apache.rocketmq.client.exception.MQClientException: The producer service state not OK, maybe started once, RUNNING

啥意思:生产者服务状态不正常,可能启动过一次,正在运行

对啊,创建bean时,确实调用了product.start(),为什么导致报错呢?

springboot集成rocketmq失败记录

 那把启动的代码注释掉...确实没有报错了

继续

用生产者发消息能否成功,真成功了,而且被正常监听到,并消息

springboot集成rocketmq失败记录

 消费者

/**
 * consumerMode=ConsumeMode.ORDERLY(这个是顺序消费,默认是ConsumeMode.CONCURRENT异步多线程消费)
 * <p>
 * consumeMode = ConsumeMode.CONCURRENTLY, //默认值 ConsumeMode.CONCURRENTLY 并行处理 ConsumeMode.ORDERLY 按顺序处理
 * messageModel = MessageModel.CLUSTERING, //消息模式:广播和集群,默认是集群
 */
@Component
@Slf4j
@RocketMQMessageListener(topic = "topic1", consumerGroup = "group_1")
public class MyConsumer implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt messageExt) {
	log.info("消息内容 body:{}, tag:{}", new String(messageExt.getBody()), messageExt.getTags());
	log.info("消费内容:{}", JSON.toJSONString(messageExt));
    }
}

再继续发送消息,也是正常的,配置如下

springboot集成rocketmq失败记录

 至此,文章开头的常规步骤,自己定义配置类,创建bean,可以完成整个功能。

上一篇:开发者不可错过的开源工具 —— Android 篇


下一篇:Servlet的起源以及Servlet到底是什么,rocketmq教程教程