springcloud一springcloudalibaba集成rocketmq

jiezi

6 年前

在之前的工作中，微服务框架使用的是 springcloud，消息中间件使用的 rocketmq，这段时间看到阿里出了 spring cloud alibaba 集成了 rocketmq，出于好奇，写了个 demo

官方对 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).【生产者和消费者用于与目标绑定器通信的规范数据结构。】

第一步：下载：https://www.apache.org/dyn/cl…
第二步：解压
第三步：修改三个配置文件：runbroker.sh,runserver.sh,tools.sh, 将其中 JAVA_HOME 改成自己电脑的环境配置, 修改完如下

[! -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)!"

第四步：依次执行命令

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

如果启动成功，没有报错，代表启动成功哈，下面就可以开发了

第一步：导入相关的 pom

<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 项目，启动类如下：

@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);
    }
}

第四步：创建 consumer

@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，启动项目，访问接口

@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";
    }
}

会看到控制台输出：input receive: test rocketmq message

浏览器输入：http://127.0.0.1:8087/actuator/rocketmq-binder

这一篇文章只是将 spring cloud stream 和 rocketmq 跑通了，其实对于 spring cloud stream 和 rocketmq 还是学习的阶段，只能感叹 spring cloud 博大精深

更多网站可以访问 https://www.zplxjj.com 或关注公众号：

springcloud一springcloudalibaba集成rocketmq

前言

一些概念

快速在本地启动 rocketmq

开发 demo

Endpoint 信息查看

结语