Spring boot 集成Kafka

搭建Kafka集群,参考:

https://www.cnblogs.com/jonban/p/kafka.html

源码示例如下:

1、新建 Maven 项目 kafka

2、pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
        http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>
    <groupId>com.java</groupId>
    <artifactId>kafka</artifactId>
    <version>1.0.0</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.5.RELEASE</version>
    </parent>

    <dependencies>

        <!-- Spring Boot -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <!-- 热部署 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>springloaded</artifactId>
            <version>1.2.8.RELEASE</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>provided</scope>
        </dependency>

    </dependencies>

    <build>
        <finalName>${project.artifactId}</finalName>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

3、KafkaStarter.java

package com.java;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * 主启动类
 *
 * @author Logan
 * @version 1.0.0
 * @createDate 2019-05-07
 *
 */
@SpringBootApplication
public class KafkaStarter {

    public static void main(String[] args) {
        SpringApplication.run(KafkaStarter.class, args);
    }

}

4、MessageHandler.java

package com.java.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 *
 * 任务处理器,监听kafka队列中的消息,消费并处理
 *
 * @author Logan
 * @version 1.0.0
 * @createDate 2019-05-07
 *
 */
@Component
public class MessageHandler {

    @KafkaListener(topics = { "test-topic" })
    public void handle(String message) {
        System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis());

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(message);

        System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());
    }

    @KafkaListener(topics = { "test-topic" })
    public void handle(ConsumerRecord<String, String> record) {
        System.out.println("[ 处理器开始处理消息 ]" + System.currentTimeMillis());

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(record);

        System.out.println("[ 处理器处理消息完成 ]" + System.currentTimeMillis());
    }

}

5、SendMessageController.java

package com.java.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 发送消息类
 *
 * @author Logan
 * @version 1.0.0
 * @createDate 2019-05-07
 *
 */
@RestController
public class SendMessageController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private String topic = "test-topic";

    @GetMapping("/send")
    public String send(String params) {
        System.out.println("[ 收到请求 ]");

        kafkaTemplate.send(topic, params);

        System.out.println("[ 返回响应 ]");
        return "您的任务已提交";
    }

}

6、application.properties

#生产者配置,参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Producer
spring.kafka.producer.bootstrapServers=s1:9092,s2:9092,s3:9092

#消费者配置,参见org.springframework.boot.autoconfigure.kafka.KafkaProperties.Consumer
spring.kafka.consumer.bootstrapServers=s1:9092,s2:9092,s3:9092
spring.kafka.consumer.groupId=kafka-test
spring.kafka.consumer.autoOffsetReset=latest
spring.kafka.consumer.enableAutoCommit=true

7、运行KafkaStarter.java 启动

浏览器输入:http://127.0.0.1:8080/send?params=Good

可以向主题中发送消息Good,(params参数就是发送的内容)

程序中的消费者会监听到消息并开始处理

在Kafka消费者控制台可以监听到同样的消息

同样,在Kafka生产者控制台发送消息到主题 test-topic

程序中的消费者也会监听到消息并开始处理。

结论:测试生产者和消费者功能一切正常!

.

原文地址:https://www.cnblogs.com/jonban/p/spring-kafka.html

时间: 2024-10-08 18:54:02

Spring boot 集成Kafka的相关文章

spring boot 集成kafka (多线程,消费者使用kafka的原生api实现,因为@KakfkaListener修改groupId无效)

application-test.properties 1 #kafka 2 kafka.consumer.zookeeper.connect=*:2181 3 kafka.consumer.servers=*:9092 4 kafka.consumer.enable.auto.commit=true 5 kafka.consumer.session.timeout=6000 6 kafka.consumer.auto.commit.interval=1000 7 #保证每个组一个消费者消费同一

Kafka 入门和 Spring Boot 集成

Kafka 入门和 Spring Boot 集成 标签:博客 [TOC] 概述 kafka 是一个高性能的消息队列,也是一个分布式流处理平台(这里的流指的是数据流).由java 和 Scala 语言编写,最早由 LinkedIn 开发,并 2011年开源,现在由 Apache 开发维护. 应用场景 下面列举了一些kafka常见的应用场景. 消息队列 : Kafka 可以作为消息队列使用,可用于系统内异步解耦,流量削峰等场景. 应用监控:利用 Kafka 采集应用程序和服务器健康相关的指标,如应用

170711、spring boot 集成shiro

这篇文章我们来学习如何使用Spring Boot集成Apache Shiro.安全应该是互联网公司的一道生命线,几乎任何的公司都会涉及到这方面的需求.在Java领域一般有Spring Security.Apache Shiro等安全框架,但是由于Spring Security过于庞大和复杂,大多数公司会选择Apache Shiro来使用,这篇文章会先介绍一下Apache Shiro,在结合Spring Boot给出使用案例. Apache Shiro What is Apache Shiro?

Spring Boot集成Jasypt安全框架

Jasypt安全框架提供了Spring的集成,主要是实现 PlaceholderConfigurerSupport类或者其子类. 在Sring 3.1之后,则推荐使用PropertySourcesPlaceholderConfigurer类作为属性替换配置类,这里Spring集成Jasypt则使用Jasypt对属性替换配置类的实现.EncryptablePropertySourcesPlaceholderConfigurer. 在Spring中集成比较容易,而且Jasypt官方也给出了配置Bea

spring boot 集成 hbase

spring boot 集成 hbase 会启动报错 主要因为Spring Boot内嵌了Web容器,方便对应用进行微服务化开发和部署.所以打算将HBase的业务应用作为一个单服务进行开发和发布,其他相关的子系统通过RESTful API来访问. 搭建项目环境时,需要注意的事项: 由于Spring Boot内嵌了Web容器,所以框架默认导入了依赖:tomcat-embed-core-8.5.5.jar.tomcat-embed-el-8.5.5.jar等包.而HBase的jar中包含了:serv

spring boot集成swagger2

做java Web的后端开发已经两年多了,一般都是开发完了接口,都把接口更新到wiki文档上,然后通知前端去文档上去查阅接口的详细描述, 当时经常接口会有变动,加参数或返回值夹字段,所以维护语线上一致的文档是一件非常麻烦的事情,前一段时间同事聊天说他们公司用的swagger2,这个不需要写文档,它是自动生成文档,只要会使用它提供的几个的注解就行,于是上网找了下资料,发现它于spring boot集成也非常方便.不废话直接看了代码. 首先,在maven项目的pom.xml加上他需要的依赖. <de

Spring Boot 集成MyBatis

Spring Boot 集成MyBatis 在集成MyBatis前,我们先配置一个druid数据源. Spring Boot 系列 Spring Boot 入门 Spring Boot 属性配置和使用 Spring Boot 集成MyBatis Spring Boot 静态资源处理 Spring Boot - 配置排序依赖技巧 Spring Boot - DevTools 介绍 Spring Boot 集成druid druid有非常多个配置选项,使用Spring Boot 的配置文件能够方便的

Quartz与Spring Boot集成使用

上次自己搭建Quartz已经是几年前的事了,这次项目中需要定时任务,需要支持集群部署,想到比较轻量级的定时任务框架就是Quartz,于是来一波. 版本说明 通过搜索引擎很容易找到其官网,来到Document的页面,当前版本是2.2.x. 简单的搭建操作 通过Maven引入所需的包: <dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId>

spring boot 整合kafka 报错 Exception thrown when sending a message with key=&#39;null&#39; and payload=JSON to topic proccess_trading_end: TimeoutException: Failed to update metadata after 60000 ms.

org.springframework.kafka.support.LoggingProducerListener- Exception thrown when sending a message with key='null' and payload='{"dataDts":["20180329","20180328","20180327","20180326","20180323"]