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

栏目: Java · 发布时间: 6年前

内容简介:在之前的工作中,微服务框架使用的是springcloud,消息中间件使用的rocketmq,这段时间看到阿里出了spring cloud alibaba集成了rocketmq,出于好奇,写了个demo第一步:下载:第二步:解压

在之前的工作中,微服务框架使用的是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).【生产者和消费者用于与目标绑定器通信的规范数据结构。】

快速在本地启动rocketmq

第一步:下载: https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip

第二步:解压

第三步:修改三个配置文件: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

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

开发demo

第一步:导入相关的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


以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

失控

失控

[美]凯文·凯利(Kevin Kelly) / 张行舟 等 / 译言·东西文库/电子工业出版社 / 2016-1 / 89.00元

《失控:全人类的最终命运和结局》(全新修订本)是一部思考人类社会(或更一般意义上的复杂系统)进化的“大部头”著作,对于那些不惧于“头脑体操”的读者来说,必然会开卷有益。 “大众智慧、云计算、物联网、虚拟现实、网络社区、网络经济、协作双赢、电子货币……我们今天所知的,绝大多数是我们二十年前就已知的,并且都在这本书中提及了。”——凯文·凯利 《失控》成书于1994年,2010年中文版首次面......一起来看看 《失控》 这本书的介绍吧!

JSON 在线解析
JSON 在线解析

在线 JSON 格式化工具

随机密码生成器
随机密码生成器

多种字符组合密码

RGB CMYK 转换工具
RGB CMYK 转换工具

RGB CMYK 互转工具