spring cloud stream kafka 处理消息

spring cloud stream kafka

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

自定义管道

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;@Component
public interface MyProcessor {String MESSAGE_INPUT = "log_input";String MESSAGE_OUTPUT = "log_output";String LOG_FORMAT_INPUT = "log_format_input";String LOG_FORMAT_OUTPUT = "log_format_output";@Input(MESSAGE_INPUT)SubscribableChannel logInput();@Output(MESSAGE_OUTPUT)MessageChannel logOutput();@Input(LOG_FORMAT_INPUT)SubscribableChannel logFormatInput();@Output(LOG_FORMAT_OUTPUT)MessageChannel logFormatOutput();}

发送，需要绑定@EnableBinding(MyProcessor.class)

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.support.MessageBuilder;@EnableBinding(MyProcessor.class)
public class SendService {@Autowiredprivate MyProcessor source;public void sendMsg(String msg) {source.logOutput().send(MessageBuilder.withPayload(msg).build());}public void sendFormatMsg(String msg) {source.logFormatOutput().send(MessageBuilder.withPayload(msg).build());}
}

接收，receive方法在接收到消息后，可以直接通过@SendTo 转发到其他的消息队列中

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;@EnableBinding(MyProcessor.class)
@Slf4j
public class ReceiveListener {@StreamListener(value = MyProcessor.MESSAGE_INPUT)@SendTo(value = MyProcessor.LOG_FORMAT_OUTPUT)public String receive(String payload) {log.info("接收到的信息:" + payload);String result="格式化的消息:"+payload;return result;}@StreamListener(value = MyProcessor.LOG_FORMAT_INPUT)public void receiveFormat(String payload) {log.info("接收到的信息:" + payload);}}

绑定了2个kafka，注意要指定默认的绑定器default-binder，不然启动报错

spring:cloud:stream:default-binder: kafka1bindings:#与@StreamListener注解中的value一致，是绑定的渠道名log_input:binder: kafka1#绑定的kafka topic名称为testdestination: topicOne#消费组#          group: cloud-test2-group1#          content-type: application/jsonlog_format_input:binder: kafka2#绑定的kafka topic名称为testdestination: topicTwo#消费组# group: cloud-test2-group1#          content-type: application/jsonlog_output:binder: kafka1destination: topicOne#          content-type: application/jsonlog_format_output:binder: kafka2destination: topicTwobinders:kafka1:type: kafkaenvironment:spring:cloud:stream:kafka:binder:brokers: localhost:9092zkNodes: localhost:2181kafka2:type: kafkaenvironment:spring:cloud:stream:kafka:binder:brokers: localhost:9093zkNodes: localhost:2182

spring cloud stream kafka 处理消息相关推荐

spring cloud stream kafka 动态写入不同的topic(Using dynamically bound destinations)
引入依赖: <dependency><groupId>org.springframework.boot</groupId><artifactId>spr ...
Spring Cloud Stream 使用延迟消息实现定时任务（RabbitMQ）
应用场景我们在使用一些开源调度系统(比如:elastic-job等)的时候,对于任务的执行时间通常都是有规律性的,可能是每隔半小时执行一次,或者每天凌晨一点执行一次.然而实际业务中还存在另外一种定时 ...
Spring Cloud Stream 与 Kafka 整合
重要注意事项: 1. 尽管 Spring Cloud Stream Binder 中存在 Kafka 的整合,然而Spring Kafka 和 Spring Cloud Stream Kafka 在处 ...
Kafka的Spring Cloud Stream
总览该示例项目演示了如何使用事件驱动的体系结构 , Spring Boot ,Spring Cloud Stream, Apache Kafka和Lombok构建实时流应用程序. 在本教程结束时,您 ...
一. spring cloud gateway集成 spring cloud stream binder kafka，实现“动态路由“刷新与加载之采坑记录
一.前言 Spring Cloud Stream是用于构建消息驱动的微服务应用程序的框架. 本文主要介绍如何集成 Spring Cloud Stream,以 Kafka发布订阅模式(topic),实现 ...
Spring Cloud Stream消息中间件通过RabbitMQ实现消息推送
一.Spring Cloud Stream是什么 Spring Cloud Stream 是一个构建消息驱动微服务的框架. Spring Cloud Stream解决了开发人员无感知的使用消息中间件的 ...
【本人秃顶程序员】使用Spring Cloud Stream和RabbitMQ实现事件驱动的微服务
←←←←←←←←←←←← 快!点关注让我们展示如何使用Spring Cloud Stream来设计事件驱动的微服务.首先,Spring Cloud Stream首先有什么好处?因为Spring AM ...
Spring Cloud Stream核心原理介绍
一.简介目前,市面上消息中间件产品种类繁多,譬如RabbitMq,RocektMq,Kafka,Azure EventHub, Amazon Kenesis.各种中间件的原理.机制差异很大,但归根 ...
Spring Cloud Stream中文指导手册
Spring Cloud Stream中文指导手册 source 文章目录 Spring Cloud Stream中文指导手册 @[toc] Spring Cloud Stream 核心 1.简介 2 ...

spring cloud stream kafka 处理消息

spring cloud stream kafka 处理消息相关推荐

最新文章

热门文章