日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RocketMQ內(nèi)置了大量的測(cè)試案例，并且這些測(cè)試案例可以通過RocketMQ的bin目錄下的tools.cmd命令進(jìn)行測(cè)試。接下來，我們就使用RocketMQ自帶的tools.cmd命令測(cè)試RocketMQ的環(huán)境。

（1）啟動(dòng)生產(chǎn)者程序向RocketMQ發(fā)送消息。

重新打開cmd命令行，進(jìn)入RocketMQ的bin目錄，在命令行輸入如下命令調(diào)用RocketMQ自帶的生產(chǎn)者程序向RocketMQ發(fā)送消息。

set NAMESRV_ADDR=localhost:9876
tools.cmd org.apache.rocketmq.example.quickstart.Producer

可以看到，執(zhí)行完上述兩條命令后，生產(chǎn)者程序開始向RocketMQ發(fā)送消息。

（2）啟動(dòng)消費(fèi)者程序消費(fèi)RocketMQ中的消息。

重新打開cmd命令行，進(jìn)入RocketMQ的bin目錄，在命令行輸入如下命令調(diào)用RocketMQ自帶的消費(fèi)者程序消費(fèi)RocketMQ中的消息。

set NAMESRV_ADDR=localhost:9876
tools.cmd org.apache.rocketmq.example.quickstart.Consumer

可以看到，執(zhí)行完上述兩條命令后，消費(fèi)者程序開始消費(fèi)RocketMQ中的消息。

說明我們使用源碼編譯搭建RocketMQ環(huán)境成功了。

源碼編譯RocketMQ控制臺(tái)

這里需要注意的是：RocketMQ控制臺(tái)本質(zhì)上是一個(gè)SpringBoot程序，啟動(dòng)后默認(rèn)監(jiān)聽的端口是8080。RocketMQ的新版控制臺(tái)已經(jīng)從RocketMQ的rocketmq-externals項(xiàng)目中分離出來了。也就是說，新版的RocketMQ控制臺(tái)已經(jīng)從https://github.com/apache/rocketmq-externals鏈接所示的項(xiàng)目中分離出來，新版控制臺(tái)的鏈接地址為：https://github.com/apache/rocketmq-dashboard。

（1）從鏈接https://github.com/apache/rocketmq-dashboard下載新版的RocketMQ控制臺(tái)源碼。下載后解壓。

（2）進(jìn)入到RocketMQ控制臺(tái)源碼解壓目錄的src/main/resources目錄下，編輯application.yml文件，修改??namesrvAddrs??地址，去掉多余的namesrvAddrs地址。

application.yml文件中原來的配置如下所示。

rocketmq:
  config:
    # if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, default localhost:9876
    # configure multiple namesrv addresses to manage multiple different clusters
    namesrvAddrs:
      - 127.0.0.1:9876
      - 127.0.0.2:9876

將127.0.0.2:9876刪除或者注釋掉，如下所示。

rocketmq:
  config:
    # if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR | now, default localhost:9876
    # configure multiple namesrv addresses to manage multiple different clusters
    namesrvAddrs:
      - 127.0.0.1:9876
    #  - 127.0.0.2:9876

RocketMQ控制臺(tái)啟動(dòng)時(shí)默認(rèn)監(jiān)聽的端口是8080，由于我們項(xiàng)目中訂單微服務(wù)監(jiān)聽的端口也是8080，所以，將RocketMQ控制臺(tái)監(jiān)聽的端口修改為10003，修改前的配置如下所示。

server:
  port: 8080

修改后的配置如下所示。

server:
  port: 10003

（3）修改完application.yml文件后，打開cmd命令行，進(jìn)入RocketMQ控制臺(tái)源碼的根目錄，輸入如下Maven命令開始編譯RocketMQ控制臺(tái)的源碼。

mvn clean install -Dmaven.test.skip=true

編譯過程如下所示。

（4）編譯完成后，會(huì)在RocketMQ控制臺(tái)源碼的根目錄下生成target目錄，如下所示。

進(jìn)入target目錄后，可以看到生成了rocketmq-dashboard-1.0.1-SNAPSHOT.jar文件，如下所示。

這個(gè)jar文件就是RocketMQ控制臺(tái)的運(yùn)行文件。

（5）重新打開cmd命令行，進(jìn)入rocketmq-dashboard-1.0.1-SNAPSHOT.jar文件所在的命令，在命令行直接輸入如下命令啟動(dòng)RocketMQ控制臺(tái)程序。

java -jar rocketmq-dashboard-1.0.1-SNAPSHOT.jar

驗(yàn)證RocketMQ控制臺(tái)

在瀏覽器中輸入??http://localhost:10003??后，出現(xiàn)如下畫面說明RocketMQ啟動(dòng)成功。

界面默認(rèn)是英文，我們也可以點(diǎn)擊右上角的??changeLanguage??切換語(yǔ)言，切換成中文顯示，如下所示。

選擇主題菜單想后如下所示。

可以看到目前RocketMQ中存在一個(gè)名稱為TopicTest的主題，點(diǎn)擊TopicTest主題的狀態(tài)按鈕，如下所示。

會(huì)顯示TopicTest主題的消息隊(duì)列信息，如下所示。

可以看到，正確顯示出了TopicTest主題的消息隊(duì)列信息，說明RocketMQ控制臺(tái)啟動(dòng)成功了。

編碼測(cè)試RocketMQ?

我們使用RocketMQ自帶的生產(chǎn)者和消費(fèi)者程序?qū)崿F(xiàn)了消息的生成與消費(fèi)，為了讓小伙伴們能夠更加直觀的感受到消息中間件在項(xiàng)目中的作用，接下來，我們自己編碼測(cè)試下RocketMQ。

導(dǎo)入RocketMQ依賴

在用戶微服務(wù)shop-user的pom.xml中，添加RocketMQ相關(guān)的依賴，如下所示。

    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.0.3


    org.apache.rocketmq
    rocketmq-client
    4.5.2

編寫生產(chǎn)者代碼

在用戶微服務(wù)的sec/test/java目錄下新建??io.binghe.shop.rocketmq.test??包，在包下創(chuàng)建RocketMQProducer類，作為RocketMQ的生產(chǎn)者，代碼如下所示。

/**
 * @author binghe
 * @version 1.0.0
 * @description RocketMQ生產(chǎn)者
 */
public class RocketMQProducer {

    public static void main(String[] args) throws Exception {
        //創(chuàng)建消息生產(chǎn)者
        DefaultMQProducer producer = new DefaultMQProducer("bingheProducerGroup");
        //設(shè)置NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //啟動(dòng)生產(chǎn)者
        producer.start();
        //構(gòu)建消息對(duì)象
        Message message = new Message("bingheTopic", "bingheTag", "Hello RocketMQ".getBytes());
        System.out.println("生產(chǎn)者發(fā)出的消息為：" + JSONObject.toJSONString(message));
        //發(fā)送消息并接收結(jié)果
        SendResult sendResult = producer.send(message);
        //打印結(jié)果信息
        System.out.println("生產(chǎn)者收到的發(fā)送結(jié)果信息為：" + JSONObject.toJSONString(sendResult));
        //關(guān)閉生產(chǎn)者
        producer.shutdown();
    }
}

生產(chǎn)者的代碼比較簡(jiǎn)單，這里就不再贅述了。

編寫消費(fèi)者代碼

在??io.binghe.shop.rocketmq.test??包下新建RocketMQConsumer類，作為RocketMQ的消費(fèi)者，代碼如下所示。

/**
 * @author binghe
 * @version 1.0.0
 * @description RocketMQ消費(fèi)者
 */
public class RocketMQConsumer {
   public static void main(String[] args) throws Exception {
       try{
           //創(chuàng)建消息消費(fèi)者
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("bingheConsumerGroup");
           //設(shè)置NameServer地址
           consumer.setNamesrvAddr("127.0.0.1:9876");
           //訂閱bingheTopic主題
           consumer.subscribe("bingheTopic", "*");
           //設(shè)置消息監(jiān)聽，當(dāng)收到消息時(shí)RocketMQ會(huì)回調(diào)消息監(jiān)聽
           consumer.registerMessageListener(new MessageListenerConcurrently() {
               @Override
               public ConsumeConcurrentlyStatus consumeMessage(List list,
                                                               ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                   //打印消息消費(fèi)者收到的RocketMQ消息
                   System.out.println("消費(fèi)者收到的消息為：" + list);
                   //返回消息消費(fèi)成功的標(biāo)識(shí)
                   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }
           });
           //啟動(dòng)消費(fèi)者
           consumer.start();
           System.out.println("消費(fèi)者啟動(dòng)成功");
       }catch (Exception e){
           e.printStackTrace();
       }
   }
}

測(cè)試消息的生產(chǎn)與消費(fèi)

（1）為了便于觀察，這里我們先啟動(dòng)消費(fèi)者程序RocketMQConsumer，啟動(dòng)RocketMQConsumer后會(huì)在IDEA的控制臺(tái)打印如下信息。

消費(fèi)者啟動(dòng)成功

說明消費(fèi)者啟動(dòng)成功了。

（2）運(yùn)行生產(chǎn)者程序RocketMQProducer，運(yùn)行后RocketMQProducer程序控制臺(tái)會(huì)輸出如下信息。

生產(chǎn)者發(fā)出的消息為：{"body":"SGVsbG8gUm9ja2V0TVE=","delayTimeLevel":0,"flag":0,"properties":{"WAIT":"true","TAGS":"bingheTag"},"tags":"bingheTag","topic":"bingheTopic","waitStoreMsgOK":true}
生產(chǎn)者收到的發(fā)送結(jié)果信息為：{"messageQueue":{"brokerName":"DESKTOP-PSKC7T1","queueId":1,"topic":"bingheTopic"},"msgId":"C0A8006F538418B4AAC25B9EDDAC0000","offsetMsgId":"C0A8B80100002A9F0000000000036B16","queueOffset":2,"regionId":"DefaultRegion","sendStatus":"SEND_OK","traceOn":true}

說明生產(chǎn)者程序RocketMQProducer成功將消息發(fā)送到RocketMQ。

（3）接下來，再看下消費(fèi)者程序RocketMQConsumer的控制臺(tái)，如下所示。

消費(fèi)者收到的消息為：[MessageExt [queueId=1, storeSize=206, queueOffset=2, sysFlag=0, bornTimestamp=1652871538093, bornHost=/192.168.184.1:52915, storeTimestamp=1652871538099, storeHost=/192.168.184.1:10911, msgId=C0A8B80100002A9F0000000000036B16, commitLogOffset=224022, bodyCRC=1774740973, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='bingheTopic', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=3, CONSUME_START_TIME=1652871538103, UNIQ_KEY=C0A8006F538418B4AAC25B9EDDAC0000, CLUSTER=DefaultCluster, WAIT=true, TAGS=bingheTag}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81], transactionId='null'}]]

說明生成者發(fā)送到RocketMQ的消息，被消費(fèi)者成功消費(fèi)到了。

項(xiàng)目整合RocketMQ?

我們?cè)陧?xiàng)目中模擬一個(gè)用戶成功下單后，為用戶發(fā)送通知，通知用戶下單成功的邏輯，具體的流程就是下單成功后將訂單的信息發(fā)送到RocketMQ，然后用戶微服務(wù)訂閱RocketMQ的消息，接收到消息后進(jìn)行打印。

用戶微服務(wù)整合RocketMQ

（1）編碼測(cè)試RocketMQ時(shí)，導(dǎo)入了RocketMQ的依賴，這里就不用再次導(dǎo)入了。

（2）在用戶微服務(wù)shop-user的application.yml文件中添加如下RocketMQ的配置。

rocketmq:
  name-server: 127.0.0.1:9876

（3）在用戶微服務(wù)shop-user中創(chuàng)建??io.binghe.shop.user.rocketmq??包，在包下創(chuàng)建RocketConsumeListener，實(shí)現(xiàn)org.apache.rocketmq.spring.core.RocketMQListener接口，具體代碼如下所示。

/**
 * @author binghe
 * @version 1.0.0
 * @description 監(jiān)聽消費(fèi)
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user-group", topic = "order-topic")
public class RocketConsumeListener implements RocketMQListener {
    @Override
    public void onMessage(Order order) {
        log.info("用戶微服務(wù)收到了訂單信息：{}", JSONObject.toJSONString(order));
    }
}

其中，RocketConsumeListener類上的@RocketMQMessageListener注解，表示當(dāng)前類是一個(gè)RocketMQ的消費(fèi)者，在@RocketMQMessageListener注解中配置了消費(fèi)者組為user-group，主題為order-topic。

至此，用戶微服務(wù)整合RocketMQ完畢。

訂單微服務(wù)整合RocketMQ

（1）在訂單微服務(wù)shop-order的pom.xml文件中添加RocketMQ的依賴，如下所示。

    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.0.3



    org.apache.rocketmq
    rocketmq-client
    4.5.2

（2）在訂單微服務(wù)shop-order的application.yml文件中添加如下配置。

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: order-group

（3）將??io.binghe.shop.order.service.impl.OrderServiceV6Impl???類，復(fù)制一份成??io.binghe.shop.order.service.impl.OrderServiceV7Impl???類，接下來，在??io.binghe.shop.order.service.impl.OrderServiceV7Impl??類中操作。

將??io.binghe.shop.order.service.impl.OrderServiceV7Impl??類上的@Service注解中的名稱修改為orderServiceV7，如下所示。

@Slf4j
@Service("orderServiceV7")
public class OrderServiceV7Impl implements OrderService {
    //省略具體代碼
}

（4）在??io.binghe.shop.order.service.impl.OrderServiceV7Impl??類中，注入RocketMQTemplate對(duì)象，如下所示。

@Autowired
private RocketMQTemplate rocketMQTemplate;

（5）在??io.binghe.shop.order.service.impl.OrderServiceV7Impl#saveOrder()??方法中，提交訂單成功后將訂單信息寫入RocketMQ，如下所示。

@Override
@Transactional(rollbackFor = Exception.class)
public void saveOrder(OrderParams orderParams) {
    //省略上面所有代碼
    rocketMQTemplate.convertAndSend("order-topic", order);
}

（6）在??io.binghe.shop.order.controller.OrderController??中，將注入的OrderService的名稱修改成orderServiceV7，如下所示。

@Autowired
@Qualifier(value = "orderServiceV7")
private OrderService orderService;

「注意：訂單微服務(wù)shop-order中，修改后的代碼見源碼工程，冰河在這里不再粘貼完整的源代碼?！?/p>

測(cè)試項(xiàng)目整合的RocketMQ

（1）分別啟動(dòng)Nacos，Sentinel，ZipKin和RocketMQ。

（2）分別啟動(dòng)用戶微服務(wù)、商品微服務(wù)、訂單微服務(wù)和網(wǎng)關(guān)服務(wù)。

（3）在瀏覽器中輸入??localhost:10001/server-order/order/submit_order?userId=1001&productId=1001&count=1??，如下所示。

（4）查看用戶微服務(wù)shop-user的控制臺(tái)，發(fā)現(xiàn)會(huì)輸出訂單的信息，如下所示。

2022-05-18 20:37:26.440  INFO [server-user,,,] 18064 --- [MessageThread_1] i.b.s
                                                本文題目：消息服務(wù)：項(xiàng)目整合RocketMQ                                                

                                                鏈接URL：http://m.5511xx.com/article/cccihed.html