springboot 2.x 整合rocketmq-spring-boot-starter

一、导入maven依赖，rocketmq-spring-boot-starter可在maven中央仓库查找

org.apache.rocketmq rocketmq-spring-boot-starter 2.0.4 rocketmq 服务搭建在此省略.....

二、yml 配置

rocketmq:
name-server: http://127.0.0.1:9876 #rocketmq服务地址
producer:
group: rocketmq_test #自定义的组名称
send-message-timeout: 3000 #消息发送超时时长
若rocketmq是在阿里云平台买的，使用以下配置即可

rocketmq:
name-server: http://xxxxx.mq-internet-access.mq-internet.aliyuncs.com:80 #阿里云rocketmq连接地址
producer:
group: rocketmq_test #自定义的组名称
access-key: your aliyun accessKey
secret-key: your aliyun secretKey
send-message-timeout: 3000 #消息发送超时时长
三、发送消息

@Service
public class RocketMQProducer{

@Autowired
private RocketMQTemplate rocketMQTemplate;

@Value("${rocketmq.producer.send-message-timeout}")
private Integer messageTimeOut;

/**

• 发送普通消息
  */
  public void sendMsg(String msgBody){
  rocketMQTemplate.syncSend(“queue_test_topic”,MessageBuilder.withPayload(msgBody).build());
  }

/**

• 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
  */
  public void sendAsyncMsg(String msgBody){
  rocketMQTemplate.asyncSend(“queue_test_topic”,MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
  @Override
  public void onSuccess(SendResult sendResult) {
  // 处理消息发送成功逻辑
  }
  @Override
  public void onException(Throwable e) {
  // 处理消息发送异常逻辑
  }
  });
  }

/**

• 发送延时消息
  
• 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
  
  */
  public void sendDelayMsg(String msgBody, Integer delayLevel){
  rocketMQTemplate.syncSend(“queue_test_topic”,MessageBuilder.withPayload(msgBody).build(),messageTimeOut,delayLevel);
  }

/**

• 发送带tag的消息,直接在topic后面加上":tag"
  */
  public void sendTagMsg(String msgBody){
  rocketMQTemplate.syncSend(“queue_test_topic:tag1”,MessageBuilder.withPayload(msgBody).build());
  }
  }
  四、接收消息

/**

rocketmq 消息监听，@RocketMQMessageListener中的selectorExpression为tag，默认为*
/
@Slf4j
@Component
@RocketMQMessageListener(topic = “queue_test_topic”,selectorExpression="",consumerGroup = “queue_group_test”)
public class RocketMQMsgListener implements RocketMQListener {

@Autowired
private RocketMQTemplate rocketMQTemplate;

@Override
public void onMessage(MessageExt message) {
byte[] body = message.getBody();
String msg = new String(body);
log.info(“接收到消息：{}”, msg);
}

}

https://blog.csdn.net/qq_38306688/article/details/107716046