内容简介：在之前的工作中，微服务框架使用的是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
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网
猜你喜欢:
- 持续集成：数据库集成及快速构建
- ShareSDK集成及集成后遇到的一些问题【原创】
- 持续集成与持续部署宝典Part 3：创建集成环境
- 持续集成与持续部署宝典Part 2：创建持续集成流水线
- 禅道 12.3.stable 版本发布，全面集成八种单元测试框架，打通持续集成闭环
- 持续集成将死
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
