文档章节

spring cloud stream kafka实例

go4it
 go4it
发布于 2017/04/10 13:23
字数 241
阅读 118
收藏 4

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
粉丝 53
博文 705
码字总数 506188
作品 0
深圳
Spring Cloud (十五)Stream 入门、主要概念与自定义消息发送与接收

前言 不写随笔的日子仿佛就是什么都没有产出一般……上节说到要学Spring Cloud Bus,这里发现按照官方文档的顺序反而会更好些,因为不必去后边的章节去为当前章节去打基础,所以我们先学习S...

东北小狐狸
07/31
0
0
Spring Cloud Stream应用与自定义RocketMQ Binder:编程模型

前言: 本文作者张天,节选自笔者与其合著的《Spring Cloud微服务架构进阶》,即将在八月出版问世。本文将其中Spring Cloud Stream应用与自定义Rocketmq Binder的内容抽取出来,主要介绍Spr...

aoho
06/23
0
0
SpringCloud实战9-Stream消息驱动

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

狂小白
05/29
0
0
【Spring Cloud Stream】异步任务

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

qq_26545305
05/20
0
0
Spring Cloud-Honghu Cloud分布式微服务云系统—技术点

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

itcloud
04/25
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

内存模型是怎么解决缓存一致性的?

在再有人问你Java内存模型是什么,就把这篇文章发给他。这篇文章中,我们介绍过关于Java内容模型的来龙去脉。 我们在文章中提到过,由于CPU和主存的处理速度上存在一定差别,为了匹配这种差距...

Java填坑之路
12分钟前
1
0
vue-cli 3.0 初体验

最近复习了下vue,突然发现vue-cli已经更新到3.0版本了,并且变化蛮大,看来要不停的学习,真是一入前端深似海。 安装步骤: 1、全局安装 npm install -g @vue/cli Vue CLI 的包名称由 vue-...

tianyawhl
14分钟前
0
0
Angular进阶之路

【初级】会写页面,能出东西。 给定环境和 rest API,不用第三方库,能在十分钟内完成一个 master/detail 结构的带路由的应用(可以不管美观)。 知识点:Angular CLI、组件、路由、HTTP 服务...

陆小七的主页
16分钟前
0
0
Redis缓存数据库安全加固指导(一)

背景 在众多开源缓存技术中,Redis无疑是目前功能最为强大,应用最多的缓存技术之一,参考2018年国外数据库技术权威网站DB-Engines关于key-value数据库流行度排名,Redis暂列第一位,但是原生...

中间件小哥
17分钟前
0
0
百万级数据mysql分区

1. 什么是表分区? 表分区,是指根据一定规则,将数据库中的一张表分解成多个更小的,容易管理的部分。从逻辑上看,只有一张表,但是底层却是由多个物理分区组成。 2. 表分区与分表的区别 分表...

罗文浩
19分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部