springboot之RocketMq實現

環境：win10

1、下載安裝包

http://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.4.0/rocketmq-all-4.4.0-bin-release

2、解壓到本地，並到系統環境中設置環境變量(必須添加，否則啟動不了，也會提示你添加)

3、進入bin目錄，運行broker和nameserver

start mqnamesrv.cmd
start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

4、下載mq的監控web

https://github.com/apache/rocketmq-externals/archive/master.zip

 5、web界面進行配置

進入 D:\software\rocketmq\rocketmq-externals-master\rocketmq-console\src\main\resources，
找到application.properties文件

6、編譯打包

進入 D:\software\rocketmq\rocketmq-externals-master\rocketmq-console 
運行mvn clean package -Dmaven.test.skip=true

7、運行編譯好的jar包

進入D:\software\rocketmq\rocketmq-externals-master\rocketmq-console\target
運行java -jar rocketmq-console-ng-1.0.0.jar

8、查看web監控界面

http://localhost:8080

 

#########springboot集成###########

1、pom.xml文件添加依賴

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

2、yml文件添加rocketmq配置

apache:
  rocketmq:
    #消費者的配置
    consumer:
      pushConsumer: myConsumer
    #生產者的配置
    producer:
      producerGroup: myGroup
    namesrvAddr: 127.0.0.1:9876

3、生產者類RocketProducer

import org.apache.commons.lang3.time.StopWatch;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

@Component
public class RocketProducer {

    /**
     * 生產者的組名
     */
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup; //myGroup

    private DefaultMQProducer producer;
    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr; //127.0.0.1:9876

    @PostConstruct
    public void defaultMQProducer() {

        //生產者的組名
        producer= new DefaultMQProducer(producerGroup);
        //指定NameServer地址，多個地址以 ; 隔開
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);
        try {
            producer.start();
            System.out.println("-------->:producer啟動了");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public String send(String topic,String tags,String body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        StopWatch stop = new StopWatch();
        stop.start();
        SendResult result = producer.send(message);
        System.out.println("發送響應：MsgId:" + result.getMsgId() + "，發送狀態:" + result.getSendStatus());
        stop.stop();
        return "{\"MsgId\":\""+result.getMsgId()+"\"}";
    }
}

4、消費者類RocketConsumer

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;


@Component
public class RocketConsumer implements CommandLineRunner {

    /**
     * 消費者
     */
    @Value("${apache.rocketmq.consumer.pushConsumer}")
    private String pushConsumer;  //myConsumer

    /**
     * NameServer 地址
     */
    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr; //127.0.0.1:9876


    /**
     * 初始化RocketMq的監聽信息，渠道信息
     */
    public void messageListener(){

        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("SpringBootRocketMqGroup");

        consumer.setNamesrvAddr(namesrvAddr);
        try {

            // 訂閱PushTopic下Tag為push的消息,都訂閱消息
            consumer.subscribe("firstPushTopic", "push");

            // 程序第一次啟動從消息隊列頭獲取數據
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //可以修改每次消費消息的數量，默認設置是每次消費一條
            consumer.setConsumeMessageBatchMaxSize(1);

            //在此監聽中消費信息，並返回消費的狀態信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {

                // 會把不同的消息分別放置到不同的隊列中
                for(Message msg:msgs){

                    System.out.println("接收到了消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });

            consumer.start();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run(String... args) throws Exception {
        this.messageListener();
    }
}

5、controller中編寫發送消息

 @Autowired
    private RocketProducer producer;

    @RequestMapping("/myFirstProducer")
    public String pushMsg(String msg){
        try {
            return producer.send("firstPushTopic","push",msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return "ERROR";
    }

完畢！

自己訪問，並通過web界面訪問

 

FAQ

1、出現磁盤不足問題,rocketmq默認的磁盤水位為88%，但是本人的磁盤使用是0.94了，所以會報這種錯。

org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [32]ms, Topic: firstPushTopic, BrokersSent: [DESKTOP-N10172E, DESKTOP-N10172E, DESKTOP-N10172E]
See http://rocketmq.apache.org/docs/faq/ for further details.
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:632)
Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 14  DESC: service not available now, maybe disk full, CL:  0.94 CQ:  0.94 INDEX:  0.94, maybe your broker machine memory too small.
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.processSendResponse(MQClientAPIImpl.java:556)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageSync(MQClientAPIImpl.java:358)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:340)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:294)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendKernelImpl(DefaultMQProducerImpl.java:807)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:551)
    ... 59 more

解決方法：找到runbroker.cmd文件，在文件中添加

set "JAVA_OPT=%JAVA_OPT% -Drocketmq.broker.diskSpaceWarningLevelRatio=0.98"