SpringBoot集成RocketMQ

實戰,用案例來講話

前面已經說了JMS和RocketMQ一些概念和安裝,下面使用SpringBoot來親身操做一下.

生產者的操做

1. SpringBoot項目建立完成,引入依賴是第一步:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>    
    <version>4.3.0</version>
</dependency>

1. 建立生產者是第二步,生產者必須依賴於生產組,並且須要指定nameServer

@Component
public class PayProducer {

    /**
     * 生產組,生產者必須在生產組內
     */
    private String producerGroup = "pay_group";

    /**
     * 端口
     */
    private String nameServer = "39.106.214.179:9876";


    private DefaultMQProducer producer;

    public PayProducer() {
        producer = new DefaultMQProducer(producerGroup);
        // 指定nameServer地址,多個地址之間以 ; 隔開
        producer.setNamesrvAddr(nameServer);
        start();
    }

    public DefaultMQProducer getProducer() {
        return producer;
    }

    /**
     * 對象在使用以前必須調用一次,而且只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 通常在應用上下文,使用上下文監聽器,進行關閉
     */
    public void shutdown() {
        producer.shutdown();
    }

}

1. 建立Controller進行測試發送消息,必需要指定topic,消息依賴於主題

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    /**
     * topic,消息依賴於topic
     */
    private static final String topic = "pay_test_topic";


    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        // 建立消息  主題   二級分類   消息內容好的字節數組
        Message message = new Message(topic, "taga", ("hello rocketMQ " + text).getBytes());

        SendResult send = payProducer.getProducer().send(message);

        System.out.println(send);

        return new HashMap<>();
    }

}

1. 採坑記錄
   • 上面完成就能夠啓動項目了,訪問以後報錯了:

   MQClientException: No route info of this topic, TopicTest1
   這個的緣由就是Broker禁止自動建立Topic且用戶沒有經過手動方式建立Topic, 或者是broker與Nameserver網絡不通
   解決:
       使用手動建立Topic,在RocketMQ控制檯的主題中建立就好,最主要的是指定topic name,以下圖
           出現建立不了的狀況往下看
   
       若是還出現這個問題,請關閉防火牆

   • 此次說下上面可能建立不了的問題,前面說了安裝開放安全組,此次就是由於rocketMQ虛擬的端口問題,須要開放10909,也就是說ECS最終開放的端口號: 8080,10911,9876,10909

   • 繼續採坑

   org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout
   
   這個問題是阿里雲服務器存在多個網卡,rocketMQ會根據當前網卡選擇一個IP使用,咱們須要制定一個IP:
       路徑是: /usr/local/software/rocketmq/distribution/target/apache-rocketmq
   
       vim ./conf/broker.conf
   
       添加配置: brokerIP1=公網IP
   
       從新啓動: 
           nohup sh bin/mqbroker -n localhost:9876 -c ./conf/broker.conf &
           tail -f nohup.out

   • 其餘問題

   https://blog.csdn.net/qq_14853889/article/details/81053145
       https://blog.csdn.net/wangmx1993328/article/details/81588217#%E5%BC%82%E5%B8%B8%E8%AF%B4%E6%98%8E
       https://www.jianshu.com/p/bfd6d849f156
       https://blog.csdn.net/wangmx1993328/article/details/81588217

消費者操做

1. 在前一個項目的基礎上,將公共內容提取出來,建立一個JsmConfig的類,來聲明公共內容:
   ```
   public class JmsConfig {

   /**
     * 端口
     */
    public static final String NAME_SERVER = "39.106.214.179:9876";
   
    /**
     * topic,消息依賴於topic
     */
    public static final String TOPIC = "pay_test_topic";

   }
   ```

2. 生產者內容變爲
   ```
   @Component
   public class PayProducer {

   /**
     * 生產組,生產者必須在生產組內
     */
    private String producerGroup = "pay_producer_group";
   
    private DefaultMQProducer producer;
   
    public PayProducer() {
        producer = new DefaultMQProducer(producerGroup);
        // 指定nameServer地址,多個地址之間以 ; 隔開
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        start();
    }
   
    public DefaultMQProducer getProducer() {
        return producer;
    }
   
    /**
     * 對象在使用以前必須調用一次,而且只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
   
    /**
     * 通常在應用上下文,使用上下文監聽器,進行關閉
     */
    public void shutdown() {
        producer.shutdown();
    }

   }
   ```

3. 建立消費者
   ```
   @Component
   public class PayConsumer {

   private DefaultMQPushConsumer consumer;
   
    private String consumerGroup = "pay_consumer_group";
   
    public PayConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        // 設置消費地點,從最後一個進行消費(其實就是消費策略)
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 訂閱主題的哪些標籤
        consumer.subscribe(JmsConfig.TOPIC, "*");
        // 註冊監聽器
        consumer.registerMessageListener((MessageListenerConcurrently)
                (msgs, context) -> {
                    try {
                        // 獲取Message
                        Message msg = msgs.get(0);
                        System.out.printf("%s Receive New Messages: %s %n",
                                Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                        String topic = msg.getTopic();
                        String body = new String(msg.getBody(), "utf-8");
                        // 標籤
                        String tags = msg.getTags();
                        String keys = msg.getKeys();
                        System.out.println("topic=" + topic + ", tags=" + tags + ",keys=" + keys + ", msg=" + body);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                });
        consumer.start();
        System.out.println("Consumer Listener");
    }

   }
   ```

4. Controller的變化:
   ```
   @RestController
   public class PayController {

   @Autowired
    private PayProducer payProducer;
   
   
   
    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        // 建立消息  主題   二級分類   消息內容好的字節數組
        Message message = new Message(JmsConfig.TOPIC, "taga", ("hello rocketMQ " + text).getBytes());
   
        SendResult send = payProducer.getProducer().send(message);
   
        System.out.println(send);
   
        return new HashMap<>();
    }

   }
   ```

梳理一下整個流程,生產者存在於生產組,因此生產組很重要,建立生產者須要指定生產組.消費者同理,建立消費者也須要指定消費組. 而且兩者都須要指定NameServer. 有了生產者就要發送消息,也就是Message,建立Message須要指定Topic,二級分類和消息體等信息. 那消費者如何獲取呢? 無非就是綁定Topic和二級分類就能夠,這就是整個流程. 中間少說了消息的存放,消息是在broker中,這個至關於倉庫,因此就是生產者生產消息到Broker,Consumer從Broker中獲取消息進行消費.