`
terry0501
  • 浏览: 314649 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

spring kafka 整合

 
阅读更多
1.引用POM:
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>1.1.3.RELEASE</version>
</dependency>

2,消费者配置-spring-kafka-consumer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="consumerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="192.168.10.240:9092" />
<entry key="group.id" value="test" />
<entry key="enable.auto.commit" value="false" />
<entry key="retries" value="10"/>
<entry key="auto.commit.interval.ms" value="1000" />
<entry key="session.timeout.ms" value="15000" />
<entry key="key.deserializer"
value="org.apache.kafka.common.serialization.IntegerDeserializer" />
<entry key="value.deserializer"
value="org.apache.kafka.common.serialization.StringDeserializer" />
</map>
</constructor-arg>
</bean>
<!-- 创建consumerFactory bean -->
<bean id="consumerFactory"
class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<ref bean="consumerProperties" />
</constructor-arg>
</bean>
<bean id="concurrentKafkaListenerContainerFactory"
class="org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory">
<property name="consumerFactory" ref="consumerFactory" />
<property name="concurrency" value="3" />
<!-- <property name="containerProperties.pollTimeout" value="3000"/> -->
</bean>

<!-- 消费者容器配置信息 -->
<bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
<!-- 重要!配置topic -->
<constructor-arg value="my-topic"/>
<!--<property name="ackOnError" value="false"/>-->
<property name="messageListener" ref="messageListernerConsumerService"/>
<property name="errorHandler" ref="messageListernerConsumerService" />
</bean>
<bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart">
<constructor-arg ref="consumerFactory"/>
<constructor-arg ref="containerProperties"/>
</bean>

<!--如果需要失败重试就需要配置-->

<!-- 实际执行消息消费的类 -->
<bean id="messageListernerConsumerService" class="com.shenma.paulfrank.kafka.KafkaConsumerListener">
<property name="taskExecutorUtil" ref="taskExecutorUtil" />
</bean>
<!-- 异步线程池 -->
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<!-- 核心线程数 -->
<property name="corePoolSize" value="20" />
<!-- 最大线程数 -->
<property name="maxPoolSize" value="100" />
<!-- 队列最大长度 >=mainExecutor.maxSize -->
<property name="queueCapacity" value="1000" />
<!-- 线程池维护线程所允许的空闲时间 -->
<property name="keepAliveSeconds" value="300" />
<!-- 线程池对拒绝任务(无线程可用)的处理策略 -->
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
</property>
</bean>
<bean id="taskExecutorUtil" class="com.shenma.paulfrank.kafka.TaskExecutorUtil">
<!-- <constructor-arg ref="taskExecutor" /> -->
<property name="taskExecutor" ref="taskExecutor" />
</bean>
</beans>


3,生产者配置 - spring-kafka-producer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd">

<!-- 定义producer的参数 -->
<bean id="producerProperties" class="java.util.HashMap">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="${com.shenma.mq.address}" />
<entry key="group.id" value="0" />
<entry key="retries" value="10" />
<entry key="batch.size" value="16384" />
<entry key="linger.ms" value="1" />
<entry key="buffer.memory" value="33554432" />
<entry key="key.serializer"
value="org.apache.kafka.common.serialization.IntegerSerializer" />
<entry key="value.serializer"
value="org.apache.kafka.common.serialization.StringSerializer" />
</map>
</constructor-arg>
</bean>

<!-- 创建kafkatemplate需要使用的producerfactory bean -->
<bean id="producerFactory"
class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
<constructor-arg>
<ref bean="producerProperties" />
</constructor-arg>
</bean>

<!-- 创建kafkatemplate bean,使用的时候,只需要注入这个bean,即可使用template的send消息方法 -->
<bean id="KafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
<constructor-arg ref="producerFactory" />
<constructor-arg name="autoFlush" value="true" />
<property name="defaultTopic" value="test1" />
</bean>
</beans>

4,发送案例:
     
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void send(){
kafkaTemplate.send("my-topic", "hello");
}

5,消费案例:
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MsgConsumer {
@KafkaListener(topics = { "my-topic" })
public void processMessage2(String content) {
System.out.println("test2:"+content);
}
}
地址:
开发环境:
kafka开发环境服务地址:192.168.10.240:9092
kafka监控:192.168.10.240:9000
sit环境
172.16.10.44:9092
生产环境:
10.10.212.27:9092,10.10.228.81:9092,10.10.234.137:9092,10.10.163.126:9092

http://www.cnblogs.com/luotianshuai/p/5206662.html  kafka安装
kafka版本:0.11.0
分享到:
评论

相关推荐

    spring-kafka 整合官方文档

    从这些内容来看,spring-kafka整合官方文档提供了全面的信息,涵盖了从基础的Kafka配置和消息处理,到复杂的事务管理和错误处理机制,再到Kafka Streams的支持。它旨在帮助开发者利用Spring的强大功能,简化Kafka在...

    22.Spring Cloud整合RabbitMQ或Kafka消息驱动

    在Spring Cloud框架中,整合RabbitMQ或Kafka作为消息驱动是常见的微服务间通信方式。这两种技术都是流行的消息中间件,用于实现异步处理、解耦和扩展性。下面将详细阐述它们在Spring Cloud中的应用。 首先,...

    spring-kafka整合.docx

    Spring for Apache Kafka整合概述 Spring for Apache Kafka是Apache Kafka的一个Spring整合实现,旨在提供一个简洁、灵活的消息队列解决方案。下面是Spring for Apache Kafka的概述和关键特性: 概述 Spring for ...

    spring+kafka中间件整合

    spring整合kafka集群,init.properties配置kafka集群信息(也可以单个kafka服务),kafka-consumer.xml配置消费者监听,kafka-producer.xml配置消息生产者。

    Spring 集成 Kafka的配置文件及代码讲解

    在本文中,我们将深入探讨如何将Spring框架与Apache Kafka集成,以便实现在Spring应用中发送和接收消息。Kafka是一个高吞吐量、分布式的发布/订阅消息系统,而Spring框架是Java开发中最广泛使用的应用框架之一。通过...

    spring-kafka

    Spring Kafka 是一个由 Spring 社区开发的库,它为 Apache Kafka 提供了轻量级的、基于 Spring 的整合框架。Kafka 是一个分布式流处理平台,常用于构建实时数据管道和流应用。Spring Kafka 的设计目标是简化与 Kafka...

    kafka环境搭建并和spring整合

    ### Kafka环境搭建与Spring整合详解 #### 一、Kafka基本概念 Kafka是一款开源的分布式消息系统,它能够提供高吞吐量的数据管道和存储服务。为了更好地理解和使用Kafka,我们首先需要了解以下几个核心概念: 1. **...

    【基于xml方式】spring-kafka.zip

    当我们谈论"基于xml方式的Spring整合Kafka"时,意味着我们将使用XML配置文件来设置Spring与Kafka的集成。 Spring框架允许开发者通过XML配置来声明式地配置bean,包括数据源、事务管理器、以及与Kafka相关的消费者和...

    springboot - 2.7.3版本 - (七)整合Kafka

    在本文中,我们将深入探讨如何在Spring Boot 2.7.3版本中集成Apache Kafka,以便在微服务架构中实现高效的数据流处理。首先,让我们理解Spring Boot和Kafka的基本概念,然后逐步介绍如何配置和使用它们。 **Spring ...

    spring-boot集成kafka

    Spring for Apache Kafka提供了与Spring Boot整合的便利,通过引入`spring-kafka`和`spring-boot-starter`依赖,我们可以轻松地在Spring Boot应用中使用Kafka。添加以下依赖: ```xml &lt;groupId&gt;org.spring...

    SpringBoot整合kafka,代码简洁,自动分配分区和指定分区消费(亲测可用)

    总结,SpringBoot整合Kafka可以简化消息中间件的使用,通过Spring提供的API和注解,我们可以轻松地实现消息的生产和消费,同时支持自定义分区策略。这使得在实际项目中处理实时数据流变得更加便捷。

    Spirng整合Kafka的Demo

    Spring整合Kafka是一个常见的任务,尤其对于构建大数据流处理或者实时消息传递的系统至关重要。Spring框架提供了Spring Kafka模块,使得开发者能够轻松地在Java应用中集成Apache Kafka。在这个"Spring整合Kafka的...

    springboot - 2.7.3版本 - (八)ELK整合Kafka

    在本文中,我们将深入探讨如何在Spring Boot 2.7.3版本的项目中整合ELK(Elasticsearch、Logstash、Kafka)堆栈,以便实现高效且可扩展的日志管理和分析。ELK组合提供了实时日志收集、处理和搜索的能力,而Kafka作为...

    spring整合kafka注解和xml两种方式

    在Spring框架中整合Apache Kafka,可以使用注解或者XML配置的方式。这两种方法都允许开发者方便地将消息生产者和消费者集成到Spring应用中。这里我们将详细介绍这两种方式。 ### 一、注解方式 #### 1. 添加依赖 ...

    Spring Boot 整合 Apache Kafka.zip

    《Spring Boot 整合 Apache Kafka》博文对应的源代码;博文地址:https://blog.csdn.net/shawearn1027/article/details/107067328

    springboot整合kafka实现生产者和消费者

    在本文中,我们将深入探讨如何使用Spring Boot整合Apache Kafka,以构建一个生产者和消费者的示例。Apache Kafka是一个分布式流处理平台,常被用于构建实时数据管道和流应用。Spring Boot简化了Java开发,提供了开箱...

    springboot整合kafka的发布/消费demo项目源码

    在本文中,我们将深入探讨如何将Spring Boot与Apache Kafka整合,以便实现发布/消费消息队列模式。Apache Kafka是一款高效、可扩展且分布式的消息中间件,而Spring Boot则是一个简化了Spring应用程序开发的框架。...

Global site tag (gtag.js) - Google Analytics