日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

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

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

重新打開cmd命令行，進入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）啟動消費者程序消費RocketMQ中的消息。

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

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

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

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

源碼編譯RocketMQ控制臺

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

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

（2）進入到RocketMQ控制臺源碼解壓目錄的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控制臺啟動時默認監(jiān)聽的端口是8080，由于我們項目中訂單微服務(wù)監(jiān)聽的端口也是8080，所以，將RocketMQ控制臺監(jiān)聽的端口修改為10003，修改前的配置如下所示。

server:
  port: 8080

修改后的配置如下所示。

server:
  port: 10003

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

mvn clean install -Dmaven.test.skip=true

編譯過程如下所示。

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

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

這個jar文件就是RocketMQ控制臺的運行文件。

（5）重新打開cmd命令行，進入rocketmq-dashboard-1.0.1-SNAPSHOT.jar文件所在的命令，在命令行直接輸入如下命令啟動RocketMQ控制臺程序。

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

驗證RocketMQ控制臺

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

界面默認是英文，我們也可以點擊右上角的??changeLanguage??切換語言，切換成中文顯示，如下所示。

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

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

會顯示TopicTest主題的消息隊列信息，如下所示。

可以看到，正確顯示出了TopicTest主題的消息隊列信息，說明RocketMQ控制臺啟動成功了。

編碼測試RocketMQ?

我們使用RocketMQ自帶的生產(chǎn)者和消費者程序?qū)崿F(xiàn)了消息的生成與消費，為了讓小伙伴們能夠更加直觀的感受到消息中間件在項目中的作用，接下來，我們自己編碼測試下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");
        //啟動生產(chǎn)者
        producer.start();
        //構(gòu)建消息對象
        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)者的代碼比較簡單，這里就不再贅述了。

編寫消費者代碼

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

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

測試消息的生產(chǎn)與消費

（1）為了便于觀察，這里我們先啟動消費者程序RocketMQConsumer，啟動RocketMQConsumer后會在IDEA的控制臺打印如下信息。

消費者啟動成功

說明消費者啟動成功了。

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

生產(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）接下來，再看下消費者程序RocketMQConsumer的控制臺，如下所示。

消費者收到的消息為：[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的消息，被消費者成功消費到了。

項目整合RocketMQ?

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

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

（1）編碼測試RocketMQ時，導(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，實現(xiàn)org.apache.rocketmq.spring.core.RocketMQListener接口，具體代碼如下所示。

/**
 * @author binghe
 * @version 1.0.0
 * @description 監(jiān)聽消費
 */
@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)前類是一個RocketMQ的消費者，在@RocketMQMessageListener注解中配置了消費者組為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對象，如下所示。

@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>

測試項目整合的RocketMQ

（1）分別啟動Nacos，Sentinel，ZipKin和RocketMQ。

（2）分別啟動用戶微服務(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的控制臺，發(fā)現(xiàn)會輸出訂單的信息，如下所示。

2022-05-18 20:37:26.440  INFO [server-user,,,] 18064 --- [MessageThread_1] i.b.s
                                                網(wǎng)頁名稱：消息服務(wù)：項目整合RocketMQ                                                

                                                URL地址：http://www.dlmjj.cn/article/cccihed.html