`
苦艾酒
  • 浏览: 13975 次
  • 性别: Icon_minigender_2
  • 来自: 广州
社区版块
存档分类
最新评论

spring+springmvc+kafka分布式消息中间件集成方案

 
阅读更多

Honghu的消息服务平台已经抛弃了之前的ActiveMQ,改用高吞吐量比较大的Kafka分布式消息中间件方案:
kafka消息平台使用spring+kafka的集成方案,详情如下:
1. 使用最高版本2.1.0.RELEASE集成jar包:spring-integration-kafka


2. Zookeeper、Kafka分布式集群使用init.properties配置化方案。

 

kafka.servers=127.0.0.1:9092    
kafka.topic=xxxooo  

 3. 使用消息生产者spring-context-producer配置化方案。

<?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="localhost:9092" />  
                <entry key="group.id" value="2" />  
                <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="test" />  
    </bean>  
</beans>  

 4. 使用消息消费者spring-context-producer配置化方案。

<?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">   
  
  
    <!-- 定义consumer的参数 -->  
    <bean id="consumerProperties" class="java.util.HashMap">  
        <constructor-arg>  
            <map>  
                <entry key="bootstrap.servers" value="localhost:9092" />  
                <entry key="group.id" value="0" />  
                <entry key="enable.auto.commit" value="true" />  
                <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="messageListernerConsumerService" class="com.sml.sz.kafka.KafKaConsumer" />  
  
    <!-- 消费者容器配置信息 -->  
    <bean id="containerProperties"  
        class="org.springframework.kafka.listener.config.ContainerProperties">  
        <constructor-arg value="test" />  
        <property name="messageListener" ref="messageListernerConsumerService" />  
    </bean>  
  
    <!-- 创建kafkatemplate bean,使用的时候,只需要注入这个bean,即可使用template的send消息方法 -->  
    <bean id="messageListenerContainer"  
        class="org.springframework.kafka.listener.KafkaMessageListenerContainer"  
        init-method="doStart">  
        <constructor-arg ref="consumerFactory" />  
        <constructor-arg ref="containerProperties" />  
    </bean>  
</beans>    

 5. 使用注解方式注入消息类型

@Autowired  
private KafkaTemplate<xxx, ooo> kafkaTemplate;

6. 重写MessageListener 的getMessage方法获取消息(业务实现)

7. RestFul服务方式测试消息服务
@CrossOrigin(origins = "*", maxAge = 3600, methods = { RequestMethod.GET, RequestMethod.POST, RequestMethod.DELETE,  
        RequestMethod.PUT })  
@RestController  
@RequestMapping(value = "/rest/kafka")  
public class KafKaProducer {  
      
    @RequestMapping(value = "/send", method = RequestMethod.GET)  
    public JSONObject save() {  
        System.out.println("+++++++++++++++++++++++++++++++");  
        kafkaTemplate.sendDefault("HongHu KAFKA分布式消息服务测试");    
        return null;  
    }  
      
    @Autowired    
    private KafkaTemplate<Integer, String> kafkaTemplate;  
      
    
  
}  
 
@RestController  
public class KafKaConsumer implements MessageListener<Integer, String> {  
  
    @Autowired  
    private LogService logService;  
    public void onMessage(ConsumerRecord<Integer, String> records) {  
        System.out.println("====================" + records);  
        Object o = records.value();  
        Log log = new Log();  
        log.setIsNewRecord(true);  
        log.setId(IdGen.uuid());  
        log.setTitle(String.valueOf(o));  
        logService.save(log);  
  
    }  
  
}  
 接受消息了------------------:ConsumerRecord(topic = xxxooo, partition = 0, offset = 2489, CreateTime = 1479647648299, checksum = 3372898135, serialized key size = -1, serialized value size = 40, key = null, value = HongHu KAFKA分布式消息服务测试)
 
到此结束!
 

核心架构:Spring Cloud、Spring Boot、Mybatis、Redis、Rabbit MQ、微服务、分布式、电子商务

核心思想:产品微服务、模块化、原子化、持续集成、分布式、集群部署

开发模式:代码生成工具、驱动式开发模式、提高开发效率

(企业架构源码可以加求球:三五三六二四七二五九)

分享到:
评论

相关推荐

    springMVC-kafka消息分发

    Apache Kafka是一种分布式流处理平台,常被用作实时数据管道和消息中间件。它具有高吞吐量、低延迟、持久化、可分区和可复制等特性,使得Kafka成为大规模数据处理和实时数据流的理想选择。Kafka以主题(Topic)和...

    Spring-Mybatis-Druid-Kafka示例集成Example

    在本项目"Spring-Mybatis-Druid-Kafka示例集成Example"中,我们探讨了如何将四个关键组件——SpringMVC、Mybatis、Druid和Kafka——整合到一个应用中,以实现高效的数据处理和通信。让我们逐一深入解析这些技术及其...

    SpringMaven:spirngmvc + maven + redis + kafka + mybtis

    标题 "SpringMaven:spirngmvc + maven + redis + kafka + mybtis" 提到的是一个基于Java的Web开发项目,它集成了多个核心技术组件。让我们深入了解一下这些技术及其在项目中的作用。 1. **SpringMVC**: Spring ...

    kafkaspringMVC

    将Kafka与Spring MVC结合,可以创建一个事件驱动的架构,其中Kafka作为后台的消息中间件,负责在不同的服务之间传输数据。Spring MVC应用可以作为生产者,发布消息到Kafka主题,也可以作为消费者,从主题中接收和...

    kafkaSpringMVC

    在现代大数据处理和实时流计算中,Apache Kafka作为一个分布式消息中间件,扮演着至关重要的角色。它提供了高吞吐量、低延迟的消息传递能力,使得实时数据处理成为可能。而Spring MVC是Spring框架的一部分,用于构建...

    kafka生产和消费示例

    Apache Kafka是一个高性能、分布式的消息中间件,它主要用于构建实时数据管道和流处理应用。Spring MVC则是一个用于构建Web应用程序的轻量级框架,它提供了模型-视图-控制器(MVC)架构模式的实现。 首先,我们需要...

    Web网站架构演变趋势.docx

    开发通常采用Java Web技术,如JSP/Servlet,或者使用成熟的框架组合,如Maven+Spring+Struts+Hibernate或Maven+Spring+SpringMVC+MyBatis。通过JDBC进行数据库操作。这种架构成本低,易于部署,但存在扩展性和容错性...

    SpringMVC精品资源--订单派单中心、springBoot、springMVC、Mybatis、Redis、K.zip

    我们可以预想这个压缩包中包含了一个完整的微服务架构的订单派送系统实例,包括Spring Boot启动器、SpringMVC负责处理HTTP请求、Mybatis用于数据库操作、Redis用于缓存和队列管理,而Kafka则可能作为消息中间件,...

    基于 springboot 开发的论坛源码.zip

    4. Kafka:作为一个分布式消息中间件,Kafka在论坛中可能用于异步处理,例如处理用户发表的帖子、评论,或者进行后台数据分析,提高系统吞吐量。 5. Elasticsearch:作为实时的分布式搜索和分析引擎,Elasticsearch...

    基于springboot网上图书商城.zip

    随着业务规模的扩大,可能需要引入分布式服务,如分布式缓存Redis用于减轻数据库压力,RabbitMQ或Kafka作为消息中间件实现异步处理,以及Docker和Kubernetes进行微服务化部署。 【持续集成与部署】 为了保证项目的...

    SpringMVC精品资源--涵盖大部分Java进阶需要掌握的知识,包括【微服务】【中间件】【缓存】【数据库优化】【.zip

    在Spring MVC中,Spring Cache抽象可以方便地集成各种缓存解决方案。 4. **数据库优化**: 数据库优化涉及到SQL查询优化、索引设计、事务管理、读写分离、分库分表等。理解如何编写高效的SQL语句,避免全表扫描,...

    mvctest.zip

    在SpringMVC+MyBatis的环境中,我们可以利用RabbitMQ、ActiveMQ或Kafka等MQ实现消息传递,将耗时的操作放到后台处理,从而改善用户体验。 当涉及到事务管理时,Spring提供了声明式事务的支持,使得开发者可以在不...

    2023最新Java面试题

    中间件如消息队列(RabbitMQ、Kafka)、分布式缓存(Memcached、Redis)、负载均衡(Nginx、Zuul)等,这些都是大型系统中的重要组件,面试时会考察你如何选择和使用这些中间件。 九、设计模式 设计模式是解决常见...

    大数据学习进阶实战大纲,学习路径

    同时,Kafka作为消息中间件,它的角色是处理大规模数据流,确保消息的可靠传输。了解这些工具将帮助你构建高可用和高并发的数据处理系统。 这个学习大纲涵盖了从基础的Java编程到复杂的大数据处理和分布式系统的...

    JavaEE求职简历-姓名-JAVA开发工程师.docx

    5. **消息中间件**:使用基于消息表的一致性策略,确保了消息传输的可靠性,这可能是RabbitMQ或Kafka等中间件的实现。 6. **缓存管理**:借助Redis实现高效的数据缓存,提高数据读取速度,同时使用Quartz定时服务...

    大数据开发高薪训练营【课程大纲】.pdf

    大数据开发高薪训练营课程旨在为学员提供系统的大数据开发知识,课程内容包括JavaSE核心技术、JavaWeb开发、数据可视化、Hadoop生态圈技术栈以及分布式缓存与消息中间件等知识点。接下来将详细解释各阶段的知识点。 ...

    Java架构师简历模板(P6、P7)

    他还使用MongoDB处理大规模终端Id信息更新,ElasticSearch+Kafka实现链路追踪和日志统计,TCC事务保障支付一致性,以及基于Redis的分布式锁方案。 这位Java架构师的综合能力和专业技能覆盖了系统设计、开发、优化、...

    Java架构师的工作职责.docx

    2. 分布式处理中间件:熟悉 Kafka、Zookeeper、Dubbo、RocketMQ 等分布式处理中间件的使用。 3. 高并发、高负载、高可用性系统:熟悉高并发、高负载、高可用性系统的设计开发及调优。 ### 大数据处理技术 1. 大...

Global site tag (gtag.js) - Google Analytics