快捷導(dǎo)航

SpringBoot中使用RocketMQ的示例代碼

更新時(shí)間：2021年09月03日 14:44:58 作者：陳家寶

本文主要介紹SpringBoot中使用RocketMQ的示例代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

1 訂單微服務(wù)發(fā)送消息

1.1 訂單微服務(wù)添加rocketmq的依賴(lài)

        <!--rocketmq-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.1</version>
        </dependency>

1.2 添加配置

rocketmq:
  name-server: 192.168.232.129:9876 # rocketmq服務(wù)的地址
  producer:
    group: jack-producer-group  # 生產(chǎn)者組名

1.3 編寫(xiě)測(cè)試代碼

package cn.jack.controller;
 
import cn.jack.domain.Order;
import cn.jack.domain.Product;
import cn.jack.service.OrderService;
import cn.jack.service.ProductService;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;
 
@RestController
@Slf4j
public class OrderController {
 
    @Autowired
    private RestTemplate restTemplate;
    @Autowired
    private OrderService orderService;
    @Autowired
    private DiscoveryClient discoveryClient;
    @Autowired
    private ProductService productService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
 
    /**
     * 下單 -- 基于Feign實(shí)現(xiàn)服務(wù)調(diào)用
     * @param pid   商品id
     * @return
     */
    @RequestMapping("/order/prod/{pid}")
    public Order order(@PathVariable("pid") Long pid) {
        log.info("收到下單請(qǐng)求，準(zhǔn)備查詢(xún)商品信息。pid={}", pid);
 
        // 通過(guò)Feign調(diào)用商品微服務(wù)，查詢(xún)商品信息
        Product product = this.productService.findByPid(pid);
        log.info("商品信息查詢(xún)成功。內(nèi)容為：{}", JSON.toJSONString(product));
 
        // 進(jìn)行容錯(cuò)判斷
        if (product.getPid() == -100) {
            Order order = new Order();
            order.setOid(-100L);
            order.setPname("下單失敗");
            return order;
        }
 
        // 生成商品信息保存
        Order order = new Order();
        order.setNumber(1);
        order.setPid(pid);
        order.setPname(product.getPname());
        order.setPprice(product.getPprice());
        order.setUid(1);
        order.setUsername("陳家寶");
 
        this.orderService.createOrder(order);
        log.info("訂單信息保存成功。內(nèi)容為：{}", JSON.toJSONString(order));
 
        // TODO 下單成功之后，mq發(fā)送訂單內(nèi)容消息
        rocketMQTemplate.convertAndSend("jack-topic", order);
 
        return order;
    }
}

1.4 測(cè)試

訪問(wèn)下單接口，確認(rèn)RocketMQ是否發(fā)送了消息。從RocketMQ控制臺(tái)得到確認(rèn)，消息發(fā)送成功。

2 用戶(hù)微服務(wù)訂閱消息

2.1 用戶(hù)微服務(wù)增加rocketmq依賴(lài)

        <!--nacos客戶(hù)端-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>
 
        <!--rocketmq-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.7.1</version>
        </dependency>

2.2 修改主類(lèi)，啟動(dòng)nacos客戶(hù)端

package cn.jack;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 
@SpringBootApplication
@EnableDiscoveryClient
public class UserApplication {
    public static void main(String[] args) {
        SpringApplication.run(UserApplication.class);
    }
}

2.3 修改配置文件

rocketmq:
  name-server: 192.168.232.129:9876 # rocketmq服務(wù)的地址

2.4 編寫(xiě)消息接收服務(wù)

package cn.jack.service;
 
import cn.jack.domain.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
 
@Slf4j
@Service
@RocketMQMessageListener(consumerGroup = "jack-consumer-group", topic = "jack-topic")
public class SmsService implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        log.info("收到訂單信息，內(nèi)容如下：{}， 接下來(lái)進(jìn)行短信發(fā)送", order);
    }
}

2.5 測(cè)試

啟動(dòng)用戶(hù)微服務(wù)，收到了來(lái)自RocketMQ的消息。

2020-09-16 21:46:30.465 INFO [service-user,,,] 660 --- [MessageThread_2] cn.jack.service.SmsService : 收到訂單信息，內(nèi)容如下：Order(oid=66, uid=1, username=陳家寶, pid=2, pname=雪梨, pprice=2000.0, number=1)，接下來(lái)進(jìn)行短信發(fā)送
2020-09-16 21:46:30.465 INFO [service-user,,,] 660 --- [MessageThread_1] cn.jack.service.SmsService : 收到訂單信息，內(nèi)容如下：Order(oid=65, uid=1, username=陳家寶, pid=3, pname=華為, pprice=8000.0, number=1)，接下來(lái)進(jìn)行短信發(fā)送

消息成功消費(fèi)：