Springboot最简单的实战介绍 整合kafka-生产者与消费者(消息推送与订阅获取)

简介: Springboot最简单的实战介绍 整合kafka-生产者与消费者(消息推送与订阅获取)

我们开始整合:


首先,先往pom.xml文件添加Kafka的依赖,


<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.0.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>


然后,接下来是配置文件(以下提供properties格式,yml格式 ,供大家随便自取),

当然注释也是要好好看看的,毕竟都一字字敲的。


application.properties


#============== kafka ===================
# 指定kafka 代理地址,可以多个
#spring.kafka.bootstrap-servers=123.xxx.x.xxx:19092,123.xxx.x.xxx:19093,123.xxx.x.xxx:19094
spring.kafka.bootstrap-servers=192.168.x.xxx:9092
#=============== producer生产者  =======================
spring.kafka.producer.retries=0
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=16384
# 缓存容量
spring.kafka.producer.buffer-memory=33554432
# 指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#=============== consumer消费者  =======================
# 指定默认消费者group id
spring.kafka.consumer.group-id=test-app
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100ms
# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#spring.kafka.consumer.bootstrap-servers=192.168.8.111:9092
#spring.kafka.consumer.zookeeper.connect=192.168.8.103:2181
#指定tomcat端口
server.port=8063 


application.yml:


spring:
  # KAFKA
  kafka:
    # ָkafka服务器地址,可以指定多个
    bootstrap-servers: 123.xxx.x.xxx:19092,123.xxx.x.xxx:19093,123.xxx.x.xxx:19094
    #=============== producer生产者配置 =======================
    producer:
      retries: 0
      # 每次批量发送消息的数量
      batch-size: 16384
      # 缓存容量
      buffer-memory: 33554432
      # ָ指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    #=============== consumer消费者配置  =======================
    consumer:
      #指定默认消费者的group id
      group-id: test-app
      #earliest
      #当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
      #latest
      #当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
      #none
      #topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
      auto-offset-reset: latest
      enable-auto-commit: true
      auto-commit-interval: 100ms
      #指定消费key和消息体的编解码方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer


好了,配置工作准备完毕。


我们先来搞Kafka的生产者,也就是负责推送消息的模块:

创建一个类, 叫KafkaSender(注解不能少,留意代码),


package com.kafkademo.producer;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
 * Hello!
 * Created By  JCccc on 2018/11/24
 * 11:25
 */
@Component
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    private final Logger logger = LoggerFactory.getLogger(KafkaSender.class);
    public void send(String topic, String taskid, String jsonStr) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, taskid, jsonStr);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            //推送成功
            public void onSuccess(SendResult<String, Object> result) {
                logger.info(topic + " 生产者 发送消息成功:" + result.toString());
            }
            @Override
            //推送失败
            public void onFailure(Throwable ex) {
                logger.info(topic + " 生产者 发送消息失败:" + ex.getMessage());
            }
        });
    }
}


以上就是kafka生产者了,到此刻,你已经可以开始往kafka服务器推送消息了

事不宜迟,我们立马试试:


创建个controller,搞个接口试试推送下消息,


@GetMapping("/sendMessageToKafka")
public  String sendMessageToKafka() {
    Map<String,String> messageMap=new HashMap();
    messageMap.put("message","我是一条消息");
    String taskid="123456";
    String jsonStr=JSONObject.toJSONString(messageMap);
//kakfa的推送消息方法有多种,可以采取带有任务key的,也可以采取不带有的(不带时默认为null)
    kafkaSender.send("testTopic",taskid,jsonStr);
    return "hi guy!";
}


用postman测一下(对了,这些推送的前提是你的kafka服务器是没问题的,能正常连接)


image.png看看控制台反应:


image.png


可以看到,我们的kafka生产者再推送消息成功后,成功进入了我们的回调函数onSuccess,也打印了日志。


没错,你已经掌握kafak生产者了,你已经掌握推送消息了。 那么接下来,我们继续搞下kafka的消费者。


我们创一个类,叫KafkaConsumer (同样,注意看代码,注解不能少) :


package com.kafkademo.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;
import javax.servlet.http.HttpSession;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
 * Hello!
 * Created By  JCccc on 2018/11/24
 * 13:13
 */
@Component
public class KafkaConsumer  {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
//下面的主题是一个数组,可以同时订阅多主题,只需按数组格式即可,也就是用“,”隔开
    @KafkaListener(topics = {"testTopic"})
    public void receive(ConsumerRecord<?, ?> record){
        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());
    }
}


好,到此,kafka的消费者就这么简单完成了。

 

那么,我们接下来验证下,生产者推送消息到主题“testTopic”,消费者订阅主题“testTopic”,把消息消费下来:


一样,用postman来模拟下第三方调用接口,


image.png


我们看看控制台,


没错,kafka生产者跟刚刚一样,成功把消息推送到了主题testTopic上去了,回调函数OnSuccess打印了相关日志;


而,我们的kafka消费者,也是很有效率,再检测到自己订阅的主题testTopic有消息,立马消费了下来。


image.png


好了,springboot整合kafka,生产者、消费者就是这么轻松简单结束了。


当然了,该篇案例的生产者和消费者都放在了一个demo去介绍了,实际上大家使用的是按照业务场景,数据量去选择是否需要分开生产者项目&消费者项目,哪些是同时有生产者和消费者的身份的,哪些是只有生产者身份的,哪些是只有消费者身份的。


相关文章
|
7天前
|
消息中间件 大数据 Kafka
【Kafka】Kafka 中生产者运行流程
【4月更文挑战第10天】【Kafka】Kafka 中生产者运行流程
|
7天前
|
消息中间件 Java Kafka
Springboot集成高低版本kafka
Springboot集成高低版本kafka
|
7天前
|
消息中间件 缓存 Java
✈️【Kafka技术专题】「开发实战篇」深入实战探索Kafka的生产者的开发实现及实战指南
✈️【Kafka技术专题】「开发实战篇」深入实战探索Kafka的生产者的开发实现及实战指南
48 0
|
7天前
|
消息中间件 Kafka 流计算
Flink的分区表订阅功能是通过Kafka的topic分区来实现的
Flink的分区表订阅功能是通过Kafka的topic分区来实现的【1月更文挑战第6天】【1月更文挑战第26篇】
100 1
|
7天前
|
消息中间件 缓存 Kafka
探究Kafka原理-5.Kafka设计原理和生产者原理解析(下)
探究Kafka原理-5.Kafka设计原理和生产者原理解析
46 0
|
7天前
|
消息中间件 存储 负载均衡
探究Kafka原理-5.Kafka设计原理和生产者原理解析(上)
探究Kafka原理-5.Kafka设计原理和生产者原理解析
68 0
|
7天前
|
消息中间件 分布式计算 Java
探究Kafka原理-3.生产者消费者API原理解析(上)
探究Kafka原理-3.生产者消费者API原理解析
42 0
|
7天前
|
消息中间件 存储 负载均衡
Kafka【付诸实践 01】生产者发送消息的过程描述及设计+创建生产者并发送消息(同步、异步)+自定义分区器+自定义序列化器+生产者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka生产者】
【2月更文挑战第21天】Kafka【付诸实践 01】生产者发送消息的过程描述及设计+创建生产者并发送消息(同步、异步)+自定义分区器+自定义序列化器+生产者其他属性说明(实例源码粘贴可用)【一篇学会使用Kafka生产者】
274 4
|
7天前
|
消息中间件 缓存 安全
Kafka 的生产者优秀架构设计
Kafka 的生产者优秀架构设计
31 0
|
7天前
|
消息中间件 SQL druid
最新版 springboot集成kafka
最新版 springboot集成kafka
37 0