SpringBoot整合rockerMQ消息隊(duì)列的方法

這篇文章主要介紹“SpringBoot整合rockerMQ消息隊(duì)列的方法”，在日常操作中，相信很多人在SpringBoot整合rockerMQ消息隊(duì)列的方法問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對(duì)大家解答”SpringBoot整合rockerMQ消息隊(duì)列的方法”的疑惑有所幫助！接下來，請(qǐng)跟著小編一起來學(xué)習(xí)吧！

Springboot整合RockerMQ

1、maven依賴

<dependencies>
    <!-- springboot-web組件 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.0.3</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

2、yml配置文件

rocketmq:
  ###連接地址nameServer
  name-server: www.kaicostudy.com:9876;
  producer:
    group: kaico_producer
server:
  port: 8088

3、生產(chǎn)者

@RequestMapping("/sendMsg")
    public String sendMsg() {
        OrderEntity orderEntity = new OrderEntity("123456","騰訊視頻會(huì)員");
        SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);
        System.out.println("返回發(fā)送消息狀態(tài)：" + kaicoTopic);
        return "success";
    }

4、消費(fèi)者

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag1", consumerGroup = "kaico_consumer", messageModel = MessageModel.CLUSTERING)
public class OrdeConsumer2 implements RocketMQListener<OrderEntity> {
    @Override
    public void onMessage(OrderEntity o) {
        System.out.println("kaico_consumer2消費(fèi)者接收對(duì)象:" + o.toString());
    }
}

使用總結(jié)

消費(fèi)模式

集群消費(fèi)
當(dāng) consumer 使用集群消費(fèi)時(shí)，每條消息只會(huì)被 consumer 集群內(nèi)的任意一個(gè) consumer 實(shí)例消費(fèi)一次。
同時(shí)記住一點(diǎn)，使用集群消費(fèi)的時(shí)候，consumer 的消費(fèi)進(jìn)度是存儲(chǔ)在 broker 上，consumer 自身是不存儲(chǔ)消費(fèi)進(jìn)度的。消息進(jìn)度存儲(chǔ)在 broker 上的好處在于，當(dāng)你 consumer 集群是擴(kuò)大或者縮小時(shí)，由于消費(fèi)進(jìn)度統(tǒng)一在broker上，消息重復(fù)的概率會(huì)被大大降低了。
注意： 在集群消費(fèi)模式下，并不能保證每一次消息失敗重投都投遞到同一個(gè) consumer 實(shí)例。

注解配置：messageModel = MessageModel.CLUSTERING

廣播消費(fèi)
當(dāng) consumer 使用廣播消費(fèi)時(shí)，每條消息都會(huì)被 consumer 集群內(nèi)所有的 consumer 實(shí)例消費(fèi)一次，也就是說每條消息至少被每一個(gè) consumer 實(shí)例消費(fèi)一次。
與集群消費(fèi)不同的是，consumer 的消費(fèi)進(jìn)度是存儲(chǔ)在各個(gè) consumer 實(shí)例上，這就容易造成消息重復(fù)。還有很重要的一點(diǎn)，對(duì)于廣播消費(fèi)來說，是不會(huì)進(jìn)行消費(fèi)失敗重投的，所以在 consumer 端消費(fèi)邏輯處理時(shí)，需要額外關(guān)注消費(fèi)失敗的情況。
雖然廣播消費(fèi)能保證集群內(nèi)每個(gè) consumer 實(shí)例都能消費(fèi)消息，但是消費(fèi)進(jìn)度的維護(hù)、不具備消息重投的機(jī)制大大影響了實(shí)際的使用。因此，在實(shí)際使用中，更推薦使用集群消費(fèi)，因?yàn)榧合M(fèi)不僅擁有消費(fèi)進(jìn)度存儲(chǔ)的可靠性，還具有消息重投的機(jī)制。而且，我們通過集群消費(fèi)也可以達(dá)到廣播消費(fèi)的效果。

注解配置：messageModel = MessageModel.BROADCASTING

生產(chǎn)者組和消費(fèi)者組

生產(chǎn)者組
一個(gè)生產(chǎn)者組，代表著一群topic相同的Producer。即一個(gè)生產(chǎn)者組是同一類Producer的組合。

如果Producer是TransactionMQProducer，則發(fā)送的是事務(wù)消息。如果節(jié)點(diǎn)1發(fā)送完消息后，消息存儲(chǔ)到broker的Half Message Queue中，還未存儲(chǔ)到目標(biāo)topic的queue中時(shí)，此時(shí)節(jié)點(diǎn)1崩潰，則可以通過同一Group下的節(jié)點(diǎn)2進(jìn)行二階段提交，或回溯。

使用時(shí)，一個(gè)節(jié)點(diǎn)下，一個(gè)topic會(huì)對(duì)應(yīng)一個(gè)producer

消費(fèi)者組
一個(gè)消費(fèi)者組，代表著一群topic相同，tag相同（即邏輯相同）的Consumer。通過一個(gè)消費(fèi)者組，則可容易的進(jìn)行負(fù)載均衡以及容錯(cuò)

使用時(shí)，一個(gè)節(jié)點(diǎn)下，一個(gè)topic加一個(gè)tag可以對(duì)應(yīng)一個(gè)consumer。一個(gè)消費(fèi)者組就是橫向上多個(gè)節(jié)點(diǎn)的相同consumer為一個(gè)消費(fèi)組。

首先分析一下producer。習(xí)慣上我們不會(huì)創(chuàng)建多個(gè)訂閱了相同topic的Producer實(shí)例，因?yàn)橐粋€(gè)Producer實(shí)例發(fā)送消息時(shí)是通過ExecutorService線程池去異步執(zhí)行的，不會(huì)阻塞完全夠用，如果創(chuàng)建了多個(gè)相同topic的Producer則會(huì)影響性能。而Consumer則不同。消息會(huì)在一topic下會(huì)細(xì)分多個(gè)tag，需要針對(duì)tag需要針對(duì)不同的tag創(chuàng)建多個(gè)消費(fèi)者實(shí)例。

注意：多個(gè)不同的消費(fèi)者組訂閱同一個(gè)topic、tag，如果設(shè)定的是集群消費(fèi)模式，每一個(gè)消費(fèi)者組中都會(huì)有一個(gè)消費(fèi)者來消費(fèi)。也就是說不同的消費(fèi)者組訂閱同一個(gè)topic相互之間是沒有影響的。

生產(chǎn)者投遞消息的三種方式

同步： 發(fā)送消息后需等待結(jié)果，消息的可靠性高發(fā)送速度慢；

 SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);

異步： 消息發(fā)送后，回調(diào)通知結(jié)果，消息發(fā)送速度快，消息可靠性低；

 //異步發(fā)送
        rocketMQTemplate.asyncSend("kaicoTopic" + ":" + "tag1", orderEntity, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("異步發(fā)送消息成功");
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("異步發(fā)送消息失敗");
            }
        });

單向(oneway)：消息發(fā)送后，不關(guān)心結(jié)果，發(fā)送速度最快，消息可靠性最差，適用于在大量日志數(shù)據(jù)和用戶行為數(shù)據(jù)等場景發(fā)送數(shù)據(jù)。

//單向(oneway)發(fā)送
        rocketMQTemplate.sendOneWay("kaicoTopic"+":"+"tag1", orderEntity);

如何保證消息不丟失

主要三個(gè)步驟
1、生產(chǎn)者保證消息發(fā)送成功
采用同步發(fā)送消息的方式，發(fā)送消息后有返回結(jié)果，保證消息發(fā)送成功。（代碼見上面）
返回四個(gè)狀態(tài)

• SEND_OK：消息發(fā)送成功。需要注意的是，消息發(fā)送到 broker 后，還有兩個(gè)操作：消息刷盤和消息同步到 slave 節(jié)點(diǎn)，默認(rèn)這兩個(gè)操作都是異步的，只有把這兩個(gè)操作都改為同步，SEND_OK 這個(gè)狀態(tài)才能真正表示發(fā)送成功。

• FLUSH_DISK_TIMEOUT：消息發(fā)送成功但是消息刷盤超時(shí)。

• FLUSH_SLAVE_TIMEOUT：消息發(fā)送成功但是消息同步到 slave 節(jié)點(diǎn)時(shí)超時(shí)。

• SLAVE_NOT_AVAILABLE：消息發(fā)送成功但是 broker 的 slave 節(jié)點(diǎn)不可用。

2、rocketMQ將消息持久化，保證宕機(jī)后消息不會(huì)丟失。持久化策略（刷盤策略）

• 異步刷盤：默認(rèn)。消息寫入 CommitLog 時(shí)，并不會(huì)直接寫入磁盤，而是先寫入 PageCache 緩存后返回成功，然后用后臺(tái)線程異步把消息刷入磁盤。異步刷盤提高了消息吞吐量，但是可能會(huì)有消息丟失的情況，比如斷點(diǎn)導(dǎo)致機(jī)器停機(jī)，PageCache 中沒來得及刷盤的消息就會(huì)丟失。

• 同步刷盤：消息寫入內(nèi)存后，立刻請(qǐng)求刷盤線程進(jìn)行刷盤，如果消息未在約定的時(shí)間內(nèi)(默認(rèn) 5 s)刷盤成功，就返回 FLUSH_DISK_TIMEOUT，Producer 收到這個(gè)響應(yīng)后，可以進(jìn)行重試。同步刷盤策略保證了消息的可靠性，同時(shí)降低了吞吐量，增加了延遲。要開啟同步刷盤，需要增加下面配置：

flushDiskType=SYNC_FLUSH

3、Broker 多副本和高可用
Broker 為了保證高可用，采用一主多從的方式部署。
消息發(fā)送到 master 節(jié)點(diǎn)后，slave 節(jié)點(diǎn)會(huì)從 master 拉取消息保持跟 master 的一致。這個(gè)過程默認(rèn)是異步的，即 master 收到消息后，不等 slave 節(jié)點(diǎn)復(fù)制消息就直接給 Producer 返回成功。

這樣會(huì)有一個(gè)問題，如果 slave 節(jié)點(diǎn)還沒有完成消息復(fù)制，這時(shí) master 宕機(jī)了，進(jìn)行主備切換后就會(huì)有消息丟失。為了避免這個(gè)問題，可以采用 slave 節(jié)點(diǎn)同步復(fù)制消息，即等 slave 節(jié)點(diǎn)復(fù)制消息成功后再給 Producer 返回發(fā)送成功。只需要增加下面的配置：
brokerRole=SYNC_MASTER

改為同步復(fù)制后，消息復(fù)制流程如下：

• slave 初始化后，跟 master 建立連接并向 master 發(fā)送自己的 offset；

• master 收到 slave 發(fā)送的 offset 后，將 offset 后面的消息批量發(fā)送給 slave；

• slave 把收到的消息寫入 commitLog 文件，并給 master 發(fā)送新的 offset；

• master 收到新的 offset 后，如果 offset >= producer 發(fā)送消息后的 offset，給 Producer 返回 SEND_OK。

4、消費(fèi)者保證消息消費(fèi)成功
消費(fèi)者消費(fèi)消息后，如果 Consumer 消費(fèi)成功，返回 CONSUME_SUCCESS，提交 offset 并從 Broker 拉取下一批消息。

@Service
public class NoSpringBootOrderConsumer {
    private DefaultMQPushConsumer defaultMQPushConsumer;
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    protected String consumerGroup;
    protected String topic;
    protected String topicTag;
    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }
    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }
    public void setTopic(String topic) {
        this.topic = topic;
    }
    public void setTopicTag(String topicTag) {
        this.topicTag = topicTag;
    }
    public static String encoding = System.getProperty("file.encoding");
    /*
     * @Author ex_fengkai
     * @Description //TODO 初始化數(shù)據(jù)（消費(fèi)者組名稱、topic、topic的tag、nameServer的信息）
     * @Date 2020/11/9 14:36
     * @Param []
     * @return void
     **/
    private void initParam() {
        this.consumerGroup = "kaico_consumer3";
        this.topic = "kaicoTopic";
        this.topicTag = "tag1";
        this.setNamesrvAddr(namesrvAddr);
    }
    @PostConstruct
    private void init() throws InterruptedException, MQClientException {
        initParam();
        // ConsumerGroupName需要由應(yīng)用來保證唯一，用于把多個(gè)Consumer組織到一起，提高并發(fā)處理能力
        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr); //設(shè)置nameServer服務(wù)器
        defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        defaultMQPushConsumer.setVipChannelEnabled(false);
        // 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 訂閱指定Topic下的topicTag
        System.out.println("consumerGroup:" + consumerGroup + " topic:" + topic + " ,topicTag:" + topicTag);
        defaultMQPushConsumer.subscribe(topic, topicTag);
        // 設(shè)置為集群消費(fèi)
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        // 通過匿名消息監(jiān)聽處理消息消費(fèi)
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            // 默認(rèn)msgs里只有一條消息，可以通過設(shè)置consumeMessageBatchMaxSize參數(shù)來批量接收消息
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                if (msg.getTopic().equals(topic) && msg.getTags() != null && msg.getTags().equals(topicTag)) {
                    // 執(zhí)行topic下對(duì)應(yīng)tag的消費(fèi)邏輯
                    try {
                        onMessage(new String(msg.getBody(),"utf-8"));

                    } catch (UnsupportedEncodingException e) {
                        System.out.println("系統(tǒng)不支持消息編碼格式:" + encoding);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                    } catch (Exception e) {
                        System.out.println("消息處理異常");
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    System.out.println("consumerGroup:" + consumerGroup + " MsgId:" + msg.getMsgId() + " was done!");
                }
                // 如果沒有return success ，consumer會(huì)重新消費(fèi)該消息，直到return success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Consumer對(duì)象在使用之前必須要調(diào)用start初始化，初始化一次即可
        defaultMQPushConsumer.start();
        System.out.println("consumerGroup:" + consumerGroup + " namesrvAddr:" + namesrvAddr + "  start success!");
    }
    @PreDestroy
    public void destroy() {
        defaultMQPushConsumer.shutdown();
    }
    private void onMessage(String s) {
        System.out.println(consumerGroup + "用spring的方式的消費(fèi)者消費(fèi)：" + s);
    }
}

Consumer 重試
Consumer 消費(fèi)失敗，這里有 3 種情況：

• 返回 RECONSUME_LATER

• 返回 null

• 拋出異常

Broker 收到這個(gè)響應(yīng)后，會(huì)把這條消息放入重試隊(duì)列，重新發(fā)送給 Consumer。

注意：Broker 默認(rèn)最多重試 16 次，如果重試 16 次都失敗，就把這條消息放入死信隊(duì)列，Consumer 可以訂閱死信隊(duì)列進(jìn)行消費(fèi)。重試只有在集群模式（MessageModel.CLUSTERING）下生效，在廣播模式（MessageModel.BROADCASTING）下是不生效的。Consumer 端一定要做好冪等處理。

順序消息

• 生產(chǎn)者投遞消息根據(jù)key投遞到同一個(gè)隊(duì)列中存放

• 消費(fèi)者應(yīng)該訂閱到同一個(gè)隊(duì)列實(shí)現(xiàn)消費(fèi)

• 最終應(yīng)該使用同一個(gè)線程去消費(fèi)消息(不能夠?qū)崿F(xiàn)多線程消費(fèi)。)

生產(chǎn)者代碼

//發(fā)送順序消息
    @RequestMapping("/sendMsg1")
    public String sendMsg1() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Long orderId = System.currentTimeMillis();
        String insertSql = getSqlMsg("insert", orderId);
        String updateSql = getSqlMsg("update", orderId);
        String deleteSql = getSqlMsg("delete", orderId);
        Message insertMsg = new Message("kaicoTopic", "tag6", insertSql.getBytes());
        Message updateMsg = new Message("kaicoTopic", "tag6", updateSql.getBytes());
        Message deleteMsg = new Message("kaicoTopic", "tag6", deleteSql.getBytes());
        DefaultMQProducer producer = rocketMQTemplate.getProducer();
        rocketMQTemplate.getProducer().send(insertMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return  mqs.get(0);
                    }
                }, orderId);
        rocketMQTemplate.getProducer().send(updateMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return mqs.get(0);
                    }
                }, orderId);
        rocketMQTemplate.getProducer().send(deleteMsg
                , new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                               Object arg) {
                        // 該消息存放到隊(duì)列0中
                        return  mqs.get(0);
                    }
                }, orderId);
        return orderId + "";
    }

消費(fèi)者代碼

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag6", consumerGroup = "kaico_consumer1",
        messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.ORDERLY, consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt msg) {
        System.out.println(Thread.currentThread().getName() + "-kaico_consumer1消費(fèi)者接收對(duì)象:隊(duì)列" + msg.getQueueId()
                + "=消息：" +  new String(msg.getBody()));
    }
}

分布式事務(wù)

實(shí)現(xiàn)思路

• 生產(chǎn)者（發(fā)送方）投遞事務(wù)消息到Broker中，設(shè)置該消息為半消息 不可以被消費(fèi)；

• 開始執(zhí)行我們的本地事務(wù)，將本地事務(wù)執(zhí)行的結(jié)果（回滾或者提交）發(fā)送給Broker

• Broker獲取回滾或者提交，如果是回滾的情況則刪除該消息、如果是提交的話，該消息就可以被消費(fèi)者消費(fèi)；

• Broker如果沒有及時(shí)的獲取發(fā)送方本地事務(wù)結(jié)果的話，會(huì)主動(dòng)查詢本地事務(wù)結(jié)果。

1、生產(chǎn)者發(fā)送事務(wù)消息sendMessageInTransaction

 public String saveOrder() {
     // 提前生成我們的訂單id
     String orderId = System.currentTimeMillis() + "";
     /**
      * 1.提前生成我們的半消息
      * 2.半消息發(fā)送成功之后，在執(zhí)行我們的本地事務(wù)
      */
     OrderEntity orderEntity = createOrder(orderId);
     String msg = JSONObject.toJSONString(orderEntity);
     MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(msg);
     stringMessageBuilder.setHeader("msg", msg);
     Message message = stringMessageBuilder.build();
     // 該消息不允許被消費(fèi)者消費(fèi)，生產(chǎn)者的事務(wù)邏輯代碼在生產(chǎn)者事務(wù)監(jiān)聽類中executeLocalTransaction方法中執(zhí)行。
     rocketMQTemplate.sendMessageInTransaction("kaicoProducer",
             "orderTopic", message, null);
     return orderId;

}

2、事務(wù)監(jiān)聽類

@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "kaicoProducer") //這個(gè)mayiktProducer生產(chǎn)者的事務(wù)管理
public class SyncProducerListener implements RocketMQLocalTransactionListener {
    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private TransationalUtils transationalUtils;
    /**
     * 執(zhí)行我們訂單的事務(wù)
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        //拿到消息
        Object object = headers.get("msg");
        if (object == null) {
            return null;
        }
        String orderMsg = (String) object;
        OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
        TransactionStatus begin = null;
        try {
            begin = transationalUtils.begin();
            int result = orderMapper.addOrder(orderEntity);
            transationalUtils.commit(begin);
            if (result <= 0) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            // 告訴我們的Broke可以消費(fèi)者該消息
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            if (begin != null) {
                transationalUtils.rollback(begin);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        //add.Order
        return null;
    }
    /**
     * 提供給我們的Broker定時(shí)檢查
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        Object object = headers.get("msg");
        if (object == null) {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        String orderMsg = (String) object;
        OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
        String orderId = orderEntity.getOrderId();
        // 直接查詢我們的數(shù)據(jù)庫
        OrderEntity orderDbEntity = orderMapper.findOrderId(orderId);
        if (orderDbEntity == null) {
            //不確認(rèn)，繼續(xù)重試
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        //提交事務(wù)
        return RocketMQLocalTransactionState.COMMIT;
    }
}

3、消費(fèi)者消費(fèi)消息

@Service
@RocketMQMessageListener(topic = "orderTopic", consumerGroup = "kaicoTopic")
public class OrdeConsumer implements RocketMQListener<String> {
    @Autowired
    private DispatchMapper dispatchMapper;
    @Override
    public void onMessage(String msg) {
        OrderEntity orderEntity = JSONObject.parseObject(msg, OrderEntity.class);
        String orderId = orderEntity.getOrderId();
        // 模擬userid為=123456
        DispatchEntity dispatchEntity = new DispatchEntity(orderId, 123456L);
        dispatchMapper.insertDistribute(dispatchEntity);
    }
}

到此，關(guān)于“SpringBoot整合rockerMQ消息隊(duì)列的方法”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！