Spring Kafka整合Spring Boot创建生产者客户端案例

每天学习一点点编程PDF电子书、视频教程免费下载：
http://www.shitanlife.com/code

创建一个kafka-producer-master的maven工程。整个项目结构如下：

Maven的依赖

<?xml version="1.0" encoding="UTF-8"?>

<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.gzh.kafka.producer</groupId>

    <artifactId>producer</artifactId>

    <version>0.0.1-SNAPSHOT</version>

    <packaging>jar</packaging>

    <name>kafka-producer-master</name>

    <description>demo project for kafka producer</description>

    <parent>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-parent</artifactId>

        <version>1.5.9.RELEASE</version>

        <relativePath/> <!-- lookup parent from repository -->

    </parent>

    <properties>

        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

        <spring-kafka.version>2.1.5.RELEASE</spring-kafka.version>

        <java.version>1.8</java.version>

    </properties>

    <dependencies>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter</artifactId>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->

        <dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka</artifactId>

            <version>${spring-kafka.version}</version>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-test</artifactId>

            <scope>test</scope>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test -->

        <dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka-test</artifactId>

            <version>${spring-kafka.version}</version>

            <scope>test</scope>

        </dependency>

        <!-- https://mvnrepository.com/artifact/io.springfox/springfox-swagger2 -->

        <dependency>

            <groupId>io.springfox</groupId>

            <artifactId>springfox-swagger2</artifactId>

            <version>2.8.0</version>

        </dependency>

        <!-- https://mvnrepository.com/artifact/io.springfox/springfox-swagger-ui -->

        <dependency>

            <groupId>io.springfox</groupId>

            <artifactId>springfox-swagger-ui</artifactId>

            <version>2.8.0</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-maven-plugin</artifactId>

            </plugin>

        </plugins>

    </build>

</project>

使用application.properties配置应用程序

当然，根据个人喜好，你也可以使用application.yml属性文件重写配置。Spring Boot会尝试根据pom.xml文件中指定的依赖关系自动配置应用程序，并设置合理的默认值。

server.port=8000

spring.application.name=kafka-producer

#kafka configuration

spring.kafka.producer.bootstrap-servers=192.168.1.130:9092,192.168.1.101:9093,192.168.1.101:9094

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#topic

kafka.app.topic.foo=test20180430

在上面的配置中，我给生产者分配的端口号是8000，服务器有3台，采用先前window环境搭建zookeeper,kafka集群中配置的服务器。想了解关于kafka生产者相关的更多配置的话，可以阅读关于Spring Boot Kafka Properties的配置信息。

使用Spring Boot发送Spring Kafka消息

SpringKafka提供了使用Producer的KafkaTemplate类发送消息，并提供将数据发送到Kafka主题的高级操作。提供异步和同步方法，异步方法返回Future。Spring Boot根据application.properties属性文件中配置的属性自动配置并初始化KafkaTemplate。为了方便测试发送消息，使用了Spring的定时任务，在类上使用@EnableScheduling 注解开启定时任务，通过@Scheduled注解指定发送消息规则。

package com.gzh.kafka.producer.component;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.kafka.support.SendResult;

import org.springframework.scheduling.annotation.EnableScheduling;

import org.springframework.scheduling.annotation.Scheduled;

import org.springframework.stereotype.Component;

import org.springframework.util.concurrent.ListenableFuture;

@Component

@EnableScheduling

public class KafkaMessageProducer {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaMessageProducer.class);

    @Autowired

    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${kafka.app.topic.foo}")

    private String topic;

    @Scheduled(cron = "00/5 * * * * ?")

    public void send() {

        String message = "Hello World---" + System.currentTimeMillis();

        LOG.info("topic="+topic+",message="+message);

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);

        future.addCallback(success -> LOG.info("KafkaMessageProducer 发送消息成功！"),

                fail -> LOG.error("KafkaMessageProducer 发送消息失败！"));

    }

}

创建消息生产者启动类

package com.gzh.kafka.producer;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.boot.context.properties.EnableConfigurationProperties;

@SpringBootApplication

@EnableConfigurationProperties

public class KafkaProducerApplication{

    public static void main(String[] args) {

        SpringApplication.run(KafkaProducerApplication.class, args);

    }

}

至此，Spring Boot整合Spring Kafka消息生产者应用已经整合完毕。启动zookeeper、kafka各个服务器。启动生产者应用，查看消息生产者应用控制台日志，如下图说明整合OK。

Spring Kafka整合Spring Boot创建生产者客户端案例

当然在创建消息生产者类时，我们可以更加灵活，可以不使用定时任务，通过界面请求的方式，发送我们想要发送的内容。简单案例如下：

消息发送者类

package com.gzh.kafka.producer.service;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.kafka.support.SendResult;

import org.springframework.stereotype.Service;

import org.springframework.util.concurrent.ListenableFuture;

@Service

public class KafkaMessageSendService {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaMessageSendService.class);

    @Autowired

    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${kafka.app.topic.foo}")

    private String topic;

    public void send(String message){

        LOG.info("topic="+topic+",message="+message);

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);

        future.addCallback(success -> LOG.info("KafkaMessageProducer 发送消息成功！"),

                fail -> LOG.error("KafkaMessageProducer 发送消息失败！"));

    }

}

界面请求处理controller类

package com.gzh.kafka.producer.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.http.MediaType;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

import com.gzh.kafka.producer.service.KafkaMessageSendService;

@RestController

@RequestMapping(value="send",produces=MediaType.APPLICATION_JSON_UTF8_VALUE)

public class KafkaMessageSendController {

    @Autowired

    private KafkaMessageSendService kafkaMessageSendService;

    @RequestMapping(value="/sendMessage",method=RequestMethod.POST)

    public String send(@RequestParam(required=true) String message){

        try {

            kafkaMessageSendService.send(message);

        } catch (Exception e) {

            return "send failed.";

        }

        return message;

    }

}

通过Swagger访问测试Controller服务请求

Spring Kafka整合Spring Boot创建生产者客户端案例

Spring Kafka整合Spring Boot创建消费者客户端案例：

创建一个kafka-consumer-master的maven工程。整个项目结构如下：

Spring Kafka整合Spring Boot创建生产者客户端案例

Maven的依赖

<?xml version="1.0" encoding="UTF-8"?>

<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.gzh.kafka.consumer</groupId>

    <artifactId>consumer</artifactId>

    <version>0.0.1-SNAPSHOT</version>

    <packaging>jar</packaging>

    <name>kafka-consumer-master</name>

    <description>demo project for kafka consumer</description>

    <parent>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-parent</artifactId>

        <version>1.5.9.RELEASE</version>

        <relativePath/> <!-- lookup parent from repository -->

    </parent>

    <properties>

        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

        <spring-kafka.version>1.3.4.RELEASE</spring-kafka.version>

        <java.version>1.8</java.version>

    </properties>

    <dependencies>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter</artifactId>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->

        <dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka</artifactId>

            <version>${spring-kafka.version}</version>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-test</artifactId>

            <scope>test</scope>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test -->

        <dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka-test</artifactId>

            <version>${spring-kafka.version}</version>

            <scope>test</scope>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-maven-plugin</artifactId>

            </plugin>

        </plugins>

    </build>

</project>

注意，这是使用Spring-Kafka时一定要注意版本问题，否则会报各种奇葩错误。Spring官方网站上给出了SpringKafka和kafka-client版本（它的版本号要和kafka服务器的版本保持一致）的对应关系：

Spring Kafka整合Spring Boot创建生产者客户端案例

使用application.properties配置应用程序

Spring Boot会尝试根据pom.xml文件中指定的依赖关系自动配置应用程序，并设置合理的默认值。

server.port=8001

spring.application.name=kafka-consumer

#kafka configuration

#指定消息被消费之后自动提交偏移量，以便下次继续消费

spring.kafka.consumer.enable-auto-commit=true

#指定消息组

spring.kafka.consumer.group-id=guan

#指定kafka服务器地址

spring.kafka.consumer.bootstrap-servers=192.168.1.130:9092,192.168.1.101:9093,192.168.1.101:9094

#指定从最近地方开始消费(earliest)

spring.kafka.consumer.auto-offset-reset=latest

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

#topic

kafka.app.topic.foo=test20180430

使用Spring Boot消费Spring Kafka消息

通过使用@KafkaListener来注解一个方法Spring Kafka会自动创建一个消息监听器容器。使用该注解，并指定要消费的topic（也可以指定消费组以及分区号，支持正则表达式匹配），这样，消费者一旦启动，就会监听kafka服务器上的topic，实时进行消费消息。

package com.gzh.kafka.consumer.service;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.messaging.MessageHeaders;

import org.springframework.messaging.handler.annotation.Headers;

import org.springframework.messaging.handler.annotation.Payload;

import org.springframework.stereotype.Component;

@Component

public class KafkaMessageConsumer {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaMessageConsumer.class);

    @KafkaListener(topics={"${kafka.app.topic.foo}"})

    public void receive(@Payload String message, @Headers MessageHeaders headers){

        LOG.info("KafkaMessageConsumer 接收到消息："+message);

        headers.keySet().forEach(key->LOG.info("{}: {}",key,headers.get(key)));

    }

}

创建消息消费者启动类

package com.gzh.kafka.consumer;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.boot.context.properties.EnableConfigurationProperties;

@SpringBootApplication

@EnableConfigurationProperties

public class KafkaConsumerApplication {

    public static void main(String[] args) {

        SpringApplication.run(KafkaConsumerApplication.class, args);

    }

}

消费者应用已经完成，接下来让我们验证Spring Kafka消息发送和接收效果。先依次启动zookeeper、kafka服务器，然后在启动生产者(kafka-producer-master)应用，再启动消费者(kafka-consumer-master)应用，然后观察生产者和消费者启动类日志：

Spring Kafka整合Spring Boot创建生产者客户端案例

每天学习一点点编程PDF电子书、视频教程免费下载：
http://www.shitanlife.com/code

Spring Kafka整合Spring Boot创建生产者客户端案例的更多相关文章

Spring Kafka和Spring Boot整合实现消息发送与消费简单案例
本文主要分享下Spring Boot和Spring Kafka如何配置整合,实现发送和接收来自Spring Kafka的消息. 先前我已经分享了Kafka的基本介绍与集群环境搭建方法.关于Kafka的 ...
spring中整合memcached,以及创建memcache的put和get方法
spring中整合memcached,以及创建memcache的put和get方法: 1:在项目中导入memcache相关的jar包 2:memcache在spring.xml的配置: 代码: &lt ...
Spring Cloud实战 | 第九篇：Spring Cloud整合Spring Security OAuth2认证服务器统一认证自定义异常处理
本文完整代码下载点击一. 前言相信了解过我或者看过我之前的系列文章应该多少知道点我写这些文章包括创建有来商城youlai-mall 这个项目的目的,想给那些真的想提升自己或者迷茫的人(包括自己- ...
SpringMVC + Spring + MyBatis 整合 + Spring shrio + easyUI + 权限管理框架,带shrio session和shrio cache集群实现方案
工作之余先来写了一个不算规范的简单架子基于spring mvc + spring + mybatis + Spring shrio 基于redis的集群方案系统权限部分,分成多个机构,其中每个机构 ...
Spring、整合Spring+JDBC
首先引入Spring包和JDBC所使用到的包: 配置beans.xml步骤: 1.配置dataSource的属性 2.将DataSource交给DataSourceTransactionManager ...
一步一步深入spring(7)-- 整合spring和JDBC的环境
1.配置数据源 (1).添加支持数据源的jar包commons-dbcp.jar .commons-pool.jar 当然也要添加其他的spring用到的jar以及这里用到的数据库mysql的jar ...
kafka生产者客户端
kafka的生产者 1. 生产者客户端开发熟悉kafka的朋友都应该知道kafka客户端有新旧版本,老版本采用scala编写,新版本采用java编写.随着kafka版本的升级,旧版本客户端已经快 ...
spring security整合QQ登录
最近在了解第三方登录的内容,尝试对接了一下QQ登录,此次记录一下如何实现QQ登录的过程,在这个例子中是和spring secuirty整合的,不整合spring secuirty也是一样的. 需求: ...
spring boot+kafka整合
springboot版本是2.0.4 首先,在maven中引入spring-kafka的jar包 <dependency> <groupId>org.springframewo ...

随机推荐

windows下常查看端口占用方法总结
启动Tomcat时又出现端口占用错误.现在把对端口的处理命令和方法进行汇总. 1.查看所有连接的PID 开始--运行--cmd ,输入netstat -ano 找到端口号对应的PID后,从任务管理器 ...
The Layout Process on Mac OSX and iOS
First we will recap the steps it takes to bring views on screen with Auto Layout enabled. When you’r ...
自己实现的库函数2（memset,memcmp,memcpy,memmove）
memset,memcmp,memcpy,memmove是对内存进行管理的库函数,为了更好的理解和使用这几个函数,自己用C语言实现一下~ //内存设置函数void* my_memset(void* d ...
HUD 1251 难题统计
/* 这题倒是没啥难度字典树可搞但是吧空间是个问题开始写成这样 struct node { int next[27],sum[27]; bool over; }t[maxn]; 死活过不了开 ...
sqlserver内存释放
由于Sql Server对于系统内存的管理策略是有多少占多少,除非系统内存不够用了(大约到剩余内存为4M左右), Sql Server才会释放一点点内存.所以很多时候,我们会发现运行Sql Serv ...
中药饮片ERP案例
企业简介 [规模] 苏州市天灵中药饮片有限公司成立于2002年,为苏州首家中药饮片GMP认证企业.公司品牌“李良济”首创于1914年民国初,祖传三代,为中华老字号企业.目前,公司经营主要分为三大块:中 ...
Linux下查看alert日志文件的两种方法
--linux下查看alert日志文件的两种方法: --方法1: SQL> show parameter background_dump_dest; NAME TYPE VALUE ------ ...
Docker部署Vue 工程包
docker部署 Vue 工程包目录结构 [root@host ~]# tree front/ front/ ├── dist.conf ├── dist.zip ├── Dockerfile └─ ...
P1593 因子和
P1593 因子和新算法:#define ni 逆元先质因数分解,(1+p1^1+p1^2...p1^x)*(1+p2^1+p2^2...p2^x)然后套等比数列公式就可以了. #include&lt ...
sscanf 与 ssprintf 用法（转载--https&colon;//www&period;cnblogs&period;com/Anker/p/3351168&period;html)
sprintf函数 sprintf函数原型为 int sprintf(char *str, const char *format, ...).作用是格式化字符串,具体功能如下所示: (1)将数字变量转 ...