SpringBoot集成RocketMQ
首先依旧是引入依赖

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

然后就可以编写发送不同类型消息的代码了

package blossom.project.springbootkp.seckillproducer;import blossom.project.springbootkp.seckillproducer.entity.MsgModel;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;import java.util.Arrays;
import java.util.List;@SpringBootTest
class SecKillProducerApplicationTests {private List<MsgModel> msgModels = Arrays.asList(new MsgModel("qwer", 1L, "下单"),new MsgModel("qwer", 1L, "短信"),new MsgModel("qwer", 1L, "物流"),new MsgModel("zxcv", 2L, "下单"),new MsgModel("zxcv", 2L, "短信"),new MsgModel("zxcv", 2L, "物流"));@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Testvoid syncProducer() {rocketMQTemplate.syncSend("bootTestTopic","使用springboot集成rocketmq");}@Testvoid asyncProducer(){rocketMQTemplate.asyncSend("bootTestTopic", "发送一条异步消息", new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功");}@Overridepublic void onException(Throwable throwable) {System.out.println("发送失败"+throwable.getMessage());}});}@Testvoid oneWayProducer(){rocketMQTemplate.sendOneWay("bootTestTopic","发送一个单向消息");}@Testvoid delayProducer(){Message<String> message = MessageBuilder.withPayload("这是一条延迟消息").build();rocketMQTemplate.syncSend("bootTestTopic",message,3000,2);}@Testvoid orderedProducer(){msgModels.forEach(x->{String s = JSONObject.toJSONString(x);rocketMQTemplate.syncSendOrderly("orderlyTopic", s,x.getOrderSn());});}@Testvoid tagProducer(){rocketMQTemplate.syncSend("bootTestTopic:tagA","我是一个带标签的消息");}@Testvoid keyProducer(){Message<String> message = MessageBuilder.withPayload("我是一个带有key的消息").setHeader(RocketMQHeaders.KEYS, "testKey").build();rocketMQTemplate.syncSend("bootTestTopic",message);}
}
对于不同的消息类型，我们可以使用不同的方式去接收。
创建一个顺序消息的监听器
@Component
@RocketMQMessageListener(topic = "orderlyTopic",consumerGroup = "boot-orderly-consumer-group",consumeMode = ConsumeMode.ORDERLY, //顺序消费模式 单线程maxReconsumeTimes = 5) //最大重试次数
public class OrderlyMessageListener implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt messageExt) {MsgModel msgModel = JSON.parseObject(new String(messageExt.getBody()), MsgModel.class);System.out.println(msgModel);}
}
普通的创建一个监听器
@Component
@RocketMQMessageListener(topic = "bootTestTopic",consumerGroup = "boot-consumuer-group")
public class SimpleMessageListener implements RocketMQListener<MessageExt> {/*** 这个方法就是消费者方法* 这里的String就是消息内容* 这里的泛型就是这里的参数类型* 如果泛型指定了固定的类型 那么消息体就是我们的参数* 如果我们的类型设定为具体的类型 那么我们只能拿到消息体* 而如果我们把消息类型设定为MessageExt类型，那么我们可以拿到消息头* ------------------------------------------------* 只要这个方法不报错 就会直接完成消息的接收 而如果报错了 就会重试* @param msg*/@Overridepublic void onMessage(MessageExt msg) {String keys = msg.getKeys();System.out.println("接收到的keys为"+keys);String body = new String(msg.getBody());System.out.println("接收到的消息体为"+body);}
}
创建一个识别tag标签的监听器
@RocketMQMessageListener(topic = "bootTestTopic",consumerGroup = "boot-tag-consumer-group",
selectorType = SelectorType.TAG, //tag过滤模式
selectorExpression = "tagA || tagB") //tag标签匹配模式
public class TagMessageListener implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt msg) {System.out.println(new String(msg.getBody()));}
}