基于rocketmq-spring整合RocketMQ和Springboot

前言

之前寫過一篇RocketMQ整合Springboot的文章,但那種方式還需要自己上傳jar包,較為麻煩。最近偶然的機(jī)會(huì)發(fā)現(xiàn)github上的rocketmq-externals工程結(jié)構(gòu)變了,找不到rocketmq-spring-boot-starter目錄了,一番搜索之后發(fā)現(xiàn)apache下面多了個(gè)rocketmq-spring項(xiàng)目(https://github.com/apache/rocketmq-spring),項(xiàng)目描述為——幫助開發(fā)者在Spring Boot中快速集成RocketMQ,那么很有必要來用一下官方產(chǎn)品了。

依賴及配置

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>

這里的2.0.3是截止目前為止的最新版本,該版本依賴了spring-boot的2.0.5.RELEASE版本,因此請(qǐng)先保證自己的spring-boot版本。另外,該版本依賴rocketmq-client的4.5.1版本,該rocketmq版本已經(jīng)支持ACL功能,因此為了安全考慮這個(gè)功能理所當(dāng)然要開起來,具體配置如下:

rocketmq:
  name-server: 127.0.0.1:9876;127.0.0.2:9876
  producer:
    group: xht-group
    access-key: rocketmq
    secret-key: 12345678

合理主要是對(duì)生產(chǎn)者的信息進(jìn)行配置,并不包含消費(fèi)者的部分。

生產(chǎn)者端

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;

@Slf4j
@RestController
public class ScheduledMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @RequestMapping(value = "send")
    public String send(String msg, int level){
        //普通消息發(fā)送
        rocketMQTemplate.convertAndSend("delay-message-test", msg);
        Message<String> message = new GenericMessage<>(msg);
        //延時(shí)消息發(fā)送,level對(duì)應(yīng)的默認(rèn)延時(shí)時(shí)間依次為1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        rocketMQTemplate.syncSend("delay-message-test",message,1000, level);
        log.info("消息已發(fā)送:{}",msg);
        return "success";
    }

}

消息的發(fā)送主要通過RocketMQTemplate進(jìn)行,該對(duì)象提供了非常多已經(jīng)封裝好的基礎(chǔ)方法供調(diào)用,這里只以普通消息發(fā)送,和同步延時(shí)消息的發(fā)送做為例子。

消費(fèi)者端

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RocketMQMessageListener(topic = "delay-message-test", consumerGroup = "xht-group", accessKey = "rocketmq", secretKey = "12345678")
public class ScheduledMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        log.info("received message: {}", message);
    }
}

消費(fèi)者端的實(shí)現(xiàn)非常簡(jiǎn)單,配置參數(shù)都集中于@RocketMQMessageListener注解上。

總結(jié)

對(duì)比我大半年前的文章可以看出,目前rocketmq與springboot的集成越來越容易了,說明rocketmq這個(gè)產(chǎn)品還是有很好前景的,之前處于各種原因考慮沒有在生產(chǎn)環(huán)境使用,這下真的可以用起來了。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡(jiǎn)書系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容