本篇内容主要讲解“spring cloud stream和kafka的原理及作用是什么”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“spring cloud stream和kafka的原理及作用是什么”吧!
Spring Cloud Stream is a framework for building highly scalable event-driven microservices connected with shared messaging systems.
The framework provides a flexible programming model built on already established and familiar Spring idioms and best practices, including support for persistent pub/sub semantics, consumer groups, and stateful partitions.
野生翻译：spring cloud stream是打算统一消息中间件后宫的男人，他身手灵活，身后有靠山spring，会使十八般武器（消息订阅模式啦，消费者组，stateful partitions什么的），目前后宫有东宫娘娘kafka和西宫娘娘rabbitMQ。
八卦党：今天我们扒一扒spring cloud stream和kafka的关系，rabbitMQ就让她在冷宫里面呆着吧。
A streaming platform has three key capabilities:
Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.
Store streams of records in a fault-tolerant durable way.
Process streams of records as they occur.
野生翻译：老娘是个流处理平台，能干的活可多了：
能处理发布/订阅消息
用很稳的方式保存消息
一来就处理，真的很快
总结一句话，就是快、稳、准。
kafka的运行非常简单，从这里下载，然后先运行zookeeper。在最新的kafka的下载包里面也包含了一个zookeeper，可以直接用里面的。zookeeper启动后，需要在kafka的配置文件里面配置好zookeeper的ip和端口，配置文件是config/server.properties。
############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=localhost:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000
然后运行bin目录下的命令，启动kafka就可以啦
bin/kafka-server-start.sh -daemon config/server.properties
kafka虽然启动了，但我们需要了解她的话，还是需要一个总管来汇报情况，我这边用的就是kafka-manager，下载地址在这里。很可惜的是只有源代码的下载，没有可运行版本的，需要自行编译，这个编译速度还挺慢的，我这边提供一个编译好的版本给大家，点这里。
kafka-manager同样需要配置一下和kafka的关系，在conf/application.conf文件里面，不过配置的不是kafka自己，而是kafka挂载的zookeeper。
kafka-manager.zkhosts="localhost:2181"
然后启动bin/kafka-manager就可以了（windows环境下也有kafka-manager.bat可以运行）
这里有个坑，在windows下面运行的话，可能启动失败，提示输入行太长
这个是因为目录太长，把kafak-manager-2.0.0.2目录名缩短就可以正常运行了。
启动后通过Add Cluster把Cluster Zookeeper Host把zookeeper的地址端口填上，Kafka Version的版本一定要和正在使用的kafka版本对上，否则可能看不到kafka的内容。
然后我们就能看到kafka的broker，topic，consumers，partitions等信息了。
一切的起点，还在start.spring.io
这黑乎乎的界面是spring为了万圣节搞的事情。和我们相关的是右边这两个依赖，这两个依赖在pom.xml里面对应的是这些
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>不过只凭这些还不行，直接运行的话，会提示
Caused by: java.lang.IllegalStateException: Unknown binder configuration: kafka
还需要加上一个依赖包
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-kafka</artifactId> </dependency>
spring cloud stream项目框架搭好后，我们需要分两个部分，一个是发消息的部分，一个是收消息的地方。我们先看发消息的部分，首先是配置文件,application.yml
spring: cloud: stream: default-binder: kafka #默认的绑定器， kafka: #如果用的是rabbitMQ这里填 rabbit binder: brokers: #Kafka的消息中间件服务器地址 - localhost:9092 bindings: output: #通道名称 binder: kafka destination: test1 #消息发往的目的地，对应topic group: output-group-1 #对应kafka的group content-type: text/plain #消息的格式
注意这里的output，表示是发布消息的，和后面订阅消息是对应的。这个output的名字是消息通道名称，是可以自定义的，后面会讲到。
然后我们需要创建一个发布者
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
@EnableBinding(Source.class)
public class Producer {
private Source mySource;
public Producer(Source mySource) {
super();
this.mySource = mySource;
}
public Source getMysource() {
return mySource;
}
public void setMysource(Source mysource) {
mySource = mySource;
}
}@EnableBinding 按字面理解就知道是绑定通道的，绑定的通道名就是上面的output，Soure.class是spring 提供的，表示这是一个可绑定的发布通道，它的通道名称就是output,和application.yml里面的output对应
源码可以看的很清楚
package org.springframework.cloud.stream.messaging;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
/**
* Bindable interface with one output channel.
*
* @author Dave Syer
* @author Marius Bogoevici
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public interface Source {
/**
* Name of the output channel.
*/
String OUTPUT = "output";
/**
* @return output channel
*/
@Output(Source.OUTPUT)
MessageChannel output();
}如果我们需要定义我们自己的通道，可以自己写一个类，比如下面这种，通道名就改成了my-out
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
public interface MySource {
String INPUT = "my-in";
String OUTPUT = "my-out";
@Input(INPUT)
SubscribableChannel myInput();
@Output(OUTPUT)
MessageChannel myOutput();
}这样的话，application.yml就要改了
my-out: binder: kafka destination: mytest #消息发往的目的地，对应topic group: output-group-2 #对应kafka的group content-type: text/plain #消息的格式
Product.class的@EnableBinding也需要改，为了做对应，我另外写了一个MyProducer
import org.springframework.cloud.stream.annotation.EnableBinding;
@EnableBinding(MySource.class)
public class MyProducer {
private MySource mySource;
public MyProducer(MySource mySource) {
super();
this.mySource = mySource;
}
public MySource getMysource() {
return mySource;
}
public void setMysource(MySource mysource) {
mySource = mySource;
}
}这样，发布消息的部分就写好了，我们写个controller来发送消息
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import com.wphmoon.kscs.service.ChatMessage;
import com.wphmoon.kscs.service.MyProducer;
import com.wphmoon.kscs.service.Producer;
@RestController
public class MyController {
@Autowired
private Producer producer;
@Autowired
private MyProducer myProducer;
// get the String message via HTTP, publish it to broker using spring cloud stream
@RequestMapping(value = "/sendMessage/string", method = RequestMethod.POST)
public String publishMessageString(@RequestBody String payload) {
// send message to channel output
producer.getMysource().output().send(MessageBuilder.withPayload(payload).setHeader("type", "string").build());
return "success";
}
@RequestMapping(value = "/sendMyMessage/string", method = RequestMethod.POST)
public String publishMyMessageString(@RequestBody String payload) {
// send message to channel myoutput
myProducer.getMysource().myOutput().send(MessageBuilder.withPayload(payload).setHeader("type", "string").build());
return "success";
}
}很简单，直接调用producer发送一个字符串就行了，我使用postman来发起这个动作
消息发送出去了，我们怎么收消息呢？往下看。
同样的，我们用之前的spring cloud stream项目框架做收消息的部分，首先是application.yml文件
server: port: 8081 spring: cloud: stream: default-binder: kafka kafka: binder: brokers: - localhost:9092 bindings: input: binder: kafka destination: test1 content-type: text/plain group: input-group-1 my-in: binder: kafka destination: mytest content-type: text/plain group: input-group-2
重点关注的就是input和my-in ,这个和之前的output和my-out一一对应。
默认和Source类对应的是Sink，这个是官方提供的，代码如下
package org.springframework.cloud.stream.messaging;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
/**
* Bindable interface with one input channel.
*
* @author Dave Syer
* @author Marius Bogoevici
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public interface Sink {
/**
* Input channel name.
*/
String INPUT = "input";
/**
* @return input channel.
*/
@Input(Sink.INPUT)
SubscribableChannel input();
}调用它的类Consumer用来接收消息，代码如下
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.FormatStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.handler.annotation.Payload;
@EnableBinding(Sink.class)
public class Consumer {
private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
@StreamListener(target = Sink.INPUT)
public void consume(String message) {
logger.info("recieved a string message : " + message);
}
@StreamListener(target = Sink.INPUT, condition = "headers['type']=='chat'")
public void handle(@Payload ChatMessage message) {
final DateTimeFormatter df = DateTimeFormatter.ofLocalizedTime(FormatStyle.MEDIUM)
.withZone(ZoneId.systemDefault());
final String time = df.format(Instant.ofEpochMilli(message.getTime()));
logger.info("recieved a complex message : [{}]: {}", time, message.getContents());
}
}而我们自定义channel的类MySink和MyConsumer代码如下：
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
public interface MySink {
String INPUT = "my-in";
@Input(INPUT)
SubscribableChannel myInput();
}import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.FormatStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.handler.annotation.Payload;
@EnableBinding(MySink.class)
public class MyConsumer {
private static final Logger logger = LoggerFactory.getLogger(MyConsumer.class);
@StreamListener(target = MySink.INPUT)
public void consume(String message) {
logger.info("recieved a string message : " + message);
}
@StreamListener(target = MySink.INPUT, condition = "headers['type']=='chat'")
public void handle(@Payload ChatMessage message) {
final DateTimeFormatter df = DateTimeFormatter.ofLocalizedTime(FormatStyle.MEDIUM)
.withZone(ZoneId.systemDefault());
final String time = df.format(Instant.ofEpochMilli(message.getTime()));
logger.info("recieved a complex message : [{}]: {}", time, message.getContents());
}
}这样就OK了，当上面我们用postman发了消息后，这边就能直接在日志里面看到
2019-10-29 18:42:39.455 INFO 13556 --- [container-0-C-1] com.wphmoon.kscsclient.MyConsumer : recieved a string message : 你瞅啥 2019-10-29 18:43:17.017 INFO 13556 --- [container-0-C-1] com.wphmoon.kscsclient.Consumer : recieved a string message : 你瞅啥
我们在application.yml里面定义的destination，就是kafka的topic，在kafka-manager的topic list里面可以看到
而接收消息的consumer也可以看到
这就是spring cloud stream和kafka的帝后之恋，不过他们这种政治联姻哪有这么简单，里面复杂的部分我们后面再讲，敬请期待，起驾回宫（野生翻译:The Return of the King）
到此，相信大家对“spring cloud stream和kafka的原理及作用是什么”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
