springcloud(一)——spring-cloud-alibaba集成rocketmq

時間 2019-12-06

標籤 springcloud spring cloud alibaba 集成 rocketmq 欄目 Spring 简体版

原文原文鏈接

前言

在以前的工做中，微服務框架使用的是springcloud，消息中間件使用的rocketmq，這段時間看到阿里出了spring cloud alibaba集成了rocketmq，出於好奇，寫了個demojava

一些概念

官方對 Spring Cloud Stream 的一段介紹：Spring Cloud Stream 是一個用於構建基於消息的微服務應用框架。基於 SpringBoot 建立具備生產級別的單機 Spring 應用，而且使用 Spring Integration 與 Broker 進行鏈接。
Binder ：Components responsible to provide integration with the external messaging systems.【與外部消息中間件進行集成】
Binding：Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).【在消息中間件與應用程序提供的 Provider 和 Consumer 之間提供了一個橋樑，開發者只需使用應用程序的生產者或消費者生產或消費數據便可，屏蔽了開發者與底層消息中間件的接觸。】
Message: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).【生產者和消費者用於與目標綁定器通訊的規範數據結構。】

快速在本地啓動rocketmq

第一步：下載：www.apache.org/dyn/closer.… 第二步：解壓第三步：修改三個配置文件：runbroker.sh,runserver.sh,tools.sh,將其中JAVA_HOME改爲本身電腦的環境配置,修改完以下web

[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=本身的地址
#[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"
複製代碼

第四步：依次執行命令spring

./mqnamesrv
./mqbroker -n localhost:9876
./mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic
複製代碼

若是啓動成功，沒有報錯，表明啓動成功哈，下面就能夠開發了apache

開發demo

第一步：導入相關的pomjson

<dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>0.2.1.RELEASE</version>
                <type>pom</type>
                <scope>import</scope>
 </dependency>
複製代碼

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
         <!-- 爲了Endpoint 信息查看 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>io.dropwizard.metrics</groupId>
            <artifactId>metrics-core</artifactId>
            <version>3.2.6</version>
        </dependency>
複製代碼

第二步：建一個springboot項目，啓動類以下：api

@SpringBootApplication
@EnableBinding({ Source.class, Sink.class })
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}
複製代碼

第三步：建立provider瀏覽器

@Service
public class RocketmqProducer {
    public void send(String message) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("test_producer_group");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        Message msg = new Message("test-topic", "test-tag", message.getBytes());
        producer.send(msg);
    }
}
複製代碼

第四步：建立consumerspringboot

@Service
public class ReceiveService {

    /** * 默認是input，在Sink類中指定，若是想要多個input，須要寫一個實現Sink的類 * @param receiveMsg */
    @StreamListener("input")
    public void receiveInput1(String receiveMsg) {
        System.out.println("input receive: " + receiveMsg);
    }

}
複製代碼

第五步：加入配置文件：數據結構

server.port=8087
spring.application.name=spring-cloud-alibaba-rocketmq-demo

# 配置rocketmq的nameserver地址
spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876
# 定義name爲output的binding
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=application/json
#定義name爲input的binding
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=application/json
spring.cloud.stream.bindings.input.group=test-group

management.endpoint.health.show-details=always
複製代碼

第六步：寫一個controller，啓動項目，訪問接口app

@RestController
@RequestMapping(value = "/api/demo/test")
public class TestController {

    @Autowired
    RocketmqProducer rocketmqProducer;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String send() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        rocketmqProducer.send("test rocketmq message");
        return "success";
    }
}
複製代碼