文档章节

spring cloud stream kafka实例

go4it
 go4it
发布于 2017/04/10 13:23
字数 241
阅读 107
收藏 4
点赞 0
评论 0

maven

    <dependencyManagement>
		<dependencies>
			<dependency>
				<groupId>org.springframework.cloud</groupId>
				<artifactId>spring-cloud-dependencies</artifactId>
				<version>Camden.SR6</version>
				<type>pom</type>
				<scope>import</scope>
			</dependency>
		</dependencies>
	</dependencyManagement>

    <dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
	</dependency>

生产者配置

server:
  port: 8081
spring:
  application:
    name: output-demo
  cloud:
     instance-count: 1
     instance-index: 0
     stream:
        kafka:
          binder:
            brokers: localhost:9092
            zk-nodes: localhost:2182
            auto-add-partitions: true
            auto-create-topics: true
            min-partition-count: 1
        bindings:
          output:
            destination: event-demo
            content-type: text/plain
            producer:
              partitionCount: 1


  • java代码
@EnableBinding(Source.class)
public class SendService {

	@Autowired
	private Source source;

	public void sendMessage(String msg) {
		try {
			source.output().send(MessageBuilder.withPayload(msg).build());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

@RestController
public class ProducerController {
	
	@Autowired
	private SendService service;
	
	@RequestMapping(value = "/send/{msg}", method = RequestMethod.GET)
	public void send(@PathVariable("msg") String msg){
		service.sendMessage(msg);
	}
	
}

消费者

spring:
  application:
    name: input-demo
  cloud:
     instance-count: 1
     instance-index: 0
     stream:
        kafka:
          binder:
            brokers: localhost:9092
            zk-nodes: localhost:2182
            auto-add-partitions: true
            auto-create-topics: true
            min-partition-count: 1
        bindings:
          input:
            destination: event-demo
            group: s1
            consumer:
              autoCommitOffset: false
              concurrency: 1
              partitioned: false
  • java代码
@EnableBinding(Sink.class)
public class MsgSink {

	@StreamListener(Sink.INPUT)
	public void process(Message<?> message) {
		System.out.println(message.getPayload());
		Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
		if (acknowledgment != null) {
			System.out.println("Acknowledgment provided");
			acknowledgment.acknowledge();
		}
	}
}

运行

先运行生产者,再运行消费者

curl -i localhost:8081/send/hello1

doc

© 著作权归作者所有

共有 人打赏支持
go4it
粉丝 50
博文 643
码字总数 434875
作品 0
深圳
SpringCloud实战9-Stream消息驱动

官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。   应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互,通过我们配置来 binding ,而 Spring Clo...

狂小白 ⋅ 05/29 ⋅ 0

【Spring Cloud Stream】异步任务

一、前言 前两篇博客提高了用线程池和消息队列才实现异步任务。本篇博客谈一谈用SpringCloud Stream来实现异步任务。 Spring Cloud Stream是一个用来为微服务应用构建消息驱动能力的框架。它...

qq_26545305 ⋅ 05/20 ⋅ 0

Spring Cloud-Honghu Cloud分布式微服务云系统—技术点

鸿鹄Cloud是基于springcloud的,spring cloud本身提供的组件就很多,但我们需要按照企业的业务模式来定制企业所需要的通用架构,那我们现在需要考虑使用哪些技术呢? 下面我针对于spring cl...

itcloud ⋅ 04/25 ⋅ 0

介绍Spring Cloud Stream与RabbitMQ集成

Spring Cloud Stream是一个建立在Spring Boot和Spring Integration之上的框架,有助于创建事件驱动或消息驱动的微服务。在本文中,我们将通过一些简单的例子来介绍Spring Cloud Stream的概念...

RaiseHead ⋅ 05/13 ⋅ 0

Spring Cloud入门教程(十):消息总线(Bus)

Spring Cloud入门教程系列: Spring Cloud入门教程(一):服务治理(Eureka) Spring Cloud入门教程(二):客户端负载均衡(Ribbon) Spring Cloud入门教程(三):声明式服务调用(Feign) Spring Cl...

CD826 ⋅ 05/22 ⋅ 0

Spring Cloud Task 2.0.0.RC1 发布,添加新功能

Spring Cloud Task 2.0.0.RC1 发布,此版本包括对现有依赖项的升级以及一些不错的新功能。 这个版本的新功能: 与Spring Cloud Stream 2.0兼容,Spring Cloud Stream 2.0在几天前发布并提供了...

周其 ⋅ 04/17 ⋅ 0

spring的cloud之kafka配置

Kafka配置 在上面的例子中,由于Kafka、ZooKeeper均运行于本地,所以我们没有在测试程序中通过配置信息来指定Kafka和ZooKeeper的配置信息,就完成了本地消息总线的试验。但是我们实际应用中,...

wilesun ⋅ 2016/11/18 ⋅ 0

Spring Cloud--Honghu Cloud分布式微服务云系统—组件化

Spring Cloud集成项目有很多,下面我们列举一下和Spring Cloud相关的优秀项目,我们的企业架构中用到了很多的优秀项目,说白了,也是站在巨人的肩膀上去整合的。在学习Spring Cloud之前大家必...

itcloud ⋅ 04/26 ⋅ 0

SpringCloud实战8-Bus消息总线

好了现在我们接着上一篇的随笔,继续来讲。上一篇我们讲到,我们如果要去更新所有微服务的配置,在不重启的情况下去更新配置,只能依靠spring cloud config了,但是,是我们要一个服务一个服...

狂小白 ⋅ 05/23 ⋅ 0

Spring Cloud构建微服务架构Hystrix监控面板

在Spring Cloud中构建一个Hystrix Dashboard非常简单,只需要下面四步: 创建一个标准的Spring Boot工程,命名为:hystrix-dashboard。 编辑pom.xml,具体依赖内容如下: 为应用主类加上,启...

明理萝 ⋅ 06/11 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

分布式数据库中间件DDM的实现原理

随着数据量不断增大,传统的架构模式难以解决业务量不断增长所带来的问题,特别是在业务成线性、甚至指数级上升的情况。此时我们不得不通过水平扩展,把数据库放到不同服务器上来解决问题,也...

中间件小哥 ⋅ 10分钟前 ⋅ 0

字符编码那点事:快速理解ASCII、Unicode、GBK和UTF-8

原作者:阮一峰(ruanyifeng.com),现重新整理发布,感谢原作者的无私分享。 1、引言 今天中午,我突然想搞清楚 Unicode 和 UTF-8 之间的关系,就开始查资料。 这个问题比我想象的复杂,午饭...

JackJiang- ⋅ 17分钟前 ⋅ 0

Spring Cloud构建微服务架构:服务消费(基础)

使用LoadBalancerClient 在Spring Cloud Commons中提供了大量的与服务治理相关的抽象接口,包括DiscoveryClient、这里我们即将介绍的LoadBalancerClient等。对于这些接口的定义我们在上一篇介...

itcloud ⋅ 17分钟前 ⋅ 0

MaxCompute产品最新进展 -- 从马力到计算力

摘要:本文从马力作为功率衡量标准为切入点,介绍了大数据领域的计算力衡量标准TPCBB以及MaxCompute2.0在Big Bench上的卓越表现。同时详细地分享了取得优异成绩背后的产品在最新有哪些进展,...

猫耳m ⋅ 17分钟前 ⋅ 0

Linux系统

声明:本栏目所使用的素材都是凯哥学堂VIP学员所写,学员有权匿名,对文章有最终解释权;凯哥学堂旨在促进VIP学员互相学习的基础上公开笔记。 Linux系统: Unix:是C语言转做出来的,最早的网...

凯哥学堂 ⋅ 18分钟前 ⋅ 0

13.1 设置更改root密码 13.2 连接mysql 13.3 mysql常用命令

13.1 设置更改root密码 启动MySQL数据库 [root@linux-10 ~]# /etc/init.d/mysqld startStarting MySQL SUCCESS! 由于MySQL的相关命令的所在路径不在系统的环境变量中,因此需要将路径添...

影夜Linux ⋅ 21分钟前 ⋅ 0

jeesite shiro+redis实现cache和session共享

jeesite这个开源框架本身集成的有shiro+redis来实现cache和session共享,但是需要修改一下文件配置即可 首先找到spring-context-shiro.xml文件 找到bean id为sessionDAO,将其修改为如下 <!...

wangxujun59 ⋅ 21分钟前 ⋅ 0

基本JNI搭建

1、编写Java代码 首先我们需要编写自己的java代码 public class Hello { static{ System.loadLibrary("hello-jni"); } public native String sayHello();} 2、把...

国仔饼 ⋅ 23分钟前 ⋅ 0

MaxCompute产品最新进展 -- 从马力到计算力

摘要:本文从马力作为功率衡量标准为切入点,介绍了大数据领域的计算力衡量标准TPCBB以及MaxCompute2.0在Big Bench上的卓越表现。同时详细地分享了取得优异成绩背后的产品在最新有哪些进展,...

阿里云云栖社区 ⋅ 27分钟前 ⋅ 0

AppDelegate 设置Root相关

self.window = UIWindow.init(frame: UIScreen.main.bounds) self.window?.backgroundColor = UIColor.white self.window?.makeKeyAndVisible() self.window?.rootViewController = RootTabB......

west_zll ⋅ 35分钟前 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部