Springboot整合RocketMQ

阿新 • • 發佈：2018-12-12

官方版本的RocketMQ4.3，因為本文就是基於RocketMQ4.3的，”RocketMQ4.3正式釋出支援了分散式事務”。這一訊息讓很多開發者躍躍欲試，以前rocketMQ沒有將分散式事務作為一個發行版的部分，基本都是開發者自己實現的事務部分。下面我們從最基礎的安裝RocketMQ開始，再介紹程式碼中如何實現基本的配置。

下載安裝RocketMQ

$ wget http://mirrors.hust.edu.cn/apache/rocketmq/4.3.0/rocketmq-all-4.3.0-source-release.zip
$ unzip rocketmq-all-4.3.0-source-release.zip
$ mvn -Prelease-all -DskipTests clean install -U
$ cd distribution/target/apache-rocketmq

啟動NameServer，啟動後NameServer的埠是9876，請確保自己的9876埠未被佔用

$ nohup sh bin/mqnamesrv &
$ tail -f ~/logs/rocketmqlogs/namesrv.log
  The Name Server boot success...

啟動Broker

$ nohup sh bin/mqbroker -n localhost:9876 &
$ tail -f ~/logs/rocketmqlogs/broker.log 
  The broker[%s, 172.30.30.233:10911] boot success...

啟動成功了之後我們就可以建立新的Springboot專案了，如何建立專案這裡我就不在介紹了，Eclipse和Idea的方式大同小異，目錄結構基本都是一樣的。

首先每次說到Springboot的專案都是要先講講這個boot的配置，按照慣例呢我先給配置，配置上都有每一行配置的註釋，大家可以參考。

程式碼示例pom.xml

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.4.RELEASE</version>
        <relativePath/>
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

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

        <!--RocketMQ的dependency，目前官方還沒有boot-starter-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.3.0</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>RELEASE</version>
        </dependency>
    </dependencies>

application.yml

apache:
  rocketmq:
#消費者的配置
    consumer:
      pushConsumer: XiaourPushConsumer
#生產者的配置
    producer:
      producerGroup: Xiaour
$Nameserver的地址,這裡配置你MQ安裝的機器上的IP就好，我這裡在本機安裝的
    namesrvAddr: 127.0.0.1:9876

Producer 訊息生產者

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;

/**
 * @Author: Xiaour
 * @Description:生產者
 * @Date: 2018/8/9 14:52
 */

@Component
public class Producer {

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

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

    @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()+"\"}";
    }
}

Consumer 訊息消費者

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;

/**
 * @Author: Xiaour
 * @Description:消費者
 * @Date: 2018/8/9 14:51
 */

@Component
public class Consumer implements CommandLineRunner {

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

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


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

        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("SpringBootRocketMqGroup");

        consumer.setNamesrvAddr(namesrvAddr);
        try {

            // 訂閱PushTopic下Tag為push的訊息,都訂閱訊息
            consumer.subscribe("PushTopic", "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();
    }
}

測試介面用例

這裡我們用一個介面來測試我們的訊息傳送會不會被消費者接收。

@RestController
public class TestController {
    @Autowired
    private Producer producer;

    @RequestMapping("/push")
    public String pushMsg(String msg){
        try {
            return producer.send("PushTopic","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";
    }
}

在Springboot啟動類啟動後,在瀏覽器訪問http://127.0.0.1:8080/push?hello，頁面提示{"MsgId":"AC100AB660C618B4AAC2XXXXXXXX"}就表示訊息傳送成功啦。
我們可以再IDE控制檯中看到輸出的結果，

傳送響應：MsgId:AC100AB660C618B4AAC2XXXXXXXX，傳送狀態:SEND_OK
接收到了訊息：hello

這時候我們的整合基本上就完成啦。

專案連結：https://github.com/zhonghuaxiaotiao/rocketmq.git

Springboot整合RocketMQ

下載安裝RocketMQ

啟動NameServer，啟動後NameServer的埠是9876，請確保自己的9876埠未被佔用

啟動Broker

程式碼示例pom.xml

application.yml

Producer 訊息生產者

Consumer 訊息消費者

測試介面用例

Springboot整合RocketMQ

SpringBoot 整合 RocketMQ 消費資料存庫(MySQL)

MQ訊息佇列三（SpringBoot 整合rocketMq）

SpringBoot整合RocketMQ事務訊息

SpringBoot(17）---SpringBoot整合RocketMQ

SpringBoot與RocketMQ客戶端整合原理解讀與示例

springboot乾貨——（二十）整合rocketmq

Springboot整合redisCluster集群

springboot整合mybatis,freemarker

springboot整合activemq小demo

SpringBoot(十四)：springboot整合shiro-登錄認證和權限管理

springboot整合mybatis及封裝curd操作-配置文件

SpringBoot 2.SpringBoot整合Mybatis

SpringBoot 4.SpringBoot 整合 devtools 實現熱部署

SpringBoot整合MyBatis及Thymeleaf

springboot整合shiro應用

SpringBoot整合Quartz定時任務的簡單實例 2

springboot整合系列

springboot整合hibernate案例

SpringBoot整合Swagger2簡單的例子

Springboot整合RocketMQ

下載安裝RocketMQ

啟動NameServer，啟動後NameServer的埠是9876，請確保自己的9876埠未被佔用

啟動Broker

程式碼示例pom.xml

application.yml

Producer 訊息生產者

Consumer 訊息消費者

測試介面用例

相關推薦