<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>2.1.1.RELEASE</version>
</dependency>
rocketmq:
producer:
group: ldProducer
name-server: 192.168.3.158:9876
@Component
@RocketMQMessageListener(consumerGroup = "ldConsumer", topic = "ldDelayTaskTopic")
@Slf4j
public class LdRocketMQListener implements RocketMQListener<String> {
@Override
public void onMessage(String msg) {
log.info("获取到延迟任务消息:{}",msg);
}
}
@RestController
@Slf4j
public class RocketMQDelayTaskController {
@Resource
private DefaultMQProducer producer;
@GetMapping("/rocketmq/add")
public void addTask(@RequestParam("task") String task) throws Exception {
Message msg = new Message("ldDelayTaskTopic", "TagA", task.getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.setDelayTimeLevel(2);
// 发送消息并得到消息的发送结果,然后打印
log.info("提交延迟任务");
producer.send(msg);
}
}
org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [549]ms, Topic: ldTopicA, BrokersSent: [broker-a, broker-a, broker-a]
See http://rocketmq.apache.org/docs/faq/ for further details.
at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:665)
at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1343)
at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1289)
at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:325)
at com.example.delay.MQTest.sendTest(MQTest.java:46)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
...
Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CO