搭建MQ

使用docker搭建RocketMQ
我這里是使用docker里面的RocketMQ鏡像和本地idea交互的

編寫代碼

導入依賴pom.xml

<!-- rocketMQ --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.1.0</version></dependency>

在application.yml文件添加

apache:rocketmq:consumer:pushConsumer: PushConsumer # 消費者的組名producer:producerGroup: ProducerGroup # 生產者的組名namesrvAddr: localhost:9876 # NameServer地址

在原來訂單service里面添加

/*** 生產者的組名*/@Value("${apache.rocketmq.producer.producerGroup}")private String producerGroup;/*** NameServer 地址*/@Value("${apache.rocketmq.namesrvAddr}")private String namesrvAddr;/*** rocketmq 異步發送訂單** @param oid 訂單id* @param uid user_id* @param couponId 優惠券id*/public void asyncProducer(Long oid, Long uid, Long couponId) {//生產者的組名DefaultMQProducer producer = new DefaultMQProducer(producerGroup);//指定NameServer地址，多個地址以 ; 隔開producer.setNamesrvAddr(namesrvAddr);try {// 生產者開始producer.start();// 生產者發送失敗重試次數producer.setRetryTimesWhenSendAsyncFailed(2);// 關鍵信息String orderMsg = uid.toString() + "," + oid.toString() + "," + couponId.toString();// 發送的信息Message msg = new Message("order","push",orderMsg.getBytes(RemotingHelper.DEFAULT_CHARSET));/*** 目前RocketMQ免費版只支持固定精度級別的定時消息，服務器按照1-N定義了如下級別：* “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”；* 若要發送定時消息，在應用層初始化Message消息對象之后，* 調用setDelayTimeLevel(int level)方法來設置延遲級別，按照序列取相應的延遲級別，例如level=2，則延遲為5s:*///msg.setDelayTimeLevel(16);msg.setDelayTimeLevel(4); // 這里我為了方便查看將世界設置成為30S//重點在這里 異步發送回調producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("發送異步響應：MsgId:" + sendResult.getMsgId() + "，發送狀態:" + sendResult.getSendStatus());// 取消生產者producer.shutdown();}@Overridepublic void onException(Throwable e) {// 自定義錯誤e.printStackTrace();// 取消生產者producer.shutdown();}});} catch (Exception e) {e.printStackTrace();}}// 這里是將原來發送redis修改成為rocketmq// 這里搜索替換吧//加入到延遲消息隊列 // this.sendToRedis(order.getId(), uid, couponId);this.asyncProducer(order.getId(), uid, couponId);

設置消費者(控制訂單狀態)

@Component public class RocketMQServer {/*** 消費者的組名*/@Value("${apache.rocketmq.consumer.pushConsumer}")private String consumerGroup;/*** NameServer 地址*/@Value("${apache.rocketmq.namesrvAddr}")private String namesrvAddr;@Resourceprivate OrderCancelService orderCancelService;@Resourceprivate CouponBackService couponBackService;@PostConstruct // 注入了就開始執行、監聽生產者public void defaultMQPushConsumer() {//消費者的組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);//指定NameServer地址，多個地址以 ; 隔開consumer.setNamesrvAddr(namesrvAddr);try {//訂閱order下Tag為push的消息consumer.subscribe("order", "push");//設置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費//如果非第一次啟動，那么按照上次消費的位置繼續消費consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {try {for (MessageExt messageExt : list) {System.out.println("messageExt: " + messageExt);//輸出消息內容String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);System.out.println("消費響應：msgId : " + messageExt.getMsgId() + ", msgBody : " + messageBody);//輸出消息內容OrderMessageBO messageBO = new OrderMessageBO(messageBody);// 對訂單進行操作this.orderCancelService.cancel(messageBO);this.couponBackService.returnBack(messageBO);}} catch (Exception e) {e.printStackTrace();// 接收失敗重試if (list.get(0).getReconsumeTimes() == 3){// 重試3次return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消費成功}else {return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再試}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消費成功});consumer.start();} catch (Exception e) {e.printStackTrace();}}}

結果

剛剛下單 數據庫狀態是1

剛剛下單 發送rocketmq

到了時間接收rocketmq消費的信息狀態變成5

接收rocketmq消費的信息

總結

以上是生活随笔為你收集整理的风袖使用RocketMQ实现订单状态转变的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。