RocketMq 是一款开源的RocketMQ Binder 完成Spring CLoud应用消息的订阅和发布。
RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的，高可靠的消息发布与订阅服务。
在说明RocketMQ 的实例前，先看下 Spring Cloud Stream
这是官方对 SpringCloud Stream 的一段介绍：
Spring CLoud Stream 是一个用于构建基于微服务应用的框架。它基于springboot 来创建具有生产级别的单机Spring 应用。 并且使用Spring Integration 与Broker进行连接。
Spring Cloud Stream 提供了消息中间件配置的统一抽象对象，推出了publish-subscribe, consumer groups,partition这些统一的概念。
Spring Cloud Stream 内部有2个概念： Binder 和Binding.
Binder: 跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现。
比如：kafka的实现 kafkaMessageChannelBinder， RabbitMq的实现
RAbbitMessageChannelBinder 以及 RocketMQ的实现
RocketMQMessageChannelBinder。
Binding: 包括Input Binding和 Output Binding。
Binding 在消息中间件与应用程序提供的Provider 和Consumer 之间提供了一个桥梁，实现了开发者只需要使用应用程序的Provider 或Consumer 生产或者消费数据即可。屏蔽了开发者与底层消息中间的接触：
下图是： Spring Cloud Stream 的架构设计。
image.png
实例：
如何接入
在启动实例进行演示前，先了解下springcloud 应用如何接入RocketMQ Binder.
1，首先，修改 pom.xml 文件，引入RocketMQ Stream Starter.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.taotao</groupId>
<artifactId>rocketmq</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>rocketmq</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
<spring-cloud-alibaba.version>0.2.1.RELEASE</spring-cloud-alibaba.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<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>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2,配置Input 和 OutPut 的binding 信息并配合 @EnableBinding 注解使其生效
package com.taotao.rocketmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
@EnableBinding({ Source.class, Sink.class })
@SpringBootApplication
public class RocketmqApplication {
public static void main(String[] args) {
SpringApplication.run(RocketmqApplication.class, args);
}
}
配置yml
server:
port: 28081
spring:
application:
name: spring-cloud-alibaba-rocketmq
#配置rocketmq的nameserver
cloud:
stream:
rocketmq:
binder:
namesrv-addr: 127.0.0.1:9876
#定义name为output的binding(多个output一样的配置)
bindings:
output:
destination: test-topic
content-type: application/json
#定义name为intput的binding
input:
destination: test-topic
content-type: application/json
group: test-group
management:
endpoints:
web:
exposure:
include: '*'
- 消息发送及消息订阅
下载并启动 RocketMQ
在接入 RocketMQ Binder 之前，首先需要启动 RocketMQ 的 Name Server 和 Broker。
下载RocketMQ最新的二进制文件，并解压
启动 Name Server
sh bin/mqnamesrv
- 启动 Broker
sh bin/mqbroker -n localhost:9876
- 创建 Topic: test-topic
sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic
应用启动
消息处理
使用name 为output 对应的binding 发送消息到test-topic
创建RocketmqProducer
package com.taotao.rocketmq.service;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Service;
import java.rmi.RemoteException;
/**
* 创建provider
* @autor tom
* @date 2020/3/22 0022 8:07
*/
@Service
public class RocketmqProducer {
/**
* 使用 RocketMQ 原生的API进行消息发送
* @param message
* @throws MQClientException
* @throws RemoteException
* @throws RemotingException
* @throws InterruptedException
* @throws MQBrokerException
*/
public void send(String message) throws MQClientException, RemoteException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer =new DefaultMQProducer("test_producer");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
Message msg=new Message("test-topic","test-tag",message.getBytes());
producer.send(msg);
}
}
创建 创建consumer
package com.taotao.rocketmq.service;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
/**
* 创建consumer
* @autor tom
* @date 2020/3/22 0022 8:14
*/
@Service
public class ReceiveService {
/**
* 默认是input ，在sink类中指定，如果想要多个input，需要填写一个实现sink的类
*
*/
@StreamListener("input")
public void receiveInput1(String receiveMessage){
System.out.println("input receive:"+receiveMessage);
}
}
发送消息：
package com.taotao.rocketmq.controller;
import com.taotao.rocketmq.service.RocketmqProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.rmi.RemoteException;
/**
* @autor tom
* @date 2020/3/22 0022 9:58
*/
@RestController
public class TestController {
@Autowired
RocketmqProducer rocketmqProducer;
@GetMapping("/send")
public String send() throws InterruptedException, RemoteException, RemotingException, MQClientException, MQBrokerException {
rocketmqProducer.send("tset rocketmq message");
return "OK";
}
}
